1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
#!/usr/bin/env python3
from pathlib import Path
from datetime import datetime
def now_str() -> str:
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def nots(line: str) -> str:
if line.startswith("[") and "]" in line:
return line.split("]", 1)[1].lstrip()
return line
def tail_lines(path: Path, n: int) -> list[str]:
# grabs the last n lines without reading the whole file into memory first,
# since memory.txt only ever grows and reading it fully every message doesn't scale
if n <= 0:
return []
avg_line_bytes = 200
with path.open("rb") as f:
f.seek(0, 2)
file_size = f.tell()
if file_size == 0:
return []
block_size = n * avg_line_bytes
data = b""
lines = []
while True:
seek_pos = max(file_size - block_size, 0)
f.seek(seek_pos)
data = f.read(file_size - seek_pos)
lines = data.splitlines()
if len(lines) > n or seek_pos == 0:
break
block_size *= 2
return [l.decode("utf-8", errors="ignore").strip() for l in lines[-n:] if l.strip()]
CHUNK_LINES = 4 # consecutive messages per rag chunk
MIN_CHUNK_CHARS = 40 # windows shorter than this are basically "lol" "yeah", not worth embedding
SKIP_FRAGMENTS = ("sent sticker:", "sent media:", "sent a random dm to ", "saved media:")
def make_chunks(lines: list[str]) -> list[str]:
# rag chunks are windows of consecutive messages instead of single lines, one-liners embed badly.
# the step is 1 so when the tail of memory.txt slides, the old windows keep the exact same text and don't get re-embedded
kept = []
for l in lines:
c = l.strip()
if any(f in c for f in SKIP_FRAGMENTS):
continue
if "said:" in c and c.split("said:", 1)[1].strip() == "":
continue
kept.append(c)
if len(kept) <= CHUNK_LINES:
windows = ["\n".join(kept)] if kept else []
else:
windows = ["\n".join(kept[i:i + CHUNK_LINES]) for i in range(len(kept) - CHUNK_LINES + 1)]
return [w for w in windows if len(w) >= MIN_CHUNK_CHARS]
def pick_chunks(scored: list, top_n: int, threshold: float) -> list:
# scored is (chunk, similarity) sorted best first. drops anything under the threshold and skips windows
# that share a line with one already picked (they overlap, otherwise the top 5 would be the same messages)
picked = []
used = set()
for chunk, sim in scored:
if sim < threshold:
break
lines = set(chunk.split("\n"))
if lines & used:
continue
used |= lines
picked.append((chunk, sim))
if len(picked) >= top_n:
break
return picked
|
