#!/usr/bin/env python3 """ scada_report_scheduler.py — Scheduled SCADA report execution + email delivery (DVI-486, SCADA View Reporting ST4). Part of SCADA View Reporting (DVI-474). A standalone, systemd-timer driven CLI that runs once per invocation, scans the saved report schedules (``scada_report_schedules.json``, authored by ST3 / DVI-484), generates the XLSX for every schedule that is *due*, persists it under ``reports/``, and — when the schedule opts in — emails it as a base64 file attachment via Microsoft Graph. Modeled on ``notification_scheduler.py``: it imports the small self-contained engine modules (``scada_report`` for report/XLSX generation, ``notifications`` for the Graph send) and never imports app.py, which starts background loops at import time and is unsuitable for a one-shot job. Due-date / cadence model ------------------------ A schedule's ``cadence`` drives BOTH the report's data window and how often it fires. The window always ends on *yesterday* (the last complete log day): cadence interval data window (run date = today) last_week 7 days [today-7, today-1] last_month 30 days [today-30, today-1] custom N days [today-N, today-1] (N = schedule.custom_days) There is no separate firing anchor stored on the schedule, so the run log (``scada_report_log.json``) doubles as the last-run tracker: a schedule is due when it has never run, or when its most recent successful run is at least ``interval`` days old. First invocation after a schedule is created runs it immediately. Idempotency ----------- File-based, mirroring the ``notification_log.json`` pattern but in its own ``scada_report_log.json`` (the report run shape differs from notification sends). The idempotency key is ``(schedule_id, run_date)`` — a schedule runs at most once per calendar day, so a second invocation the same day is a no-op. A run is recorded unless the email was attempted and failed transiently (so a failed send is retried on the next daily pass); permission/config issues (no recipients, oversized attachment) are recorded to avoid daily regeneration. Attachment size guard --------------------- Graph's simple sendMail caps total message size near 4 MB. If a generated XLSX exceeds ``MAX_ATTACHMENT_BYTES`` (3 MB) the report is still persisted to ``reports/`` but the email is skipped with a warning (never silently truncated). Recipient model (DVI-487 — final shape) --------------------------------------- The Settings → Notifications pane drives two global lists plus per-schedule toggles. Recipients are OR-merged and de-duplicated by lowercased email from: 1. ``schedule["recipients"]`` — per-schedule list, included when ``schedule["notify"]`` (User notifications) is on 2. ``scada_enforced_recipients.json`` — global enforced list, always included (every report) 3. ``scada_external_recipients.json`` — global external list, included only when ``schedule["external"]`` is on Each global file may be a flat JSON list ``["a@b.com", ...]`` or an object ``{"recipients": ["a@b.com", ...]}``. Missing files resolve to no recipients. A report is emailed when ``notify`` OR ``external`` is set; if neither is on the report is persisted only. When sending but no recipients resolve, the email is skipped with a warning. Invocation contract (for the systemd timer in services/) -------------------------------------------------------- * Command: /bin/python /var/www/html/togen/scada_report_scheduler.py * Working directory: /var/www/html/togen (so ``import scada_report`` / ``import notifications`` resolve and the JSON state files are found alongside them) * Required env vars: AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID (Graph fetch + send), AZURE_MAIL_FROM (send-from address) * Optional env vars: SCADA_MAILBOX, SCADA_REPORT_SCHEDULES, SCADA_REPORTS_DIR * Cadence: run daily; the per-schedule cadence decides what is due. * Exit code: 0 on a clean scan (including "nothing due"); non-zero only on an unexpected error (e.g. unreadable schedules file). Usage: python scada_report_scheduler.py [--dry-run] [--date YYYY-MM-DD] [--schedules PATH] [--log-level LEVEL] --dry-run Generate + persist reports and resolve recipients, but do not send email or write the run log. --date Override "today" (for testing); defaults to the system date. --schedules Override the schedules JSON path. """ import argparse import json import logging import os import re import sys from datetime import date, datetime, timedelta from pathlib import Path from urllib.parse import quote try: import scada_report import scada_report_approvals import notifications except ImportError as exc: # pragma: no cover - import guard sys.exit( f"Cannot import a required module: {exc}\n" "Run this script from the togen app directory (the one containing " "scada_report.py and notifications.py), e.g. " "WorkingDirectory=/var/www/html/togen." ) _BASE_DIR = Path(__file__).resolve().parent DEFAULT_SCHEDULES_PATH = Path( os.environ.get("SCADA_REPORT_SCHEDULES", _BASE_DIR / "scada_report_schedules.json")) REPORTS_DIR = Path(os.environ.get("SCADA_REPORTS_DIR", _BASE_DIR / "reports")) RUN_LOG_FILE = _BASE_DIR / "scada_report_log.json" ENFORCED_RECIPIENTS_FILE = _BASE_DIR / "scada_enforced_recipients.json" EXTERNAL_RECIPIENTS_FILE = _BASE_DIR / "scada_external_recipients.json" # Graph simple sendMail message cap is ~4 MB; stay under it for the XLSX body. MAX_ATTACHMENT_BYTES = 3 * 1024 * 1024 XLSX_CONTENT_TYPE = ( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") # cadence -> trailing window length in days (also the firing interval). CADENCE_INTERVAL_DAYS = {"last_week": 7, "last_month": 30} # --------------------------------------------------------------------------- # Run log (idempotency + last-run tracker) # --------------------------------------------------------------------------- def load_run_log(path: Path = None) -> list: """Load the report run log (list of run records). [] when absent/invalid.""" path = path or RUN_LOG_FILE if not path.is_file(): return [] try: data = json.loads(path.read_text()) return data if isinstance(data, list) else [] except (json.JSONDecodeError, OSError): return [] def _append_run_log(record: dict, path: Path = None) -> None: path = path or RUN_LOG_FILE log = load_run_log(path) log.append(record) try: path.write_text(json.dumps(log, indent=2)) except OSError: logging.getLogger("scada_report_scheduler").exception( "Failed to persist %s", path) def _last_run_date(log: list, schedule_id: str): """Most recent *scheduled* run_date (date) for a schedule, or None. Manual (Test / run-now) runs are tagged ``source:"manual"`` and are excluded so a manual run — which sends no email (DVI-1164 D3) — never counts as the day's scheduled delivery and thus never suppresses it. """ latest = None for entry in log: if entry.get("schedule_id") != schedule_id: continue if entry.get("source") == "manual": continue try: d = date.fromisoformat(entry.get("run_date", "")) except (ValueError, TypeError): continue if latest is None or d > latest: latest = d return latest def _ran_on(log: list, schedule_id: str, run_date: date) -> bool: """True if a *scheduled* run for this schedule already landed on ``run_date``. Manual runs are ignored (see ``_last_run_date``).""" iso = run_date.isoformat() return any(e.get("schedule_id") == schedule_id and e.get("run_date") == iso and e.get("source") != "manual" for e in log) # --------------------------------------------------------------------------- # Schedule cadence helpers # --------------------------------------------------------------------------- def _interval_days(schedule: dict): """Firing interval / window length in days for a schedule, or None if the cadence is unrecognized or custom_days is invalid.""" cadence = (schedule.get("cadence") or "").strip() if cadence in CADENCE_INTERVAL_DAYS: return CADENCE_INTERVAL_DAYS[cadence] if cadence == "custom": try: n = int(schedule.get("custom_days")) except (TypeError, ValueError): return None return n if n >= 1 else None return None def _window_for(today: date, interval: int): """Trailing data window ending yesterday: (date_from, date_to).""" date_to = today - timedelta(days=1) date_from = date_to - timedelta(days=interval - 1) return date_from, date_to def is_due(schedule: dict, today: date, log: list, interval: int) -> bool: """A schedule is due when it hasn't run today and its last successful run is at least ``interval`` days old (or it has never run).""" sid = schedule.get("id") if not sid: return False if _ran_on(log, sid, today): return False last = _last_run_date(log, sid) if last is None: return True return (today - last).days >= interval # --------------------------------------------------------------------------- # Recipient resolution (interim model — see module docstring; ST6 owns final) # --------------------------------------------------------------------------- def _read_recipient_file(path: Path) -> list: if not path.is_file(): return [] try: data = json.loads(path.read_text()) except (json.JSONDecodeError, OSError): return [] if isinstance(data, dict): data = data.get("recipients", []) if not isinstance(data, list): return [] return [e for e in data if isinstance(e, str) and e.strip()] def _dedupe(addrs) -> list: ordered, seen = [], set() for addr in addrs: if not isinstance(addr, str): continue norm = addr.strip().lower() if norm and norm not in seen: seen.add(norm) ordered.append(addr.strip()) return ordered def resolve_internal_recipients(schedule: dict) -> list: """Internal audience emailed at generation time (no approval): the per-schedule list when ``notify`` is on + the global enforced list.""" addrs = [] if schedule.get("notify"): per_schedule = schedule.get("recipients") if isinstance(per_schedule, list): addrs.extend(per_schedule) addrs.extend(_read_recipient_file(ENFORCED_RECIPIENTS_FILE)) return _dedupe(addrs) def resolve_external_recipients(schedule: dict) -> list: """External audience — the global external list, included only when the schedule's ``external`` toggle is on. Gated behind approval (DVI-1102 F4).""" if not schedule.get("external"): return [] return _dedupe(_read_recipient_file(EXTERNAL_RECIPIENTS_FILE)) def resolve_recipients(schedule: dict) -> list: """Full OR-merged recipient list (internal + external). Retained for callers that want the combined audience; the scheduler itself sends the two groups separately so external delivery can be held for approval.""" return _dedupe(resolve_internal_recipients(schedule) + resolve_external_recipients(schedule)) # --------------------------------------------------------------------------- # Report rendering helpers # --------------------------------------------------------------------------- def _slug(name: str) -> str: s = re.sub(r"[^A-Za-z0-9._-]+", "-", (name or "report").strip()) return s.strip("-") or "report" def _friendly_range(date_from: date, date_to: date) -> str: def fmt(d): return f"{d.month}.{d.day}.{d.strftime('%y')}" return f"{fmt(date_from)} - {fmt(date_to)}" def _email_subject(schedule: dict, date_from: date, date_to: date) -> str: return (f"Kiln Metrics Report: {schedule.get('name', 'Report')} " f"({_friendly_range(date_from, date_to)})") def _base_url() -> str: """Public Togen base URL for building the in-app viewer deep link.""" return (os.environ.get("TOGEN_BASE_URL") or "https://togen.icastinc.com").rstrip("/") def _viewer_cta_html(out_name: str) -> str: """A "View report online" button linking into the in-app Report Viewer. Deep-links to ``?open_scada_report=`` which the frontend handles by opening the SCADA tool and the persisted-report viewer modal. The route is SSO-gated, so this is for the internal audience only (DVI-1165 D2) — the scheduler never adds it to the external send (external gets the attachment only, via the approvals path). """ link = f"{_base_url()}/?open_scada_report={quote(out_name)}" return ( '
' 'View report online' '

