295 lines
12 KiB
Python
295 lines
12 KiB
Python
"""Справка под сообщением Бобра и лог его реплик (архитектура §3.3, инвариант 8).
|
||
|
||
Gateway путей не знает; здесь - что искать по тексту сообщения и куда писать
|
||
сказанное. ``Recall.block`` отдаёт указатели, не выводы: карточка упомянутого
|
||
человека, записки агента о нём (`мета/бобер/наблюдения/`), дни дневника, где
|
||
он встречался, и раз в день - подошедшие сроки из досок. ``ReplyLog.write``
|
||
кладёт каждую реплику Бобра из мастера и веток в `мета/бобер/реплики/YYYY-MM.md`
|
||
- единственное место, где они grep-абельны: транскрипт живёт в Postgres,
|
||
а глубокие чаты и так лежат файлами в `💬 чаты/`.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, timedelta
|
||
from typing import TYPE_CHECKING
|
||
from zoneinfo import ZoneInfo
|
||
|
||
if TYPE_CHECKING:
|
||
from pathlib import Path
|
||
|
||
from beaver_gateway.core.conversations import UserSaid
|
||
from beaver_gateway.core.envelope import RecallContext
|
||
|
||
LABEL = "справка (указатели по сообщению, не выводы; открой, если относится к вопросу):"
|
||
VOWELS = "аеёиоуыэюя"
|
||
KINDS = {"master": "мастер", "branch": "ветка"}
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Person:
|
||
name: str
|
||
"""Имя файла карточки без `.md` - им же называется файл записок."""
|
||
aliases: tuple[str, ...]
|
||
path: str
|
||
"""Путь карточки от корня vault."""
|
||
plain: bool
|
||
"""Искать по голому имени в тексте (ближний круг), а не только по `[[ссылке]]`."""
|
||
|
||
@property
|
||
def keys(self) -> tuple[str, ...]:
|
||
return (self.name, *self.aliases)
|
||
|
||
|
||
def frontmatter_aliases(text: str) -> tuple[str, ...]:
|
||
"""`aliases:` из фронтматтера карточки - список строк или пусто."""
|
||
if not text.startswith("---"):
|
||
return ()
|
||
end = text.find("\n---", 3)
|
||
if end < 0:
|
||
return ()
|
||
out: list[str] = []
|
||
inside = False
|
||
for line in text[3:end].splitlines():
|
||
if line.startswith("aliases:"):
|
||
inside = True
|
||
rest = line[len("aliases:") :].strip()
|
||
if rest and rest != "[]":
|
||
out += [a.strip().strip("\"'") for a in rest.strip("[]").split(",")]
|
||
continue
|
||
if inside:
|
||
stripped = line.strip()
|
||
if stripped.startswith("- "):
|
||
out.append(stripped[2:].strip().strip("\"'"))
|
||
continue
|
||
if stripped:
|
||
inside = False
|
||
return tuple(a for a in out if a)
|
||
|
||
|
||
class People:
|
||
"""Индекс карточек `👤 люди/`: имя, алиасы, путь; пересобирается раз в `ttl`."""
|
||
|
||
def __init__(
|
||
self,
|
||
root: Path,
|
||
vault: Path,
|
||
*,
|
||
plain_dirs: tuple[str, ...] = ("личное", "профессиональное"),
|
||
skip_dirs: tuple[str, ...] = ("архив", "рандомы", "группы"),
|
||
ttl: float = 600.0,
|
||
) -> None:
|
||
self._root = root
|
||
self._vault = vault
|
||
self._plain_dirs = plain_dirs
|
||
self._skip_dirs = skip_dirs
|
||
self._ttl = ttl
|
||
self._built = 0.0
|
||
self._people: list[Person] = []
|
||
|
||
def all(self) -> list[Person]:
|
||
if time.monotonic() - self._built > self._ttl:
|
||
self._people = self._scan()
|
||
self._built = time.monotonic()
|
||
return self._people
|
||
|
||
def _scan(self) -> list[Person]:
|
||
if not self._root.exists():
|
||
return []
|
||
out: list[Person] = []
|
||
for card in sorted(self._root.rglob("*.md")):
|
||
parts = card.relative_to(self._root).parts[:-1]
|
||
plain = bool(parts) and parts[0] in self._plain_dirs
|
||
plain = plain and not any(p in self._skip_dirs for p in parts)
|
||
try:
|
||
head = card.read_text(encoding="utf-8", errors="ignore")[:2000]
|
||
except OSError:
|
||
continue
|
||
out.append(
|
||
Person(
|
||
name=card.stem,
|
||
aliases=frontmatter_aliases(head),
|
||
path=str(card.relative_to(self._vault)),
|
||
plain=plain,
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _stem_pattern(key: str) -> re.Pattern[str]:
|
||
"""Имя с падежом: `Зина` → Зина/Зине/Зину/Зиной, `Прохор` → Прохора/Прохором."""
|
||
stem = key[:-1] if key[-1].lower() in VOWELS and len(key) > 3 else key
|
||
return re.compile(rf"(?<!\w){re.escape(stem)}\w{{0,2}}(?!\w)", re.IGNORECASE)
|
||
|
||
|
||
def mentions(text: str, people: list[Person], *, limit: int = 3) -> list[Person]:
|
||
"""Кто упомянут: `[[ссылкой]]` - любой, голым именем - только ближний круг."""
|
||
found: dict[str, int] = {}
|
||
by_key = {k.casefold(): p for p in people for k in p.keys}
|
||
for m in re.finditer(r"\[\[([^\]|#]+)", text):
|
||
target = m.group(1).strip().split("/")[-1].casefold()
|
||
target = target.removesuffix(".md")
|
||
person = by_key.get(target)
|
||
if person is not None:
|
||
found.setdefault(person.name, m.start())
|
||
for person in people:
|
||
if not person.plain or person.name in found:
|
||
continue
|
||
for key in person.keys:
|
||
if " " in key:
|
||
m = re.search(rf"(?<!\w){re.escape(key)}(?!\w)", text, re.IGNORECASE)
|
||
elif len(key) >= 4:
|
||
m = _stem_pattern(key).search(text)
|
||
else:
|
||
continue
|
||
if m:
|
||
found.setdefault(person.name, m.start())
|
||
break
|
||
order = sorted(found.items(), key=lambda kv: kv[1])[:limit]
|
||
by_name = {p.name: p for p in people}
|
||
return [by_name[name] for name, _ in order]
|
||
|
||
|
||
@dataclass
|
||
class Recall:
|
||
"""Справка по сообщению: указатели в vault, которые gateway не может знать."""
|
||
|
||
vault: Path
|
||
people_dir: Path
|
||
notes_dir: Path
|
||
diary_dir: Path
|
||
boards_dir: Path
|
||
tz: str = "Europe/Warsaw"
|
||
diary_scan: int = 400
|
||
"""Сколько последних дней дневника просматривать на упоминания."""
|
||
_people: People | None = field(default=None, init=False, repr=False)
|
||
_diary_cache: dict[str, tuple[float, list[str]]] = field(
|
||
default_factory=dict, init=False, repr=False
|
||
)
|
||
_deadlines_day: date | None = field(default=None, init=False, repr=False)
|
||
|
||
@property
|
||
def people(self) -> People:
|
||
if self._people is None:
|
||
self._people = People(self.people_dir, self.vault)
|
||
return self._people
|
||
|
||
def block(self, ctx: RecallContext) -> str | None:
|
||
today = ctx.now.astimezone(ZoneInfo(self.tz)).date()
|
||
lines = [self.person_line(p) for p in mentions(ctx.text, self.people.all())]
|
||
if ctx.kind == "master":
|
||
lines += self.deadlines(today)
|
||
if not lines:
|
||
return None
|
||
return "\n".join([LABEL, *lines])
|
||
|
||
def person_line(self, person: Person) -> str:
|
||
parts = [f"👤 {person.name} - карточка `{person.path}`"]
|
||
notes = self.notes_dir / f"{person.name}.md"
|
||
rel = notes.relative_to(self.vault)
|
||
if notes.exists():
|
||
entries = [
|
||
ln.strip()
|
||
for ln in notes.read_text(
|
||
encoding="utf-8", errors="ignore"
|
||
).splitlines()
|
||
if ln.strip().startswith("- ")
|
||
]
|
||
if entries:
|
||
last = entries[-1][2:]
|
||
last = last if len(last) <= 110 else last[:109] + "…"
|
||
parts.append(f"записки `{rel}`: {len(entries)} стр., последняя: {last}")
|
||
else:
|
||
parts.append(f"записки `{rel}`: пусто")
|
||
else:
|
||
parts.append(f"записок нет (`{rel}`)")
|
||
days = self.diary_hits(person)
|
||
if days:
|
||
parts.append("дневник: " + ", ".join(f"[[{d}]]" for d in days))
|
||
return "; ".join(parts)
|
||
|
||
def diary_hits(
|
||
self, person: Person, *, limit: int = 3, ttl: float = 600.0
|
||
) -> list[str]:
|
||
cached = self._diary_cache.get(person.name)
|
||
if cached and time.monotonic() - cached[0] < ttl:
|
||
return cached[1]
|
||
keys = [f"[[{k.casefold()}" for k in person.keys]
|
||
hits: list[str] = []
|
||
files = sorted(self.diary_dir.glob("????-??-??.md"), reverse=True)
|
||
for path in files[: self.diary_scan]:
|
||
try:
|
||
text = path.read_text(encoding="utf-8", errors="ignore").casefold()
|
||
except OSError:
|
||
continue
|
||
if any(k in text for k in keys):
|
||
hits.append(path.stem)
|
||
if len(hits) >= limit:
|
||
break
|
||
self._diary_cache[person.name] = (time.monotonic(), hits)
|
||
return hits
|
||
|
||
def deadlines(
|
||
self, today: date, *, horizon_days: int = 1, limit: int = 8
|
||
) -> list[str]:
|
||
"""Открытые задачи досок с 📅 не позже завтра - раз в день, первым конвертом."""
|
||
if self._deadlines_day == today:
|
||
return []
|
||
self._deadlines_day = today
|
||
found: list[tuple[date, str, str]] = []
|
||
for board in (
|
||
sorted(self.boards_dir.glob("*.md")) if self.boards_dir.exists() else []
|
||
):
|
||
live = board.read_text(encoding="utf-8", errors="ignore").split("\n***", 1)[
|
||
0
|
||
]
|
||
for m in re.finditer(
|
||
r"^\s*- \[ \] (.+?)\s*📅 (\d{4}-\d{2}-\d{2})", live, re.MULTILINE
|
||
):
|
||
try:
|
||
due = date.fromisoformat(m.group(2))
|
||
except ValueError:
|
||
continue
|
||
if due <= today + timedelta(days=horizon_days):
|
||
found.append((due, board.stem, m.group(1).strip()))
|
||
found.sort()
|
||
return [f"📆 {due:%m-%d} {board}: {task}" for due, board, task in found[:limit]]
|
||
|
||
|
||
@dataclass
|
||
class ReplyLog:
|
||
"""`мета/бобер/реплики/YYYY-MM.md`: строка на реплику Бобра из мастера и веток."""
|
||
|
||
dir: Path
|
||
tz: str = "Europe/Warsaw"
|
||
max_chars: int = 600
|
||
|
||
def write(self, said: UserSaid) -> None:
|
||
text = said.text
|
||
if text.startswith("[сид:"):
|
||
# спавн ветки с текстом: строка сида едет над самим сообщением
|
||
text = text.split("\n\n", 1)[1] if "\n\n" in text else ""
|
||
flat = " ⏎ ".join(ln.strip() for ln in text.strip().splitlines() if ln.strip())
|
||
if not flat:
|
||
return
|
||
if len(flat) > self.max_chars:
|
||
flat = flat[: self.max_chars - 1] + "…"
|
||
local = said.at.astimezone(ZoneInfo(self.tz))
|
||
where = KINDS.get(said.kind, said.kind)
|
||
if said.kind != "master" and said.title:
|
||
where = f"{where} «{said.title}»"
|
||
path = self.dir / f"{local:%Y-%m}.md"
|
||
self.dir.mkdir(parents=True, exist_ok=True)
|
||
if not path.exists():
|
||
path.write_text(
|
||
f"# реплики Бобра, {local:%Y-%m}\n\n"
|
||
"> пишет gateway: что Бобёр сказал в мастере и ветках, "
|
||
"строка на сообщение. Глубокие чаты лежат в `💬 чаты/` сами.\n\n",
|
||
encoding="utf-8",
|
||
)
|
||
with path.open("a", encoding="utf-8") as fh:
|
||
fh.write(f"- {local:%Y-%m-%d %H:%M} · {where} · {flat}\n")
|