mirror of
https://git.squeakyrat.moe/faggot-chan/
synced 2026-11-04 18:35:45 +01:00
80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
#!/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
|