chore(scripts): one-off split of the eternal memory files by day
This commit is contained in:
hh committed
1 parent
fbfe39a968
commit
3e62529a0e
2 files changed
+284
No files matched your search
@@ -26,6 +26,7 @@ unfixable = ["F401"]
|
||||
|
||||
[lint.per-file-ignores]
|
||||
"config.py" = ["INP001"]
|
||||
"scripts/split_logs.py" = ["INP001", "T201"]
|
||||
"tests/*" = ["ALL"]
|
||||
|
||||
[lint.pydocstyle]
|
||||
|
||||
@@ -0,0 +1,283 @@
|
||||
"""Разовая нарезка вечных и месячных файлов `мета/бобер/` по дням.
|
||||
|
||||
BEAVER_VAULT=<vault> PYTHONPATH=. uv run python scripts/split_logs.py
|
||||
BEAVER_VAULT=<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()
|
||||
Reference in new issue
Block a user