from __future__ import annotations from dataclasses import dataclass, field, replace from datetime import datetime, timedelta from sources import Item from state import AlertRecord @dataclass(frozen=True) class Notice: kind: str title: str detail: str path: str | None urgent: bool @dataclass class Plan: notices: list[Notice] = field(default_factory=list) records: list[AlertRecord] = field(default_factory=list) def _reminder_due(item: Item, now: datetime) -> bool: return ( item.opened_at is not None and item.reminder_after is not None and item.opened_at + item.reminder_after <= now ) def decide( now: datetime, seen: dict[str, tuple[str, Item]], known: dict[str, AlertRecord], checked_sources: set[str], realert: timedelta | None, notify_cleared: bool, ) -> Plan: plan = Plan() for key, (source, item) in seen.items(): record = known.get(key) if record is None or not record.open: overdue = _reminder_due(item, now) plan.notices.append(Notice("new", item.title, item.detail, item.path, item.urgent)) plan.records.append(AlertRecord(key, source, item.title, now, now, overdue, True)) continue if not record.reminded and _reminder_due(item, now): opened = item.opened_at.strftime("%Y-%m-%d %H:%M UTC") if item.opened_at else "unknown" detail = f"{item.detail}\nOpen since {opened}" plan.notices.append(Notice("reminder", item.title, detail, item.path, item.urgent)) plan.records.append(replace(record, last_alerted=now, reminded=True)) elif realert is not None and now - record.last_alerted >= realert: plan.notices.append(Notice("repeat", item.title, item.detail, item.path, item.urgent)) plan.records.append(replace(record, last_alerted=now)) for key, record in known.items(): if not record.open or key in seen or record.source not in checked_sources: continue plan.records.append(replace(record, open=False)) if notify_cleared: plan.notices.append(Notice("cleared", record.title, "No longer open in the admin UI.", None, False)) return plan