"""The envelope (§3.3): a background block after the user's text. Assembled when the turn starts, never when the message is queued: the time and what changed in the vault since the last envelope (added lines for the ``full`` files, names and counts for the rest). Ceilings keep it a signal, not a document; the injects that ride along are bundled below it by the queue, with their own header. """ from __future__ import annotations from dataclasses import dataclass from datetime import UTC, datetime from typing import TYPE_CHECKING from zoneinfo import ZoneInfo if TYPE_CHECKING: from collections.abc import Sequence from beaver_gateway.core.watch import Change, VaultWatch __all__ = ["Envelope", "render"] HEADER = ( "[конверт - фоновый сигнал, не обращение; " "реагируй, только если относится к вопросу]" ) @dataclass(slots=True) class Envelope: watch: VaultWatch | None = None tz: str = "UTC" max_lines: int = 120 per_file: int = 30 names_only_within: float = 600.0 last_at: datetime | None = None def build(self, *, now: datetime | None = None) -> str: now = now or datetime.now(UTC) changes = self.watch.take() if self.watch is not None else [] names_only = ( self.last_at is not None and (now - self.last_at).total_seconds() < self.names_only_within ) text = render( now=now, tz=self.tz, changes=changes, since=self.last_at, names_only=names_only, max_lines=self.max_lines, per_file=self.per_file, ) self.last_at = now return text def render( *, now: datetime, tz: str, changes: Sequence[Change], since: datetime | None, names_only: bool, max_lines: int = 120, per_file: int = 30, ) -> str: zone = ZoneInfo(tz) stamp = now.astimezone(zone) lines = [HEADER, f"время: {stamp:%Y-%m-%d %H:%M} ({_zone_label(tz)})"] ordered = sorted(changes, key=lambda c: (not c.full, c.path)) since_label = ( f"с {since.astimezone(zone):%H:%M}" if since is not None else "со старта" # noqa: RUF001 ) if ordered: names = ", ".join(f"{c.path} (+{c.added_count})" for c in ordered) lines.append(f"vault, изменено {since_label}: {names}") if not names_only: _append_diffs(lines, ordered, max_lines=max_lines, per_file=per_file) return "\n".join(lines[:max_lines]) def _append_diffs( lines: list[str], changes: Sequence[Change], *, max_lines: int, per_file: int ) -> None: budget = max_lines - len(lines) - 1 for change in changes: if not change.full or not change.added: continue if budget < 3: lines.append("… (потолок конверта)") return shown = change.added[: min(per_file, budget - 2)] lines.append(f"--- {change.path}, только добавленное ---") lines.extend(f"+ {line}" for line in shown) budget -= 1 + len(shown) if len(change.added) > len(shown): lines.append(f"+ … ещё {len(change.added) - len(shown)}") budget -= 1 def _zone_label(tz: str) -> str: return tz.rsplit("/", 1)[-1].replace("_", " ")