402 lines
17 KiB
Python
402 lines
17 KiB
Python
"""Справка под сообщением Бобра и лог его реплик .
|
||
|
||
Gateway путей не знает; здесь - что искать по тексту сообщения и куда писать
|
||
сказанное. ``Recall.block`` отдаёт указатели, не выводы: карточка упомянутого
|
||
человека, его записки о нём (`мета/бобер/наблюдения/<имя> - наблюдения.md`),
|
||
дни дневника, где он встречался, и раз в день - подошедшие сроки из досок и
|
||
ближайшая неделя календарей.
|
||
``ReplyLog.write`` кладёт каждую реплику Бобра из мастера и веток в
|
||
`мета/бобер/реплики/YYYY-MM.md` - единственное место, где они grep-абельны:
|
||
транскрипт живёт в Postgres, а глубокие чаты и так лежат файлами в `💬 чаты/`.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
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 datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from beaver_gateway.conversations.envelope import RecallContext
|
||
from beaver_gateway.conversations.texts import UserSaid
|
||
|
||
from beaver_agent.hands.calendars import Calendars, Event, Upcoming
|
||
|
||
_log = logging.getLogger(__name__)
|
||
|
||
LABEL = "справка (указатели по сообщению, не выводы; открой, если относится к вопросу):"
|
||
VOWELS = "аеёиоуыэюя"
|
||
KINDS = {"master": "мастер", "branch": "ветка"}
|
||
CALENDAR = "🗓 календарь"
|
||
CALENDAR_LIMIT = 15
|
||
CALENDAR_ERROR_MAX = 90
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Person:
|
||
name: str
|
||
"""Имя файла карточки без `.md`; записки - `<имя> - наблюдения.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]
|
||
|
||
|
||
def notes_sections(path: Path) -> dict[str, list[str]]:
|
||
"""`## заголовок` → непустые строки под ним (до следующего `## `)."""
|
||
out: dict[str, list[str]] = {}
|
||
current: str | None = None
|
||
for raw in path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
||
line = raw.rstrip()
|
||
if line.startswith("## "):
|
||
current = line[3:].strip().lower()
|
||
out.setdefault(current, [])
|
||
continue
|
||
if current is not None and line.strip():
|
||
out[current].append(line.strip())
|
||
return out
|
||
|
||
|
||
def notes_head(path: Path, sections: tuple[str, ...], *, cap: int = 40) -> str | None:
|
||
"""Выбранные секции файла записок одним блоком - для сида (портрет Бобра)."""
|
||
if not path.exists():
|
||
return None
|
||
parsed = notes_sections(path)
|
||
lines: list[str] = []
|
||
for name in sections:
|
||
body = parsed.get(name, [])
|
||
if body:
|
||
lines.append(f"## {name}")
|
||
lines.extend(body)
|
||
if not lines:
|
||
return None
|
||
if len(lines) > cap:
|
||
lines = [*lines[:cap], f"… ещё {len(lines) - cap} строк в файле"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
@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
|
||
"""Сколько последних дней дневника просматривать на упоминания."""
|
||
mute: frozenset[str] = frozenset()
|
||
"""Агенты, которым справку не показывают: чужие люди, чужие уши."""
|
||
calendar: Calendars | None = None
|
||
"""Ical-фиды календарей; None - календари не настроены, секции нет вовсе."""
|
||
_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)
|
||
_calendar_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:
|
||
if ctx.agent in self.mute:
|
||
return 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)
|
||
lines += self.calendar_lines(ctx.now, today)
|
||
if not lines:
|
||
return None
|
||
return "\n".join([LABEL, *lines])
|
||
|
||
def person_line(self, person: Person, *, now_lines: int = 6) -> str:
|
||
"""Указатели по человеку и `## сейчас` из его записок - инлайн."""
|
||
parts = [f"👤 {person.name} - карточка `{person.path}`"]
|
||
notes = self.notes_dir / f"{person.name} - наблюдения.md"
|
||
rel = notes.relative_to(self.vault)
|
||
now: list[str] = []
|
||
if notes.exists():
|
||
sections = notes_sections(notes)
|
||
feed = sections.get("лента", [])
|
||
now = [ln for ln in sections.get("сейчас", []) if ln.startswith("- ")]
|
||
patterns = [
|
||
ln for ln in sections.get("паттерны", []) if ln.startswith("- ")
|
||
]
|
||
desc = f"записки `{rel}`: лента {len(feed)} стр."
|
||
if patterns:
|
||
desc += f", паттернов {len(patterns)}"
|
||
if not now and feed:
|
||
last = feed[-1][2:]
|
||
desc += ", последняя: " + (
|
||
last if len(last) <= 110 else last[:109] + "…"
|
||
)
|
||
parts.append(desc)
|
||
else:
|
||
parts.append(f"записок нет (`{rel}`)")
|
||
days = self.diary_hits(person)
|
||
if days:
|
||
parts.append("дневник: " + ", ".join(f"[[{d}]]" for d in days))
|
||
line = "; ".join(parts)
|
||
if now:
|
||
line += "; сейчас:\n" + "\n".join(f" {ln}" for ln in now[:now_lines])
|
||
return line
|
||
|
||
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]]
|
||
|
||
def calendar_lines(self, now: datetime, today: date) -> list[str]:
|
||
"""Ближайшая неделя календарей - раз в день, следом за сроками досок.
|
||
|
||
Секция печатается всегда, когда до неё дошло: пустое окно - строкой
|
||
«событий нет», упавший фид - строкой «ОШИБКА» и записью в лог. Молча
|
||
пропасть она не может, иначе отказ канала неотличим от спокойного
|
||
календаря - так однажды и пролежала незамеченной сессия в USOS.
|
||
"""
|
||
if self.calendar is None or self._calendar_day == today:
|
||
return []
|
||
self._calendar_day = today
|
||
try:
|
||
found = self.calendar.upcoming(now)
|
||
except Exception as exc: # noqa: BLE001
|
||
_log.exception("календарь: справка не собралась")
|
||
reason = " ".join(f"{type(exc).__name__}: {exc}".split())
|
||
return [f"{CALENDAR}: ОШИБКА {reason[:CALENDAR_ERROR_MAX]}"]
|
||
lines = [f"{CALENDAR}: ОШИБКА {reason}" for reason in found.errors]
|
||
if found.read:
|
||
lines += self.calendar_window(found, self.calendar.days)
|
||
return lines or [f"{CALENDAR}: ОШИБКА фиды не ответили"]
|
||
|
||
def calendar_window(self, found: Upcoming, days: int) -> list[str]:
|
||
"""Окно ответивших фидов: заголовок с временем данных и события строками."""
|
||
stamp = f"{found.at:%Y-%m-%d %H:%M}"
|
||
if not found.events:
|
||
return [f"{CALENDAR} ({days} дней): событий нет, данные на {stamp}"]
|
||
zone = ZoneInfo(self.tz)
|
||
shown = found.events[:CALENDAR_LIMIT]
|
||
lines = [
|
||
f"{CALENDAR} ({days} дней), событий {len(found.events)}, данные на {stamp}:"
|
||
]
|
||
lines += [f" - {event_line(event, zone)}" for event in shown]
|
||
if len(found.events) > len(shown):
|
||
lines.append(f" … и ещё {len(found.events) - len(shown)} в окне")
|
||
return lines
|
||
|
||
|
||
def event_line(event: Event, zone: ZoneInfo) -> str:
|
||
"""Дата, начало, название и место - одной строкой в зоне агента."""
|
||
start = event.start.astimezone(zone)
|
||
when = "весь день" if event.all_day else f"{start:%H:%M}"
|
||
place = f" · {event.place}" if event.place else ""
|
||
return f"{start:%m-%d} {when} {event.title}{place}"
|
||
|
||
|
||
@dataclass
|
||
class ReplyLog:
|
||
"""`мета/бобер/реплики/YYYY-MM.md`: строка на реплику Бобра из мастера и веток."""
|
||
|
||
dir: Path
|
||
tz: str = "Europe/Warsaw"
|
||
max_chars: int = 600
|
||
mute: frozenset[str] = frozenset()
|
||
"""Агенты, чьи разговоры в лог реплик не идут."""
|
||
|
||
def write(self, said: UserSaid) -> None:
|
||
if said.agent in self.mute:
|
||
return
|
||
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")
|