From 3e62529a0ecef544774cb045091317e8632038ad Mon Sep 17 00:00:00 2001 From: h Date: Tue, 13 Oct 2026 01:23:21 +0200 Subject: [PATCH] chore(scripts): one-off split of the eternal memory files by day --- ruff.toml | 1 + scripts/split_logs.py | 283 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 284 insertions(+) create mode 100644 scripts/split_logs.py diff --git a/ruff.toml b/ruff.toml index b316d3b..410f3a8 100644 --- a/ruff.toml +++ b/ruff.toml @@ -26,6 +26,7 @@ unfixable = ["F401"] [lint.per-file-ignores] "config.py" = ["INP001"] +"scripts/split_logs.py" = ["INP001", "T201"] "tests/*" = ["ALL"] [lint.pydocstyle] diff --git a/scripts/split_logs.py b/scripts/split_logs.py new file mode 100644 index 0000000..6051ceb --- /dev/null +++ b/scripts/split_logs.py @@ -0,0 +1,283 @@ +"""Разовая нарезка вечных и месячных файлов `мета/бобер/` по дням. + + BEAVER_VAULT= PYTHONPATH=. uv run python scripts/split_logs.py + BEAVER_VAULT= PYTHONPATH=. uv run python scripts/split_logs.py --apply + +Без `--apply` только печатает, что будет. + +Реплики и журнал куратора режутся точно: дата стоит в каждой записи. Дневник +режется по заголовкам и датированным строкам: записи под заголовком едут в +`дневник/`, строки вида `- 05:4x 12.10 …` - в `журнал/`. Журнал досуга на две +трети без дат, поэтому уезжает одним архивом. Оригиналы копируются в +`мета/бобер/.архив/<сегодня>/` до первой записи. +""" + +from __future__ import annotations + +import re +import shutil +import sys +from collections import defaultdict +from dataclasses import dataclass, field +from datetime import date, timedelta +from typing import TYPE_CHECKING + +from beaver_agent.vault import ( + BEAVER, + CURATOR_JOURNAL, + CURATOR_RUNS, + JOURNAL, + LEISURE, + OWN_DIARY, + REPLIES, + VAULT, + today, +) + +if TYPE_CHECKING: + from pathlib import Path + +YEAR = 2026 +OLD_DIARY = BEAVER / "дневник.md" +LEISURE_ARCHIVE = JOURNAL / "архив досуга.md" + +REPLY = re.compile(r"^- (\d{4}-\d{2}-\d{2}) \d{2}:\d{2} · ") +RUN = re.compile(r"^## (\d{4}-\d{2}-\d{2})\b") +ISO_HEAD = re.compile(r"^#{2,3}\s+(\d{4})-(\d{2})-(\d{2})\b") +HEAD = re.compile(r"^#{2,3}\s") +HEAD_DAY = re.compile(r"\b(\d{2})\.(\d{2})\b") +HEAD_HOUR = re.compile(r"^#{2,3}\s+(\d{2}):[\dx]{2}") +LINE = re.compile(r"^- (\d{2}):[\dx]{2}(?:-\d{2}:[\dx]{2})? (\d{2})\.(\d{2})\b") +LINE_DAY_FIRST = re.compile(r"^- (\d{2})\.(\d{2}) (\d{2}):[\dx]{2}\b") + +CURATOR_STUB = ( + "\n# куратор\n\n" + "> Здесь только строка прошлого прогона. Записи прогонов - в " + "`куратор/YYYY-MM-DD.md`, файл на день, свежие сверху.\n" +) +DIARY_STUB = ( + "\n---\n\n" + "> С 13.10.2026 дневник лежит по дням: `дневник/YYYY-MM-DD.md` - то, что " + "я думаю; следы тиков, логи и разбор промахов - `журнал/YYYY-MM-DD.md`. " + "Сюда больше не дописывать.\n" +) + + +@dataclass +class Plan: + writes: dict[Path, str] = field(default_factory=dict) + removes: list[Path] = field(default_factory=list) + backups: list[Path] = field(default_factory=list) + notes: list[str] = field(default_factory=list) + + def note(self, text: str) -> None: + self.notes.append(text) + + +def _join(blocks: list[str], sep: str) -> str: + return sep.join(block.strip("\n") for block in blocks if block.strip()) + "\n" + + +def _merge(path: Path, old: str, *, old_first: bool) -> str: + """Старое рядом с тем, что уже успели положить в файл дня.""" + if not path.exists(): + return old + fresh = path.read_text(encoding="utf-8").rstrip("\n") + "\n" + return f"{old}\n{fresh}" if old_first else f"{fresh}\n{old}" + + +def split_replies(plan: Plan) -> None: + for month in sorted(REPLIES.glob("????-??.md")): + days: dict[str, list[str]] = defaultdict(list) + stray = 0 + for line in month.read_text(encoding="utf-8").splitlines(): + match = REPLY.match(line) + if match: + days[match.group(1)].append(line) + elif line.strip() and not line.startswith(("#", ">")): + stray += 1 + for day, lines in days.items(): + path = REPLIES / f"{day}.md" + known = ( + [ + line + for line in path.read_text(encoding="utf-8").splitlines() + if REPLY.match(line) + ] + if path.exists() + else [] + ) + seen = set(lines) + merged = sorted( + [*lines, *(line for line in known if line not in seen)], + key=lambda line: line[:18], + ) + plan.writes[path] = ( + f"# реплики Бобра, {day}\n\n" + "> пишет gateway: что Бобёр сказал в мастере и ветках, " + "строка на сообщение. Глубокие чаты лежат в `💬 чаты/` сами.\n\n" + + "\n".join(merged) + + "\n" + ) + plan.removes.append(month) + plan.backups.append(month) + plan.note( + f"{month.name}: {sum(map(len, days.values()))} реплик на {len(days)} дней" + + (f", {stray} строк без даты пропали" if stray else "") + ) + + +def split_curator(plan: Plan) -> None: + if not CURATOR_JOURNAL.exists(): + return + lines = CURATOR_JOURNAL.read_text(encoding="utf-8").splitlines() + days: dict[str, list[str]] = defaultdict(list) + runs: list[str] | None = None + for line in lines: + match = RUN.match(line) + if match: + runs = days[match.group(1)] + runs.append(line) + elif runs is not None: + runs[-1] += "\n" + line + if not days: + return + for day, entries in days.items(): + path = CURATOR_RUNS / f"{day}.md" + plan.writes[path] = _merge(path, _join(entries, "\n\n"), old_first=False) + plan.writes[CURATOR_JOURNAL] = lines[0] + "\n" + CURATOR_STUB + plan.backups.append(CURATOR_JOURNAL) + plan.note( + f"{CURATOR_JOURNAL.name}: {sum(map(len, days.values()))} прогонов " + f"на {len(days)} дней" + ) + + +def _day(month: str, day: str) -> date: + return date(YEAR, int(month), int(day)) + + +def split_diary(plan: Plan) -> None: + if not OLD_DIARY.exists(): + return + lines = OLD_DIARY.read_text(encoding="utf-8").splitlines() + first = next((i for i, line in enumerate(lines) if HEAD.match(line)), len(lines)) + preamble = "\n".join(lines[:first]).rstrip().removesuffix("---").rstrip() + thoughts: dict[date, list[str]] = defaultdict(list) + traces: dict[date, list[str]] = defaultdict(list) + current: date | None = None + hour = 0 + chunk: list[str] | None = None + for number, line in enumerate(lines[first:], first + 1): + iso, timed = ISO_HEAD.match(line), HEAD_HOUR.match(line) + log = LINE.match(line) + day_first = LINE_DAY_FIRST.match(line) + stamp: date | None = None + target = thoughts + if iso: + stamp = date(int(iso.group(1)), int(iso.group(2)), int(iso.group(3))) + elif HEAD.match(line): + named = HEAD_DAY.search(line) + if named: + stamp = _day(named.group(2), named.group(1)) + elif timed and current is not None: + rolled = int(timed.group(1)) < hour + stamp = current + timedelta(days=1) if rolled else current + if timed: + hour = int(timed.group(1)) + elif log: + stamp, target = _day(log.group(3), log.group(2)), traces + elif day_first: + stamp, target = _day(day_first.group(2), day_first.group(1)), traces + if stamp is None: + if chunk is not None: + chunk[-1] += "\n" + line + continue + if current is not None and stamp < current: + plan.note(f"дневник, строка {number}: дата идёт назад, {current} → {stamp}") + current = stamp + chunk = target[stamp] + chunk.append(line) + if not thoughts and not traces: + return + for day, blocks in thoughts.items(): + path = OWN_DIARY / f"{day}.md" + plan.writes[path] = _merge(path, _join(blocks, "\n\n"), old_first=True) + for day, blocks in traces.items(): + path = JOURNAL / f"{day}.md" + plan.writes[path] = _merge(path, _join(blocks, "\n"), old_first=True) + plan.writes[OLD_DIARY] = preamble + "\n" + DIARY_STUB + plan.backups.append(OLD_DIARY) + plan.note( + f"{OLD_DIARY.name}: {sum(map(len, thoughts.values()))} записей на " + f"{len(thoughts)} дней в дневник, {sum(map(len, traces.values()))} строк на " + f"{len(traces)} дней в журнал" + ) + + +def split_leisure(plan: Plan) -> None: + if not LEISURE.exists(): + return + text = LEISURE.read_text(encoding="utf-8") + head, mark, journal = text.partition("\n## журнал\n") + if not mark or not journal.strip(): + return + plan.writes[LEISURE_ARCHIVE] = ( + "# журнал досуга до 13.10.2026\n\n" + "> Архив из `досуг.md`: записи шли без дат и в разном порядке, по дням " + "их не разложить. Дальше следы тиков лежат в `журнал/YYYY-MM-DD.md`.\n\n" + + journal.strip("\n") + + "\n" + ) + plan.writes[LEISURE] = head.rstrip("\n") + "\n" + plan.backups.append(LEISURE) + plan.note( + f"{LEISURE.name}: {len(journal) // 1024} КБ журнала в архив, " + f"в файле остаётся {len(head) // 1024 + 1} КБ раскладки" + ) + + +def backup(sources: list[Path]) -> Path: + target = BEAVER / ".архив" / today() + target.mkdir(parents=True, exist_ok=True) + for path in sources: + if path.exists() and not (target / path.name).exists(): + shutil.copy2(path, target / path.name) + return target + + +def main() -> None: + plan = Plan() + split_replies(plan) + split_curator(plan) + split_diary(plan) + split_leisure(plan) + for note in plan.notes: + print(note) + folders: dict[str, int] = defaultdict(int) + for path in plan.writes: + folders[str(path.parent.relative_to(VAULT))] += 1 + for folder, count in sorted(folders.items()): + print(f"{folder}: {count} файлов") + largest = max(plan.writes.items(), key=lambda item: len(item[1]), default=None) + if largest: + print( + f"самый большой: {largest[0].relative_to(VAULT)}, " + f"{len(largest[1].encode()) // 1024} КБ" + ) + if "--apply" not in sys.argv: + print("ничего не записано: это прогон без --apply") + return + if not plan.writes: + return + print(f"оригиналы: {backup(plan.backups).relative_to(VAULT)}") + for path, text in plan.writes.items(): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + for path in plan.removes: + path.unlink() + print(f"записано {len(plan.writes)} файлов, убрано {len(plan.removes)}") + + +if __name__ == "__main__": + main()