284 lines
11 KiB
Python
284 lines
11 KiB
Python
"""Разовая нарезка вечных и месячных файлов `мета/бобер/` по дням.
|
||
|
||
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()
|