72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
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
|