commit 1edf368830c570b360a48375105442838903bba2 Author: Clarth Date: Fri Nov 13 20:21:02 2026 -0500 Initial commit: Fluxer admin alert script diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..b4a24bd --- /dev/null +++ b/.env.example @@ -0,0 +1,14 @@ +# Copy this file to .env and fill in the values. Never commit .env. + +# Base URL of the Fluxer API, including /v1. +FLUXER_API_URL=https://api.example.com/v1 + +# Admin API key, sent as "Authorization: Admin ". +# It needs the ACLs listed in the README. +FLUXER_ADMIN_API_KEY= + +# Base URL of your admin UI, used to build links in alerts. No trailing slash. +ADMIN_UI_URL=https://admin.example.com + +# Webhook URL for the channel that receives alerts (channel settings > Webhooks). +FLUXER_WEBHOOK_URL= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ead39c6 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.venv/ +__pycache__/ +*.pyc +.env +config.toml +state.db +*.log diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..f8bcbd7 --- /dev/null +++ b/LICENSE @@ -0,0 +1,5 @@ +"THE BEER-WARE LICENSE" (Revision 42) + +Clarth wrote this software. As long as you retain this notice you +can do whatever you want with it. If we meet some day, and you think +this stuff is worth it, you can buy me a beer in return. diff --git a/README.md b/README.md new file mode 100644 index 0000000..f1e8f38 --- /dev/null +++ b/README.md @@ -0,0 +1,127 @@ +# Fluxer Admin Alerts + +A small Python script that checks a Fluxer instance's Admin API for things an instance admin should look at, and sends an alert when something needs attention. + +It is meant to run on a schedule (cron). Each run it remembers what it has already told you, so you only get alerts for new items, plus a reminder on a schedule until the item is cleared in the admin UI. + +Alerts go to a Fluxer channel through a webhook. Other notifiers (for example Gotify) can be added later. + +## What it checks + +| Source (config name) | Admin API endpoint | Alerts when | Required ACL | +| --- | --- | --- | --- | +| `pending_reports` | `GET /v1/admin/reports?status=pending` | A report is waiting. Reminder if it is still open after `stale_after_hours`. Marked urgent when the reason is highest priority. | `report:view` | +| `pending_registrations` | `GET /v1/admin/instance/pending-registrations` | An account is waiting for approval. | `registration:review` | +| `discovery_applications` | `GET /v1/admin/discovery/applications` | A community is waiting for discovery approval. | `discovery:review` | +| `enforcement_review` | `GET /v1/admin/enforcement-decisions?review_queue=true` | An automated moderation decision needs a person to review it. | `enforcement:view` | + +## Requirements + +- Linux (tested on Debian) +- Python 3.11 or newer +- An Admin API key on your instance with the ACLs in the table above +- A Fluxer webhook URL for the channel that should receive alerts + +## Install + +These steps use a virtual environment (`.venv`), which is the supported way to run this project. + +```sh +sudo apt install python3 python3-venv + +cd /path/to/fluxer-admin-alerts +python3 -m venv .venv +. .venv/bin/activate +pip install -r requirements.txt + +cp .env.example .env +cp config.example.toml config.toml +``` + +Then edit `.env` and `config.toml` (see below). + +To run the tests: + +```sh +python -m unittest discover -s tests +``` + +## Configure + +There are two config files. Both are ignored by git, so your values stay on your machine. + +### `.env`: URLs and secrets + +| Variable | Description | +| --- | --- | +| `FLUXER_API_URL` | Base URL of the Fluxer API, including `/v1`. | +| `FLUXER_ADMIN_API_KEY` | Admin API key. | +| `ADMIN_UI_URL` | Base URL of your admin UI. Used to build links in alerts. | +| `FLUXER_WEBHOOK_URL` | Webhook URL for the alert channel. Required when the `fluxer` notifier is enabled. | + +### `config.toml`: behaviour + +| Setting | Default | Description | +| --- | --- | --- | +| `general.realert_hours` | `12` | Hours before an open item is alerted again. `0` turns repeats off. | +| `general.notify_cleared` | `true` | Send a message when an item is cleared. | +| `general.request_timeout` | `15` | Seconds to wait for each HTTP request. | +| `general.state_file` | `state.db` | Where the alert history is stored. | +| `general.notifiers` | `["fluxer"]` | Notifiers to use. | +| `sources..enabled` | `true` | Turn a source on or off. | +| `sources.pending_reports.stale_after_hours` | `48` | Hours a report can stay open before the reminder. | + +## Admin API key + +Create an Admin API key in the admin UI with these ACLs: + +- `report:view` +- `registration:review` +- `discovery:review` +- `enforcement:view` + +Fluxer only lets a key use ACLs that the account that owns it also has, so that account needs the same ACLs. + +Give the key only what it needs. Remove a source from `config.toml` and drop its ACL if you do not want that check. + +## Run + +Check what would be sent, without sending anything or saving state: + +```sh +.venv/bin/python -m fluxer_admin_alerts --dry-run +``` + +Run for real: + +```sh +.venv/bin/python -m fluxer_admin_alerts +``` + +Use `-v` for debug logging. Exit code `0` means success, `1` means at least one source or notifier failed, and `2` means the config is wrong. + +## Schedule with cron + +Edit your crontab with `crontab -e` and add a line to run every hour: + +```cron +0 * * * * cd /path/to/fluxer-admin-alerts && .venv/bin/python -m fluxer_admin_alerts >> alerts.log 2>&1 +``` + +Replace `/path/to/fluxer-admin-alerts` with the folder you cloned into. `alerts.log` is ignored by git. + +## How alerts work + +- **New item:** alerted once, the first time it is seen. +- **Reports:** after `stale_after_hours` (48 by default) a "Still open" reminder is sent if the report is still pending. +- **Repeats:** an item still open after `realert_hours` (12 by default) is alerted again, until it is cleared. +- **Cleared:** when an item is no longer returned by the API (for example it was resolved in the admin UI), a "Cleared" message is sent if `notify_cleared` is on. +- **Failures:** if the Fluxer API or the webhook is unreachable, nothing is recorded for that run, so the alerts are sent on the next run. If one source fails, the others still run, and items from the failed source are not marked as cleared. + +## Adding a notifier + +Notifiers live in `fluxer_admin_alerts/notifiers/`. A notifier subclasses `Notifier`, implements `send(messages)`, raises `NotifierError` on failure, and is registered in `notifiers/__init__.py`. Then add its name to `KNOWN_NOTIFIERS` in `config.py` and read any settings it needs from `.env`. + +## License + +Beer-ware. See [LICENSE](LICENSE). diff --git a/config.example.toml b/config.example.toml new file mode 100644 index 0000000..29c785e --- /dev/null +++ b/config.example.toml @@ -0,0 +1,35 @@ +# Copy this file to config.toml and adjust it. +# Secrets and URLs go in .env, not here. + +[general] +# Hours before an open item is alerted again while it is still open in the admin UI. +# Set to 0 to turn repeats off. +realert_hours = 12 + +# Send a message when an item is no longer open (cleared in the admin UI). +notify_cleared = true + +# Timeout in seconds for each request to the Fluxer API or webhook. +request_timeout = 15 + +# Where the alert history is kept (SQLite). Relative paths are from the project folder. +state_file = "state.db" + +# Notifiers to send alerts through. Available: "fluxer" +notifiers = ["fluxer"] + +# Each source can be turned off with enabled = false. + +[sources.pending_reports] +enabled = true +# Hours a report can stay open before a "still open" reminder is sent. +stale_after_hours = 48 + +[sources.pending_registrations] +enabled = true + +[sources.discovery_applications] +enabled = true + +[sources.enforcement_review] +enabled = true diff --git a/fluxer_admin_alerts/__init__.py b/fluxer_admin_alerts/__init__.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/fluxer_admin_alerts/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/fluxer_admin_alerts/__main__.py b/fluxer_admin_alerts/__main__.py new file mode 100644 index 0000000..63382b8 --- /dev/null +++ b/fluxer_admin_alerts/__main__.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +from .config import PROJECT_ROOT, ConfigError, load_settings +from .runner import run + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="fluxer-admin-alerts", + description="Check the Fluxer Admin API and send alerts for items that need attention.", + ) + parser.add_argument( + "--config", + type=Path, + default=PROJECT_ROOT / "config.toml", + help="path to the config file (default: config.toml in the project folder)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="show what would be sent without sending anything or saving state", + ) + parser.add_argument("-v", "--verbose", action="store_true", help="enable debug logging") + args = parser.parse_args(argv) + + logging.basicConfig( + level=logging.DEBUG if args.verbose else logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + ) + + try: + settings = load_settings(args.config) + except ConfigError as exc: + logging.error("%s", exc) + return 2 + + return run(settings, dry_run=args.dry_run) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fluxer_admin_alerts/config.py b/fluxer_admin_alerts/config.py new file mode 100644 index 0000000..e086f09 --- /dev/null +++ b/fluxer_admin_alerts/config.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import os +import tomllib +from dataclasses import dataclass +from datetime import timedelta +from pathlib import Path + +from dotenv import load_dotenv + +PROJECT_ROOT = Path(__file__).resolve().parent.parent + +KNOWN_NOTIFIERS = ("fluxer",) + +SOURCE_DEFAULTS: dict[str, dict[str, object]] = { + "pending_reports": {"enabled": True, "stale_after_hours": 48}, + "pending_registrations": {"enabled": True}, + "discovery_applications": {"enabled": True}, + "enforcement_review": {"enabled": True}, +} + + +class ConfigError(Exception): + pass + + +@dataclass(frozen=True) +class SourceSettings: + enabled: bool + stale_after_hours: float | None + + +@dataclass(frozen=True) +class Settings: + api_url: str + api_key: str + admin_ui_url: str + request_timeout: float + realert: timedelta | None + notify_cleared: bool + state_file: Path + notifiers: tuple[str, ...] + fluxer_webhook_url: str | None + sources: dict[str, SourceSettings] + + +def _required(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value: + raise ConfigError(f"{name} is not set. Add it to .env (see .env.example).") + return value + + +def load_settings(config_path: Path) -> Settings: + load_dotenv(PROJECT_ROOT / ".env") + + if not config_path.is_file(): + raise ConfigError(f"config file not found: {config_path}. Copy config.example.toml to config.toml.") + with config_path.open("rb") as fh: + raw = tomllib.load(fh) + + general = raw.get("general", {}) + raw_sources = raw.get("sources", {}) + + notifiers = tuple(general.get("notifiers", ["fluxer"])) + for name in notifiers: + if name not in KNOWN_NOTIFIERS: + raise ConfigError(f"unknown notifier in config: {name!r}. Known: {', '.join(KNOWN_NOTIFIERS)}") + + fluxer_webhook_url = None + if "fluxer" in notifiers: + fluxer_webhook_url = _required("FLUXER_WEBHOOK_URL") + + realert_hours = float(general.get("realert_hours", 12)) + + sources: dict[str, SourceSettings] = {} + for name, defaults in SOURCE_DEFAULTS.items(): + table = {**defaults, **raw_sources.get(name, {})} + sources[name] = SourceSettings( + enabled=bool(table["enabled"]), + stale_after_hours=( + float(table["stale_after_hours"]) if table.get("stale_after_hours") is not None else None + ), + ) + + return Settings( + api_url=_required("FLUXER_API_URL").rstrip("/"), + api_key=_required("FLUXER_ADMIN_API_KEY"), + admin_ui_url=_required("ADMIN_UI_URL").rstrip("/"), + request_timeout=float(general.get("request_timeout", 15)), + realert=timedelta(hours=realert_hours) if realert_hours > 0 else None, + notify_cleared=bool(general.get("notify_cleared", True)), + state_file=PROJECT_ROOT / str(general.get("state_file", "state.db")), + notifiers=notifiers, + fluxer_webhook_url=fluxer_webhook_url, + sources=sources, + ) diff --git a/fluxer_admin_alerts/engine.py b/fluxer_admin_alerts/engine.py new file mode 100644 index 0000000..31194d9 --- /dev/null +++ b/fluxer_admin_alerts/engine.py @@ -0,0 +1,68 @@ +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 diff --git a/fluxer_admin_alerts/fluxer_api.py b/fluxer_admin_alerts/fluxer_api.py new file mode 100644 index 0000000..52f5eb0 --- /dev/null +++ b/fluxer_admin_alerts/fluxer_api.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from typing import Any + +import requests + +from . import __version__ + +USER_AGENT = f"fluxer-admin-alerts/{__version__}" + + +class FluxerAPIError(Exception): + pass + + +class FluxerAPI: + def __init__(self, base_url: str, api_key: str, timeout: float) -> None: + self._base_url = base_url + self._timeout = timeout + self._session = requests.Session() + self._session.headers.update( + { + "Authorization": f"Admin {api_key}", + "Accept": "application/json", + "User-Agent": USER_AGENT, + } + ) + + def get(self, path: str, params: dict[str, Any] | None = None) -> Any: + try: + resp = self._session.get(f"{self._base_url}{path}", params=params, timeout=self._timeout) + except requests.RequestException as exc: + raise FluxerAPIError(f"GET {path} failed: {type(exc).__name__}") from exc + + if not resp.ok: + raise FluxerAPIError(f"GET {path} returned {resp.status_code} ({_error_code(resp)})") + try: + return resp.json() + except ValueError as exc: + raise FluxerAPIError(f"GET {path} returned invalid JSON") from exc + + +def _error_code(resp: requests.Response) -> str: + try: + body = resp.json() + except ValueError: + return "no error body" + if isinstance(body, dict): + return str(body.get("code", "unknown error")) + return "unknown error" diff --git a/fluxer_admin_alerts/notifiers/__init__.py b/fluxer_admin_alerts/notifiers/__init__.py new file mode 100644 index 0000000..72b8b61 --- /dev/null +++ b/fluxer_admin_alerts/notifiers/__init__.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from ..config import Settings +from .base import Message, Notifier, NotifierError +from .fluxer import FluxerWebhookNotifier + +__all__ = ["Message", "Notifier", "NotifierError", "build_notifiers"] + + +def build_notifiers(settings: Settings) -> list[Notifier]: + notifiers: list[Notifier] = [] + for name in settings.notifiers: + if name == "fluxer": + notifiers.append(FluxerWebhookNotifier(settings.fluxer_webhook_url or "", settings.request_timeout)) + else: + raise ValueError(f"no notifier named {name!r}") + return notifiers diff --git a/fluxer_admin_alerts/notifiers/base.py b/fluxer_admin_alerts/notifiers/base.py new file mode 100644 index 0000000..d489155 --- /dev/null +++ b/fluxer_admin_alerts/notifiers/base.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Sequence +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Message: + title: str + body: str + link: str | None + urgent: bool + + +class NotifierError(Exception): + pass + + +class Notifier(ABC): + @abstractmethod + def send(self, messages: Sequence[Message]) -> None: + """Deliver messages. Raise NotifierError if delivery failed.""" diff --git a/fluxer_admin_alerts/notifiers/fluxer.py b/fluxer_admin_alerts/notifiers/fluxer.py new file mode 100644 index 0000000..c7a5a0c --- /dev/null +++ b/fluxer_admin_alerts/notifiers/fluxer.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +import requests + +from .base import Message, Notifier, NotifierError + +EMBEDS_PER_MESSAGE = 10 +COLOR_URGENT = 0xE74C3C +COLOR_NORMAL = 0x3498DB +USERNAME = "Admin Alerts" + + +class FluxerWebhookNotifier(Notifier): + def __init__(self, webhook_url: str, timeout: float) -> None: + self._url = webhook_url + self._timeout = timeout + + def send(self, messages: Sequence[Message]) -> None: + for start in range(0, len(messages), EMBEDS_PER_MESSAGE): + batch = messages[start : start + EMBEDS_PER_MESSAGE] + payload = {"username": USERNAME, "embeds": [_embed(m) for m in batch]} + try: + resp = requests.post(self._url, json=payload, timeout=self._timeout) + except requests.RequestException as exc: + # The exception text contains the webhook URL, which holds the token, so only the type is kept. + raise NotifierError(f"Fluxer webhook request failed ({type(exc).__name__})") from None + if not resp.ok: + raise NotifierError(f"Fluxer webhook returned {resp.status_code}") + + +def _embed(message: Message) -> dict[str, Any]: + embed: dict[str, Any] = { + "title": message.title[:256], + "description": message.body[:4096], + "color": COLOR_URGENT if message.urgent else COLOR_NORMAL, + } + if message.link: + embed["url"] = message.link + return embed diff --git a/fluxer_admin_alerts/runner.py b/fluxer_admin_alerts/runner.py new file mode 100644 index 0000000..3d515f3 --- /dev/null +++ b/fluxer_admin_alerts/runner.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import logging +from datetime import datetime, timezone + +from .config import Settings +from .engine import Notice, decide +from .fluxer_api import FluxerAPI, FluxerAPIError +from .notifiers import NotifierError, Message, build_notifiers +from .sources import SOURCES, Item +from .state import StateStore + +log = logging.getLogger(__name__) + +KIND_PREFIX = { + "new": "New", + "reminder": "Still open", + "repeat": "Reminder", + "cleared": "Cleared", +} + + +def _to_message(notice: Notice, admin_ui_url: str) -> Message: + link = f"{admin_ui_url}{notice.path}" if notice.path else admin_ui_url + return Message( + title=f"{KIND_PREFIX[notice.kind]}: {notice.title}", + body=notice.detail, + link=link, + urgent=notice.urgent, + ) + + +def run(settings: Settings, dry_run: bool = False) -> int: + api = FluxerAPI(settings.api_url, settings.api_key, settings.request_timeout) + now = datetime.now(timezone.utc) + failures = 0 + checked: set[str] = set() + seen: dict[str, tuple[str, Item]] = {} + + for source in SOURCES: + cfg = settings.sources[source.name] + if not cfg.enabled: + log.debug("%s: disabled", source.name) + continue + try: + items = source.fetch(api, cfg) + except FluxerAPIError as exc: + log.error("%s: %s", source.name, exc) + failures += 1 + continue + checked.add(source.name) + log.info("%s: %d open", source.name, len(items)) + for item in items: + seen[item.key] = (source.name, item) + + store = StateStore(settings.state_file) + try: + known = store.load() + plan = decide(now, seen, known, checked, settings.realert, settings.notify_cleared) + messages = [_to_message(n, settings.admin_ui_url) for n in plan.notices] + + if dry_run: + for message in messages: + log.info("would send: %s | %s | %s", message.title, message.body.replace("\n", " / "), message.link) + log.info("dry run: %d messages, %d state changes, nothing sent or saved", len(messages), len(plan.records)) + return 1 if failures else 0 + + if messages: + try: + for notifier in build_notifiers(settings): + notifier.send(messages) + except NotifierError as exc: + log.error("%s; nothing was recorded, so these will be retried next run", exc) + return 1 + log.info("sent %d messages", len(messages)) + else: + log.info("nothing new to send") + + store.save(plan.records) + finally: + store.close() + + return 1 if failures else 0 diff --git a/fluxer_admin_alerts/sources.py b/fluxer_admin_alerts/sources.py new file mode 100644 index 0000000..7b0dc15 --- /dev/null +++ b/fluxer_admin_alerts/sources.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any + +from .config import SourceSettings +from .fluxer_api import FluxerAPI + +PAGE_SIZE = 200 +MAX_PAGES = 50 + + +@dataclass(frozen=True) +class Item: + key: str + title: str + detail: str + path: str | None = None + urgent: bool = False + opened_at: datetime | None = None + reminder_after: timedelta | None = None + + +FetchFunc = Callable[[FluxerAPI, SourceSettings], list[Item]] + + +def _parse_time(value: Any) -> datetime | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed + + +def fetch_pending_reports(api: FluxerAPI, cfg: SourceSettings) -> list[Item]: + items: list[Item] = [] + reminder_after = timedelta(hours=cfg.stale_after_hours) if cfg.stale_after_hours else None + offset = 0 + for _ in range(MAX_PAGES): + data = api.get( + "/admin/reports", + {"status": "pending", "limit": PAGE_SIZE, "offset": offset}, + ) + reports = data.get("reports", []) + for report in reports: + report_id = str(report["report_id"]) + reason = report.get("reason_label") or report.get("category") or "no reason given" + items.append( + Item( + key=f"report:{report_id}", + title="Report waiting for review", + detail=f"Reason: {reason}", + urgent=bool(report.get("reason_highest_priority")), + opened_at=_parse_time(report.get("reported_at")), + reminder_after=reminder_after, + ) + ) + offset += len(reports) + if not reports or offset >= int(data.get("total", 0)): + break + return items + + +def fetch_pending_registrations(api: FluxerAPI, cfg: SourceSettings) -> list[Item]: + data = api.get("/admin/instance/pending-registrations") + items: list[Item] = [] + for entry in data.get("pending_registrations", []): + user_id = str(entry["user_id"]) + name = entry.get("username") or user_id + items.append( + Item( + key=f"registration:{user_id}", + title="Account waiting for approval", + detail=f"User: {name}", + ) + ) + return items + + +def fetch_discovery_applications(api: FluxerAPI, cfg: SourceSettings) -> list[Item]: + data = api.get("/admin/discovery/applications") + applications = data if isinstance(data, list) else data.get("applications", []) + items: list[Item] = [] + for entry in applications: + guild_id = str(entry["guild_id"]) + name = entry.get("guild_name") or guild_id + items.append( + Item( + key=f"discovery:{guild_id}", + title="Community waiting for discovery approval", + detail=f"Community: {name}", + ) + ) + return items + + +def fetch_enforcement_review(api: FluxerAPI, cfg: SourceSettings) -> list[Item]: + data = api.get("/admin/enforcement-decisions", {"review_queue": "true", "limit": 100}) + items: list[Item] = [] + for entry in data.get("decisions", []): + decision_id = str(entry["decision_id"]) + kind = entry.get("kind", "decision") + subject = entry.get("subject_name") or entry.get("subject_id") or "unknown" + items.append( + Item( + key=f"enforcement:{decision_id}", + title="Automated enforcement needs review", + detail=f"{kind} on {entry.get('subject_type', 'unknown')} {subject}", + ) + ) + return items + + +@dataclass(frozen=True) +class Source: + name: str + fetch: FetchFunc + + +SOURCES: tuple[Source, ...] = ( + Source("pending_reports", fetch_pending_reports), + Source("pending_registrations", fetch_pending_registrations), + Source("discovery_applications", fetch_discovery_applications), + Source("enforcement_review", fetch_enforcement_review), +) diff --git a/fluxer_admin_alerts/state.py b/fluxer_admin_alerts/state.py new file mode 100644 index 0000000..c7ece8f --- /dev/null +++ b/fluxer_admin_alerts/state.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS alerts ( + key TEXT PRIMARY KEY, + source TEXT NOT NULL, + title TEXT NOT NULL, + first_seen TEXT NOT NULL, + last_alerted TEXT NOT NULL, + reminded INTEGER NOT NULL DEFAULT 0, + open INTEGER NOT NULL DEFAULT 1 +) +""" + + +@dataclass(frozen=True) +class AlertRecord: + key: str + source: str + title: str + first_seen: datetime + last_alerted: datetime + reminded: bool + open: bool + + +class StateStore: + def __init__(self, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + self._conn = sqlite3.connect(path) + self._conn.row_factory = sqlite3.Row + self._conn.execute(SCHEMA) + + def close(self) -> None: + self._conn.close() + + def load(self) -> dict[str, AlertRecord]: + rows = self._conn.execute("SELECT * FROM alerts").fetchall() + return { + row["key"]: AlertRecord( + key=row["key"], + source=row["source"], + title=row["title"], + first_seen=datetime.fromisoformat(row["first_seen"]), + last_alerted=datetime.fromisoformat(row["last_alerted"]), + reminded=bool(row["reminded"]), + open=bool(row["open"]), + ) + for row in rows + } + + def save(self, records: list[AlertRecord]) -> None: + if not records: + return + with self._conn: + self._conn.executemany( + """ + INSERT INTO alerts (key, source, title, first_seen, last_alerted, reminded, open) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(key) DO UPDATE SET + source = excluded.source, + title = excluded.title, + first_seen = excluded.first_seen, + last_alerted = excluded.last_alerted, + reminded = excluded.reminded, + open = excluded.open + """, + [ + ( + r.key, + r.source, + r.title, + r.first_seen.isoformat(), + r.last_alerted.isoformat(), + int(r.reminded), + int(r.open), + ) + for r in records + ], + ) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..5160da3 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +requests>=2.31 +python-dotenv>=1.0 diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..17cff82 --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import unittest +from datetime import datetime, timedelta, timezone + +from fluxer_admin_alerts.engine import decide +from fluxer_admin_alerts.sources import Item +from fluxer_admin_alerts.state import AlertRecord + +NOW = datetime(2026, 1, 10, 12, 0, tzinfo=timezone.utc) +REALERT = timedelta(hours=12) +SOURCE = "pending_registrations" + + +def make_item(opened_at=None, reminder_after=None) -> Item: + return Item( + key="registration:1", + title="Account waiting for approval", + detail="User: test", + path="/registrations", + opened_at=opened_at, + reminder_after=reminder_after, + ) + + +def run_decide(now, items, known, checked=frozenset({SOURCE}), realert=REALERT, notify_cleared=True): + seen = {item.key: (SOURCE, item) for item in items} + return decide(now, seen, known, set(checked), realert, notify_cleared) + + +def as_known(plan): + return {r.key: r for r in plan.records} + + +class DecideTests(unittest.TestCase): + def test_new_item_alerts_once(self): + first = run_decide(NOW, [make_item()], {}) + self.assertEqual([n.kind for n in first.notices], ["new"]) + + second = run_decide(NOW + timedelta(hours=1), [make_item()], as_known(first)) + self.assertEqual(second.notices, []) + + def test_repeats_after_realert_interval(self): + first = run_decide(NOW, [make_item()], {}) + later = run_decide(NOW + timedelta(hours=12), [make_item()], as_known(first)) + self.assertEqual([n.kind for n in later.notices], ["repeat"]) + + def test_realert_disabled(self): + first = run_decide(NOW, [make_item()], {}, realert=None) + later = run_decide(NOW + timedelta(days=30), [make_item()], as_known(first), realert=None) + self.assertEqual(later.notices, []) + + def test_report_reminder_after_stale_threshold(self): + report = make_item(opened_at=NOW, reminder_after=timedelta(hours=48)) + first = run_decide(NOW, [report], {}) + self.assertEqual([n.kind for n in first.notices], ["new"]) + + at_48h = run_decide(NOW + timedelta(hours=48), [report], as_known(first)) + self.assertEqual([n.kind for n in at_48h.notices], ["reminder"]) + + at_60h = run_decide(NOW + timedelta(hours=60), [report], as_known(at_48h)) + self.assertEqual([n.kind for n in at_60h.notices], ["repeat"]) + + def test_old_report_seen_first_time_does_not_double_alert(self): + old = make_item(opened_at=NOW - timedelta(hours=72), reminder_after=timedelta(hours=48)) + first = run_decide(NOW, [old], {}) + self.assertEqual([n.kind for n in first.notices], ["new"]) + + second = run_decide(NOW + timedelta(hours=1), [old], as_known(first)) + self.assertEqual(second.notices, []) + + def test_cleared_item_sends_notice_and_closes(self): + first = run_decide(NOW, [make_item()], {}) + gone = run_decide(NOW + timedelta(hours=1), [], as_known(first)) + self.assertEqual([n.kind for n in gone.notices], ["cleared"]) + self.assertFalse(gone.records[0].open) + + def test_cleared_notice_can_be_disabled(self): + first = run_decide(NOW, [make_item()], {}) + gone = run_decide(NOW + timedelta(hours=1), [], as_known(first), notify_cleared=False) + self.assertEqual(gone.notices, []) + self.assertFalse(gone.records[0].open) + + def test_unchecked_source_is_not_closed(self): + first = run_decide(NOW, [make_item()], {}) + skipped = run_decide(NOW + timedelta(hours=1), [], as_known(first), checked=set()) + self.assertEqual(skipped.notices, []) + self.assertEqual(skipped.records, []) + + def test_reopened_item_alerts_again(self): + closed = AlertRecord("registration:1", SOURCE, "x", NOW, NOW, False, False) + back = run_decide(NOW, [make_item()], {closed.key: closed}) + self.assertEqual([n.kind for n in back.notices], ["new"]) + + +if __name__ == "__main__": + unittest.main()