#!/usr/bin/env python3 """ notification_scheduler.py — Recurring due/overdue notification scanner for Togen (DVI-338). Part of the Togen Notifications System (DVI-332). A standalone, cron/systemd-timer driven CLI that scans the recurring work-order registry (``recurring.json``) once per run and raises recurring lifecycle notifications via the core dispatcher (``notifications.notify``). It is modeled on the ``folder_watcher.py`` / ``qr-watcher.service`` pattern but is a one-shot batch job rather than a daemon — the schedule lives in a systemd timer (authored separately by L2), not in this process. What it does ------------ For each *enabled* registry entry whose recurrence window is still open: * ``next_run`` is exactly **tomorrow** → raise ``workorder.recurring.due_soon`` (the "1 day before due" reminder). * ``next_run`` is **before today** → raise ``workorder.recurring.overdue``. Recipient resolution, template rendering and the actual send are all delegated to ``notifications.notify`` (the DVI-333 core module). This script only decides *which* event fires for *which* registry entry on a given day; it does not resolve recipients or render email itself, and it does **not** mutate the registry (advancing ``next_run`` and creating new WO instances is the web app's job, via ``/wor-recurring-run`` → ``_process_recurring_work_orders``). Idempotency ----------- The core module's file-based idempotency log (``notification_log.json``) guards every recurring send by ``(wor_id, event, recipient)``. This script sets ``wor_id`` to ``"@"`` so the idempotency scope is the specific *occurrence cycle*: * Running the scheduler multiple times on the same day never double-sends. * ``due_soon`` fires at most once for a given occurrence (it only matches on the single day-before-due). * ``overdue`` fires once per occurrence cycle — it does **not** re-send every day while an item stays overdue (``next_run`` is unchanged until the web app advances the cycle, so the idempotency key is stable). When the cycle advances to a new ``next_run``, that new occurrence is eligible for a fresh notification. Invocation contract (for the L2 systemd timer) ----------------------------------------------- * Command: /var/www/html/qr/venv/bin/python /var/www/html/togen/notification_scheduler.py * Working directory: /var/www/html/togen (so ``import notifications`` resolves and notification_log.json / user_settings/ are found alongside it) * Required env vars (read by notifications.py to send via Microsoft Graph): AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID, AZURE_MAIL_FROM * Optional env var: WOR_DIR (default /var/www/html/wor) — the directory that holds recurring.json. * Cadence: once per day (the due_soon / overdue windows are day-grained). * Exit code: 0 on a clean scan (including "nothing due"), non-zero only on an unexpected error (e.g. unreadable registry). Example systemd timer (reference — L2 owns the final unit files): # /etc/systemd/system/togen-notify.service [Unit] Description=Togen recurring due/overdue notification scan After=network-online.target [Service] Type=oneshot User=www-data Group=www-data WorkingDirectory=/var/www/html/togen EnvironmentFile=/etc/togen/notify.env # AZURE_* + AZURE_MAIL_FROM ExecStart=/var/www/html/qr/venv/bin/python /var/www/html/togen/notification_scheduler.py # /etc/systemd/system/togen-notify.timer [Unit] Description=Run Togen recurring notification scan daily [Timer] OnCalendar=*-*-* 07:00:00 Persistent=true [Install] WantedBy=timers.target Usage: python notification_scheduler.py [--dry-run] [--date YYYY-MM-DD] [--registry PATH] [--log-level LEVEL] --dry-run Resolve recipients and log what *would* be sent without sending or writing to notification_log.json. --date Override "today" (for testing); defaults to the system date. --registry Override the recurring.json path; defaults to $WOR_DIR/recurring.json. """ import argparse import json import logging import os import sys from datetime import date, datetime, timedelta from pathlib import Path # notification_scheduler.py lives in the togen app dir next to notifications.py, # so the script directory (auto-added to sys.path[0]) makes this import resolve. try: import notifications except ImportError as exc: # pragma: no cover - import guard sys.exit( f"Cannot import the notifications core module: {exc}\n" "Run this script from the togen app directory (the one containing " "notifications.py), e.g. WorkingDirectory=/var/www/html/togen." ) WOR_DIR = Path(os.environ.get("WOR_DIR", "/var/www/html/wor")) DEFAULT_REGISTRY_PATH = WOR_DIR / "recurring.json" def load_registry(path: Path) -> dict: """Load the recurring registry. Returns {} when the file is absent/empty.""" if not path.exists(): return {} data = json.loads(path.read_text()) if not isinstance(data, dict): raise ValueError(f"recurring.json is not a JSON object: {path}") return data def _build_context(source_filename: str, entry: dict, next_run: str) -> dict: """Map a registry entry into the context dict consumed by notifications.notify. ``wor_id`` is occurrence-scoped (filename@next_run) so the core idempotency log treats each recurrence cycle as a distinct notification scope. """ template = entry.get("template_data", {}) or {} return { "wor_id": f"{source_filename}@{next_run}", "wor_number": template.get("_wor_number"), "wor_type": entry.get("wor_type") or template.get("_wor_type"), # Recipient resolution sources (see notifications.resolve_recipients): # Source A — per-WO opt-in sidecar block, if captured in the template. # Source B — submitter's standing per-(type x event) preference. "notify": template.get("_notify"), "submitter_email": template.get("email"), "email": template.get("email"), # Template fields for the rendered email body. "requestor_name": template.get("requestor_name"), "location": template.get("location"), "building": template.get("building"), "urgency": template.get("urgency"), "desired_completion_date": template.get("desired_completion_date"), "description": template.get("description"), "next_run": next_run, } def _event_for_entry(next_run: str, today: date, logger: logging.Logger): """Return the event constant due for this entry today, or None. due_soon when next_run is exactly tomorrow; overdue when next_run is before today. next_run == today raises nothing (the web app creates the instance on the due date; the day-before reminder already covered it). """ try: nr = date.fromisoformat(next_run) except (ValueError, TypeError): logger.warning("Skipping entry with invalid next_run %r", next_run) return None if nr == today + timedelta(days=1): return notifications.EVENT_WO_RECURRING_DUE_SOON if nr < today: return notifications.EVENT_WO_RECURRING_OVERDUE return None def _make_sender(dry_run: bool, logger: logging.Logger, stats: dict): """Build the (to, subject, html) -> bool sender passed to notify(). In dry-run mode it logs and returns False so the core never records to notification_log.json. Otherwise it delegates to the real Graph send helper. """ def sender(to, subject, html): stats["attempted"] += 1 if dry_run: logger.info("[DRY-RUN] would send to %s | %s", to, subject) return False ok = notifications.send_html_email(to, subject, html) if ok: stats["sent"] += 1 logger.info("sent to %s | %s", to, subject) else: stats["failed"] += 1 logger.error("send FAILED to %s | %s", to, subject) return ok return sender def scan(registry: dict, today: date, *, dry_run: bool, logger: logging.Logger) -> dict: """Scan the registry and dispatch due_soon / overdue events. Returns stats.""" today_str = today.isoformat() stats = { "entries": len(registry), "due_soon": 0, "overdue": 0, "skipped_disabled": 0, "skipped_expired": 0, "attempted": 0, "sent": 0, "failed": 0, } sender = _make_sender(dry_run, logger, stats) for source_filename, entry in registry.items(): if not isinstance(entry, dict): logger.warning("Skipping malformed registry entry: %s", source_filename) continue if not entry.get("enabled", True): stats["skipped_disabled"] += 1 continue end_date = entry.get("end_date", "") if end_date and end_date < today_str: stats["skipped_expired"] += 1 continue next_run = entry.get("next_run", "") if not next_run: continue event = _event_for_entry(next_run, today, logger) if not event: continue context = _build_context(source_filename, entry, next_run) label = "due_soon" if event == notifications.EVENT_WO_RECURRING_DUE_SOON else "overdue" recipients = notifications.notify( event, context, background=False, sender=sender) stats[label] += 1 logger.info( "%s: %s (next_run=%s) -> %d recipient(s) resolved", label, source_filename, next_run, len(recipients)) return stats def main() -> None: parser = argparse.ArgumentParser( description="Scan recurring.json and raise due_soon/overdue notifications", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument( "--registry", default=str(DEFAULT_REGISTRY_PATH), help="Path to recurring.json (default: $WOR_DIR/recurring.json)", ) parser.add_argument( "--date", default=None, help="Override today's date as YYYY-MM-DD (for testing)", ) parser.add_argument( "--dry-run", action="store_true", help="Log what would be sent without sending or writing the idempotency log", ) parser.add_argument( "--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"], help="Logging verbosity", ) args = parser.parse_args() logging.basicConfig( level=args.log_level, format="%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%dT%H:%M:%S", stream=sys.stdout, ) logger = logging.getLogger("notification_scheduler") if args.date: try: today = date.fromisoformat(args.date) except ValueError: logger.error("Invalid --date %r (expected YYYY-MM-DD)", args.date) sys.exit(2) else: today = datetime.now().date() registry_path = Path(args.registry) try: registry = load_registry(registry_path) except (OSError, ValueError, json.JSONDecodeError) as exc: logger.error("Could not read registry %s: %s", registry_path, exc) sys.exit(1) logger.info( "Scanning %s (today=%s%s)", registry_path, today.isoformat(), ", DRY-RUN" if args.dry_run else "") stats = scan(registry, today, dry_run=args.dry_run, logger=logger) logger.info( "Scan complete: %d entr(ies) | due_soon=%d overdue=%d | " "sends attempted=%d sent=%d failed=%d | " "skipped: disabled=%d expired=%d", stats["entries"], stats["due_soon"], stats["overdue"], stats["attempted"], stats["sent"], stats["failed"], stats["skipped_disabled"], stats["skipped_expired"]) if __name__ == "__main__": main()