' 'Opens the report in Togen (sign-in required).

' ).format(link=_esc(link)) def _email_html(schedule: dict, report: dict, date_from: date, date_to: date, *, criteria_summary=None, out_name=None, opts=None) -> str: """Internal report email body: intro + detail table + (opt) Attention summary block + (opt) viewer link CTA. The summary/link are gated by the ``report_email`` settings in ``opts`` (DVI-1165).""" opts = opts or scada_report.report_email_settings() sensors = schedule.get("sensors") or [] sensors_txt = ", ".join(sensors) if sensors else "All sensors" rows = [ ("Schedule", schedule.get("name", "")), ("Date range", f"{date_from.isoformat()} – {date_to.isoformat()}"), ("Sensors", sensors_txt), ("Rows", str(report.get("row_count", 0))), ] row_html = "".join( '{lbl}' '{val}' ''.format(lbl=_esc(lbl), val=_esc(val)) for lbl, val in rows) summary_html = scada_report.build_report_email_summary( report, criteria_summary, opts) body_html = ( '

' 'Your scheduled Kiln Metrics report is attached as an Excel workbook ' '(one tab per day).

' '' + row_html + '
' + summary_html) cta_html = "" if opts.get("include_viewer_link") and out_name: cta_html = _viewer_cta_html(out_name) footer_html = ( '
' '

