110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
"""Bounded, disk-persisted notification history.
|
|
|
|
Pure data structure — no D-Bus/GLib knowledge, mirrors the window/server
|
|
separation used elsewhere in beacon (window.py knows nothing about D-Bus
|
|
either). server.py owns turning these entries into GLib.Variant dicts and
|
|
emitting HistoryAdded/HistoryRemoved/HistoryCleared.
|
|
|
|
Entry schema (plain dict, JSON-safe, deliberately open so new keys can be
|
|
added later without breaking old readers):
|
|
id (int), app_name (str), summary (str), body (str), icon (str),
|
|
urgency (int), timestamp (float, unix seconds), reason (int),
|
|
actions (list[str], flat key/label pairs), controls (str | None — the raw
|
|
x-beacon-controls JSON hint, re-parsed via controls.parse_controls on Pop).
|
|
|
|
image_data (raw pixel bytes) is deliberately never stored here — it would
|
|
bloat history.json and doesn't round-trip through JSON cleanly; a
|
|
popped-from-history card just falls back to its `icon` string.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from typing import Optional
|
|
|
|
from paths import CACHE_DIR, HISTORY_FILE, ensure_dirs
|
|
|
|
|
|
class HistoryStore:
|
|
def __init__(self, maxlen: int, persist: bool) -> None:
|
|
self._maxlen = max(1, maxlen)
|
|
self._persist = persist
|
|
self._entries: list[dict] = self._load() if persist else [] # newest first
|
|
|
|
# -- mutation ---------------------------------------------------------------
|
|
def record(self, entry: dict) -> Optional[dict]:
|
|
"""Insert newest-first; returns the evicted entry if capacity was hit."""
|
|
self._entries.insert(0, entry)
|
|
evicted = None
|
|
while len(self._entries) > self._maxlen:
|
|
evicted = self._entries.pop()
|
|
self._save()
|
|
return evicted
|
|
|
|
def pop(self, nid: int) -> Optional[dict]:
|
|
"""Remove and return a specific entry (for History1.Pop)."""
|
|
for i, e in enumerate(self._entries):
|
|
if e["id"] == nid:
|
|
entry = self._entries.pop(i)
|
|
self._save()
|
|
return entry
|
|
return None
|
|
|
|
def pop_latest(self) -> Optional[dict]:
|
|
"""Remove and return the most recently recorded entry (LIFO, like
|
|
dunstctl history-pop)."""
|
|
if not self._entries:
|
|
return None
|
|
entry = self._entries.pop(0)
|
|
self._save()
|
|
return entry
|
|
|
|
def remove(self, nid: int) -> bool:
|
|
for i, e in enumerate(self._entries):
|
|
if e["id"] == nid:
|
|
del self._entries[i]
|
|
self._save()
|
|
return True
|
|
return False
|
|
|
|
def clear(self) -> list[int]:
|
|
ids = [e["id"] for e in self._entries]
|
|
self._entries = []
|
|
self._save()
|
|
return ids
|
|
|
|
# -- read ---------------------------------------------------------------------
|
|
def list_all(self) -> list[dict]:
|
|
return list(self._entries)
|
|
|
|
def get(self, nid: int) -> Optional[dict]:
|
|
for e in self._entries:
|
|
if e["id"] == nid:
|
|
return e
|
|
return None
|
|
|
|
# -- persistence ------------------------------------------------------------
|
|
def _load(self) -> list[dict]:
|
|
try:
|
|
data = json.loads(HISTORY_FILE.read_text())
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return []
|
|
return data if isinstance(data, list) else []
|
|
|
|
def _save(self) -> None:
|
|
if not self._persist:
|
|
return
|
|
ensure_dirs()
|
|
fd, tmp = tempfile.mkstemp(dir=str(CACHE_DIR), prefix=".history-", suffix=".json")
|
|
try:
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(self._entries, f)
|
|
os.replace(tmp, HISTORY_FILE)
|
|
except OSError:
|
|
try:
|
|
os.unlink(tmp)
|
|
except OSError:
|
|
pass
|