dc831b430b
Move modules from fluxer_admin_alerts/ to the repository root, switch to absolute imports, fix PROJECT_ROOT resolution, and update README commands and notifier paths.
68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
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
|