You are receiving this because you are a recipient of ' 'this scheduled Kiln Metrics report.

') return notifications.render_email_shell( title=_esc(_email_subject(schedule, date_from, date_to)), body_html=body_html, subtitle="Kiln Metrics Report", cta_html=cta_html, footer_html=footer_html) def _esc(value): return (str(value) .replace("&", "&").replace("<", "<").replace(">", ">")) # --------------------------------------------------------------------------- # Core run # --------------------------------------------------------------------------- def run_schedule(schedule: dict, today: date, *, dry_run: bool, logger: logging.Logger, source: str = "schedule", send_email: bool = True) -> dict: """Generate + persist a schedule's report and (optionally) email it. Returns a run record dict (also the shape appended to the run log). The ``recorded`` key tells the caller whether to persist it: a transiently failed send is not recorded so it retries on the next pass. ``source`` tags the run in the log — ``"schedule"`` for a timer-driven pass, ``"manual"`` for a Test / run-now (DVI-1164). ``send_email`` gates all email: a manual Test run (D3) sets it False so the report is generated, persisted and recorded (openable in the viewer) but nothing is emailed and no external approval is enqueued. """ sid = schedule.get("id") name = schedule.get("name", "") interval = _interval_days(schedule) date_from, date_to = _window_for(today, interval) record = { "schedule_id": sid, "schedule_name": name, "run_date": today.isoformat(), "date_from": date_from.isoformat(), "date_to": date_to.isoformat(), "row_count": 0, "email_status": "skipped_notify_off", "recipients": [], "external_recipients": [], "report_file": "", "source": source, "recorded_at": datetime.now().isoformat(), "recorded": True, } # Resolve the schedule's report type (DVI-1095 F3). Schedules created before # the registry carry no report_type and resolve to the default (Kiln Temp), # so their output is byte-identical to the pre-registry engine. rtype = scada_report.get_report_type(schedule.get("report_type")) report = rtype.build(date_from, date_to, schedule.get("sensors") or None) if report["error"]: logger.error("schedule %r: report build failed: %s", name, report["error"]) record["email_status"] = "error:" + str(report["error"]) record["recorded"] = False # transient/permission — retry next pass return record record["row_count"] = report["row_count"] # Criteria settings come from scada_config.json (same file the Flask UI # edits), so scheduled workbooks carry the identical pass/fail Summary # sheet as on-demand downloads (DVI-1095 P3). A type without criteria # (criteria_summary -> None) simply renders no Summary sheet. summary = rtype.criteria_summary(report) buf = rtype.workbook(report, summary) data = buf.getvalue() # Persist to reports/. REPORTS_DIR.mkdir(parents=True, exist_ok=True) out_name = f"{_slug(name)}_{date_from.isoformat()}_{date_to.isoformat()}.xlsx" out_path = REPORTS_DIR / out_name try: out_path.write_bytes(data) record["report_file"] = out_name logger.info("schedule %r: wrote %s (%d rows, %d bytes)", name, out_path, report["row_count"], len(data)) except OSError: logger.exception("schedule %r: failed to persist report", name) record["email_status"] = "persist_failed" record["recorded"] = False return record # Sibling view-model JSON for the in-app read-only Report Viewer (DVI-1163): # the viewer reads this instead of re-parsing the XLSX. Best-effort — a # failure here must not fail the (already-persisted) report or its email. try: scada_report.write_report_view_model( out_path, report, summary, meta={"schedule_id": sid, "schedule_name": name, "report_type": rtype.id, "date_from": date_from.isoformat(), "date_to": date_to.isoformat(), "generated_at": datetime.now().isoformat(), "source": source, "report_file": out_name}) except OSError: logger.warning("schedule %r: failed to persist view-model JSON " "(report still saved)", name) # Manual Test / run-now (DVI-1164 D3): report is generated, persisted and # recorded (openable in the viewer) but NEVER emailed and no external # approval is enqueued — Test is a dry inspection of what the report will # look like, not a delivery. if not send_email: logger.info("schedule %r: manual run — report persisted + recorded, " "no email (D3)", name) record["email_status"] = "manual_no_email" return record internal = resolve_internal_recipients(schedule) external = resolve_external_recipients(schedule) record["recipients"] = internal record["external_recipients"] = external if not (schedule.get("notify") or schedule.get("external")): logger.info("schedule %r: notifications off — report persisted, no email", name) return record if not internal and not external: logger.warning("schedule %r: notifications on but no recipients resolved " "— email skipped", name) record["email_status"] = "skipped_no_recipients" return record if len(data) > MAX_ATTACHMENT_BYTES: logger.warning("schedule %r: XLSX %d bytes exceeds %d-byte limit — " "email skipped (report still persisted)", name, len(data), MAX_ATTACHMENT_BYTES) record["email_status"] = "skipped_oversized" return record # The internal email carries the exceptions-only Attention summary + the # SSO-gated viewer link (DVI-1165). External delivery goes through the # approvals path (attachment only, no link — D2) and never uses this body. email_opts = scada_report.report_email_settings() subject = _email_subject(schedule, date_from, date_to) html = _email_html(schedule, report, date_from, date_to, criteria_summary=summary, out_name=out_name, opts=email_opts) attach_name = f"{_friendly_range(date_from, date_to)}.xlsx" if dry_run: parts = [] if internal: parts.append(f"email {attach_name} to internal {', '.join(internal)}") if external: parts.append(f"enqueue external approval for {', '.join(external)}") logger.info("[DRY-RUN] schedule %r: would %s", name, "; ".join(parts) or "do nothing") record["email_status"] = "dry_run" return record # 1) Internal audience (per-schedule notify + enforced) — sent immediately, # no approval (unchanged behavior for trusted staff). A transient failure # here means we DON'T record the run and DON'T enqueue the external # approval, so the whole schedule cleanly retries next pass. internal_ok = True if internal: internal_ok = notifications.send_email_with_attachment( internal, subject, html, filename=attach_name, content=data, content_type=XLSX_CONTENT_TYPE) if internal_ok: logger.info("schedule %r: emailed %s to %d internal recipient(s)", name, attach_name, len(internal)) else: logger.error("schedule %r: internal email send FAILED — will retry " "next pass", name) record["email_status"] = "send_failed" record["recorded"] = False # transient — retry next pass return record # 2) External audience — NEVER auto-sent. Enqueue an approval request and # notify the configured approver(s); the persisted XLSX is emailed only # when an approver approves it (DVI-1102 F4). if not external: record["email_status"] = "sent" if internal else "skipped_no_recipients" return record result = scada_report_approvals.enqueue_external_send( report_name=name, report_type=rtype.id, date_from=date_from.isoformat() if date_from else None, date_to=date_to.isoformat() if date_to else None, row_count=report["row_count"], report_file=out_name, attach_name=attach_name, subject=subject, recipients=external, source="schedule", schedule_id=sid, requested_by="scheduler") record["approval_id"] = result["request"]["id"] if result["notified"]: logger.info("schedule %r: external send held for approval " "(%d recipient(s)); notified %d approver(s)", name, len(external), len(result["approvers"])) record["email_status"] = ("sent+pending_approval" if internal else "pending_approval") else: # No approver configured — external delivery is held (invariant intact) # but nobody was notified. Surface it LOUDLY so this doesn't silently # stall external delivery (DVI-1114 class of failure). logger.warning("schedule %r: external send held for approval but NO " "approver is configured — report will NOT reach external " "recipients until an approver is set in Kiln Metrics → " "Report Notifications", name) record["email_status"] = ("sent+pending_approval_no_approver" if internal else "pending_approval_no_approver") return record def scan(schedules: list, today: date, *, dry_run: bool, logger: logging.Logger) -> dict: """Run every due schedule. Returns summary stats.""" log = load_run_log() stats = {"schedules": len(schedules), "due": 0, "generated": 0, "emailed": 0, "skipped_not_due": 0, "invalid": 0} for schedule in schedules: if not isinstance(schedule, dict) or not schedule.get("id"): stats["invalid"] += 1 continue interval = _interval_days(schedule) if interval is None: logger.warning("schedule %r: invalid cadence — skipped", schedule.get("name")) stats["invalid"] += 1 continue if not is_due(schedule, today, log, interval): stats["skipped_not_due"] += 1 continue stats["due"] += 1 record = run_schedule(schedule, today, dry_run=dry_run, logger=logger) if not record.get("email_status", "").startswith(("error:", "persist")): stats["generated"] += 1 if record.get("email_status") == "sent": stats["emailed"] += 1 if record.pop("recorded", True) and not dry_run: _append_run_log(record) log.append(record) # keep in-memory log consistent for this pass return stats def load_schedules(path: Path) -> list: if not path.exists(): return [] data = json.loads(path.read_text()) if not isinstance(data, list): raise ValueError(f"schedules file is not a JSON list: {path}") return [s for s in data if isinstance(s, dict)] def main() -> None: parser = argparse.ArgumentParser( description="Run due SCADA report schedules and email the XLSX results", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("--schedules", default=str(DEFAULT_SCHEDULES_PATH), help="Path to scada_report_schedules.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="Generate + persist reports but do not send email " "or write the run 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("scada_report_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() schedules_path = Path(args.schedules) try: schedules = load_schedules(schedules_path) except (OSError, ValueError, json.JSONDecodeError) as exc: logger.error("Could not read schedules %s: %s", schedules_path, exc) sys.exit(1) logger.info("Scanning %s (today=%s%s)", schedules_path, today.isoformat(), ", DRY-RUN" if args.dry_run else "") stats = scan(schedules, today, dry_run=args.dry_run, logger=logger) logger.info( "Scan complete: %d schedule(s) | due=%d generated=%d emailed=%d | " "skipped: not_due=%d invalid=%d", stats["schedules"], stats["due"], stats["generated"], stats["emailed"], stats["skipped_not_due"], stats["invalid"]) if __name__ == "__main__": main()