155 lines
4.6 KiB
Python
155 lines
4.6 KiB
Python
"""The envelope: a background block under the user's text - clock, changes, recall."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from beaver_gateway.conversations.texts import EnvelopeTexts
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable, Sequence
|
|
|
|
from beaver_gateway.vault.watch import Change, VaultWatch
|
|
|
|
__all__ = ["Envelope", "RecallContext", "render"]
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RecallContext:
|
|
text: str
|
|
kind: str
|
|
now: datetime
|
|
agent: str = ""
|
|
"""Agent whose turn it is; a setup with several tells them apart by it."""
|
|
|
|
|
|
@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
|
|
recall: Callable[[RecallContext], str | None] | None = None
|
|
"""Setup-side lookup on the user's text; its lines go under the change block."""
|
|
texts: EnvelopeTexts = field(default_factory=EnvelopeTexts)
|
|
|
|
def build(
|
|
self,
|
|
*,
|
|
now: datetime | None = None,
|
|
text: str = "",
|
|
kind: str = "master",
|
|
agent: str = "",
|
|
) -> 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
|
|
)
|
|
out = 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,
|
|
texts=self.texts,
|
|
)
|
|
self.last_at = now
|
|
block = self.recall_block(text=text, kind=kind, now=now, agent=agent)
|
|
return f"{out}\n{block}" if block else out
|
|
|
|
def recall_only(
|
|
self, *, text: str, kind: str, now: datetime | None = None, agent: str = ""
|
|
) -> str | None:
|
|
block = self.recall_block(
|
|
text=text, kind=kind, now=now or datetime.now(UTC), agent=agent
|
|
)
|
|
return f"{self.texts.header}\n{block}" if block else None
|
|
|
|
def recall_block(
|
|
self, *, text: str, kind: str, now: datetime, agent: str = ""
|
|
) -> str | None:
|
|
if self.recall is None or not text.strip():
|
|
return None
|
|
try:
|
|
block = self.recall(
|
|
RecallContext(text=text, kind=kind, now=now, agent=agent)
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
_log.exception("recall hook failed")
|
|
return None
|
|
return block.strip() or None if block else None
|
|
|
|
|
|
def render(
|
|
*,
|
|
now: datetime,
|
|
tz: str,
|
|
changes: Sequence[Change],
|
|
since: datetime | None,
|
|
names_only: bool,
|
|
max_lines: int = 120,
|
|
per_file: int = 30,
|
|
texts: EnvelopeTexts | None = None,
|
|
) -> str:
|
|
texts = texts or EnvelopeTexts()
|
|
zone = ZoneInfo(tz)
|
|
stamp = now.astimezone(zone)
|
|
lines = [
|
|
texts.header,
|
|
texts.time.format(stamp=f"{stamp:%Y-%m-%d %H:%M}", zone=_zone_label(tz)),
|
|
]
|
|
ordered = sorted(changes, key=lambda c: (not c.full, c.path))
|
|
since_label = (
|
|
texts.since.format(time=f"{since.astimezone(zone):%H:%M}")
|
|
if since is not None
|
|
else texts.since_start
|
|
)
|
|
if ordered:
|
|
names = ", ".join(f"{c.path} (+{c.added_count})" for c in ordered)
|
|
lines.append(texts.changed.format(since=since_label, names=names))
|
|
if not names_only:
|
|
_append_diffs(
|
|
lines, ordered, max_lines=max_lines, per_file=per_file, texts=texts
|
|
)
|
|
return "\n".join(lines[:max_lines])
|
|
|
|
|
|
def _append_diffs(
|
|
lines: list[str],
|
|
changes: Sequence[Change],
|
|
*,
|
|
max_lines: int,
|
|
per_file: int,
|
|
texts: EnvelopeTexts,
|
|
) -> 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(texts.truncated)
|
|
return
|
|
shown = change.added[: min(per_file, budget - 2)]
|
|
lines.append(texts.file_header.format(path=change.path))
|
|
lines.extend(f"+ {line}" for line in shown)
|
|
budget -= 1 + len(shown)
|
|
if len(change.added) > len(shown):
|
|
lines.append(texts.more_lines.format(count=len(change.added) - len(shown)))
|
|
budget -= 1
|
|
|
|
|
|
def _zone_label(tz: str) -> str:
|
|
return tz.rsplit("/", 1)[-1].replace("_", " ")
|