import json import logging import time from pathlib import Path logger = logging.getLogger(__name__) DAY_SECONDS = 86400 class EventLog: """Sent alerts and stream notes. Kept on the volume; the UI asks for a day.""" def __init__(self, path: Path | None = None): self.path = Path(path) if path else None self._entries = self._load() def _load(self) -> list[dict]: if not self.path or not self.path.exists(): return [] try: payload = json.loads(self.path.read_text(encoding="utf-8")) return payload if isinstance(payload, list) else [] except Exception: logger.warning("Could not read event log; starting empty", exc_info=True) return [] def _save(self) -> None: if not self.path: return try: self.path.parent.mkdir(parents=True, exist_ok=True) temporary = self.path.with_suffix(self.path.suffix + ".tmp") temporary.write_text( json.dumps(self._entries, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) temporary.replace(self.path) except Exception: logger.warning("Could not persist event log", exc_info=True) def add(self, kind: str, message: str, *, number: int | None = None, at: int | None = None) -> dict: entry = { "kind": kind, "message": message, "number": number, "at": int(at if at is not None else time.time()), } self._entries.append(entry) self._save() return entry def page( self, *, before: int | None = None, since: int | None = None, limit: int = 100, ) -> tuple[list[dict], bool]: items = self._entries if before is not None: items = [entry for entry in items if int(entry.get("at") or 0) < before] if since is not None: items = [entry for entry in items if int(entry.get("at") or 0) >= since] newest = list(reversed(items)) return newest[:limit], len(newest) > limit def recent(self, since: int | None = None) -> tuple[list[dict], bool]: cutoff = int(since if since is not None else time.time() - DAY_SECONDS) events, _ = self.page(since=cutoff, limit=1000) older = any(int(entry.get("at") or 0) < cutoff for entry in self._entries) return events, older