""" notifications.py — Event-driven notification dispatch core for Togen (DVI-333). Foundation module for the Togen Notifications System (DVI-332). Provides: * Namespaced event constants (EVENT_*). * notify(event, context) — resolves personal recipients, renders the branded HTML template for the event, and sends via Microsoft Graph on a background thread (mirrors the existing fire-and-forget send pattern in app.py). * send_html_email(to, subject, html) — shared Graph sendMail helper, reused by app.py callers (e.g. the News email path). * render_email_shell(...) — the branded, Outlook-safe table HTML shell. * A file-based idempotency log (notification_log.json) so recurring events never double-send to the same recipient. Recipient resolution OR-merges two independent sources, de-duplicated by email: 1. Per-WO opt-in (individual): the submitting user who checked "Enable notifications about this work order" — carried in the WO JSON sidecar under ``_notify``. Notifies only that submitter, about only that one work order. 2. Admin-wide grid prefs: the 3x5 User Settings / WO Admin "Settings" grid is a Work Order Admin preference that applies to ALL work orders submitted by ANY user. Every user whose grid enables (WO type x event) is notified, regardless of who submitted the work order. The grid save route is admin-gated in app.py, so only WO Admins can persist these prefs. This module is intentionally self-contained (it does not import app.py) so it can sit beneath app.py in the import graph and be imported/tested in isolation. It reads the same env vars and on-disk JSON layout that app.py uses. """ import base64 import json import os import threading from datetime import datetime, timezone from pathlib import Path import msal import requests as _http_requests # --------------------------------------------------------------------------- # Event constants (namespaced) # --------------------------------------------------------------------------- # Work order lifecycle + recurring events. Room is left for future namespaces # (mts.label.created, scada.threshold.exceeded) — do NOT implement those here. EVENT_WO_CREATED = "workorder.created" EVENT_WO_STATUS_IN_PROGRESS = "workorder.status.in_progress" EVENT_WO_STATUS_COMPLETED = "workorder.status.completed" EVENT_WO_STATUS_CANCELLED = "workorder.status.cancelled" EVENT_WO_RECURRING_DUE_SOON = "workorder.recurring.due_soon" EVENT_WO_RECURRING_OVERDUE = "workorder.recurring.overdue" # EVH sign-off lifecycle events (DVI-1017) EVENT_WO_EVH_AWAITING = "workorder.evh.awaiting_approval" EVENT_WO_EVH_FINALIZED = "workorder.evh.finalized" EVENT_WO_EVH_REJECTED = "workorder.evh.rejected" WORKORDER_EVENTS = ( EVENT_WO_CREATED, EVENT_WO_STATUS_IN_PROGRESS, EVENT_WO_STATUS_COMPLETED, EVENT_WO_STATUS_CANCELLED, EVENT_WO_RECURRING_DUE_SOON, EVENT_WO_RECURRING_OVERDUE, EVENT_WO_EVH_AWAITING, EVENT_WO_EVH_FINALIZED, EVENT_WO_EVH_REJECTED, ) # Recurring events fire on a schedule and must be idempotency-guarded. RECURRING_EVENTS = frozenset({ EVENT_WO_RECURRING_DUE_SOON, EVENT_WO_RECURRING_OVERDUE, }) # SCADA monitoring events (DVI-468). Fired by the background SCADA collector, # not by a work-order action. Recipients resolve from the per-user # ``scada_notifications`` block rather than the WO grid/Primary lists, so these # have a separate dispatch path (``notify_scada``). EVENT_SCADA_STALE = "scada.data.stale" EVENT_SCADA_OUT_OF_RANGE = "scada.sensor.out_of_range" SCADA_EVENTS = (EVENT_SCADA_STALE, EVENT_SCADA_OUT_OF_RANGE) # Maps each SCADA event to the per-user ``scada_notifications`` opt-in key. SCADA_EVENT_PREF = { EVENT_SCADA_STALE: "stale_data", EVENT_SCADA_OUT_OF_RANGE: "out_of_range", } # --------------------------------------------------------------------------- # Configuration (mirrors app.py — same env vars / on-disk layout) # --------------------------------------------------------------------------- _BASE_DIR = Path(__file__).resolve().parent USER_SETTINGS_DIR = _BASE_DIR / "user_settings" NOTIFICATION_LOG_FILE = _BASE_DIR / "notification_log.json" # Company-level Primary recipient lists, keyed by WO type (DVI-405). Shape: # {"": ["a@b.com", ...], ...} # A Primary recipient receives EVERY workorder event for that type, regardless # of submitter or per-event grid prefs. WOR_PRIMARY_FILE = _BASE_DIR / "wor_primary.json" # Company-level Policy-enforced notification grid (DVI-597). Shape mirrors the # stored per-user grid: {"": {"": true, ...}, ...}. When a # (wor_type x event) cell is enabled, the REQUESTING USER of each work order of # that type receives that event's email as a matter of policy, regardless of # their per-WO opt-in. OR-merged with the other recipient sources. WOR_POLICY_FILE = _BASE_DIR / "wor_policy.json" # RPM approval/policy store (DVI-614). Besides approval-permission and # auto-approval config, it carries a per-WO-type ``policy_notify`` email # allowlist: a listed user receives EVERY RPM part event for that type as a # matter of policy, regardless of their per-user RPM notifications grid. RPM_APPROVAL_FILE = _BASE_DIR / "rpm_approval.json" # Key under which the per-user notification grid lives in a user_settings file. # Shape (written by the User Settings grid subtask): # settings["wo_notifications"] = { # "": { "": true, ... }, # e.g. "maintenance" # ... # } USER_SETTINGS_NOTIFY_KEY = "wo_notifications" _AZ_CLIENT_ID = os.environ.get("AZURE_CLIENT_ID", "") _AZ_SECRET = os.environ.get("AZURE_CLIENT_SECRET", "") _AZ_TENANT = os.environ.get("AZURE_TENANT_ID", "") _AZ_AUTHORITY = f"https://login.microsoftonline.com/{_AZ_TENANT}" if _AZ_TENANT else "" AZURE_MAIL_FROM = os.environ.get("AZURE_MAIL_FROM", "") _GRAPH_SCOPES = ["https://graph.microsoft.com/.default"] # Serialize reads/writes of the idempotency log across dispatch threads. _log_lock = threading.Lock() import logging _log = logging.getLogger("togen.notifications") # --------------------------------------------------------------------------- # Branded HTML shell + shared email send helper # --------------------------------------------------------------------------- def render_email_shell(title, body_html, *, subtitle="Notification", meta_html="", cta_html="", footer_html=""): """Wrap content in the branded, Outlook-safe (table-based) HTML email shell. Same markup as app.py's news shell, generalized via the ``subtitle`` slot so every notification type can share one branded template. """ return ( '' '' '' '' '' '
' '' # Header bar '' # Body '' # Footer '' '
' 'togen' '' '
{subtitle}
' '
' '

{title}

' '{meta}' '
' '{body}' '{cta}' '
{footer}
' ).format(subtitle=subtitle, title=title, meta=meta_html, body=body_html, cta=cta_html, footer=footer_html) def _acquire_graph_token(): """Acquire an app-only Graph token. Returns access token str or None.""" if not (_AZ_CLIENT_ID and _AZ_SECRET and _AZ_AUTHORITY): _log.warning("Graph token unavailable: Azure credentials not configured") return None try: app_client = msal.ConfidentialClientApplication( _AZ_CLIENT_ID, authority=_AZ_AUTHORITY, client_credential=_AZ_SECRET) result = app_client.acquire_token_for_client(scopes=_GRAPH_SCOPES) token = result.get("access_token") if not token: _log.error("Graph token request returned no access_token") return token except Exception: _log.exception("Graph token acquisition error") return None def _send_graph_mail(to, subject, html, attachments=None): """Low-level Microsoft Graph sendMail. Synchronous. Returns True on accept. ``to`` may be a single address or an iterable of addresses. ``attachments`` is an optional list of ``{"name", "contentType", "contentBytes"}`` dicts where ``contentBytes`` is the base64-encoded file payload (Graph fileAttachment shape). From address is ``AZURE_MAIL_FROM``. """ if not AZURE_MAIL_FROM: _log.warning("Cannot send email: AZURE_MAIL_FROM not set") return False if isinstance(to, str): addresses = [to] else: addresses = [a for a in to if a] if not addresses: return False token = _acquire_graph_token() if not token: return False message = { "subject": subject, "body": {"contentType": "HTML", "content": html}, "toRecipients": [{"emailAddress": {"address": a}} for a in addresses], } if attachments: message["attachments"] = [ { "@odata.type": "#microsoft.graph.fileAttachment", "name": att["name"], "contentType": att.get( "contentType", "application/octet-stream"), "contentBytes": att["contentBytes"], } for att in attachments ] graph_body = {"message": message, "saveToSentItems": "false"} try: resp = _http_requests.post( f"https://graph.microsoft.com/v1.0/users/{AZURE_MAIL_FROM}/sendMail", headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, json=graph_body, timeout=60, ) if resp.status_code not in (200, 202): _log.error("sendMail failed (%s): %s", resp.status_code, resp.text[:300]) return False return True except Exception: _log.exception("sendMail error") return False def send_html_email(to, subject, html): """Send a branded HTML email via Microsoft Graph sendMail. Synchronous. ``to`` may be a single address string or an iterable of addresses (all are placed on a single message's toRecipients list). From address is ``AZURE_MAIL_FROM``. Returns True on a 200/202 accept, False otherwise. This is the shared helper extracted from app.py's admin/news send paths; callers that need fire-and-forget behaviour should run it on a thread. """ return _send_graph_mail(to, subject, html) def send_email_with_attachment(to, subject, html, *, filename, content, content_type): """Send an HTML email with a single file attachment via Graph sendMail. ``content`` is the raw file bytes; it is base64-encoded into a Graph fileAttachment. Synchronous; returns True on a 200/202 accept. Added for the SCADA report scheduler (DVI-486), which emails generated XLSX reports — ``send_html_email`` is HTML-only. Graph's simple sendMail caps total message size around 4 MB, so callers should guard large attachments before invoking this (the report scheduler skips XLSX payloads over 3 MB). """ encoded = base64.b64encode(content).decode("ascii") return _send_graph_mail( to, subject, html, attachments=[{ "name": filename, "contentType": content_type, "contentBytes": encoded, }], ) # --------------------------------------------------------------------------- # Recipient resolution (strictly personal — board decision #1) # --------------------------------------------------------------------------- def _normalize_email(addr): return addr.strip().lower() if isinstance(addr, str) else "" def _iter_grid_recipients(wor_type, event): """Yield the email of every user whose grid enables (wor_type x event). Scans every ``USER_SETTINGS_DIR/*.json``. The grid is admin-wide: a match means that user wants this notification for ALL work orders of this type, regardless of submitter. Because the grid save route is admin-gated in app.py, only WO Admins can have a ``wo_notifications`` block persisted, so this effectively yields the WO Admins subscribed to the cell. Each user's email is read from the stored ``email`` field when present, falling back to the settings filename stem (which equals the lowercased email for normal addresses). """ if not (wor_type and event): return try: files = sorted(USER_SETTINGS_DIR.glob("*.json")) except OSError: return for fp in files: try: settings = json.loads(fp.read_text()) except (json.JSONDecodeError, OSError): continue if not isinstance(settings, dict): continue prefs = settings.get(USER_SETTINGS_NOTIFY_KEY) if not isinstance(prefs, dict): continue type_prefs = prefs.get(wor_type) or {} if type_prefs.get(event): email = settings.get("email") or fp.stem if email: yield email def _iter_primary_recipients(wor_type): """Yield every email on the Primary list for this WO type (DVI-405). Primary recipients receive ALL workorder events for their WO type, regardless of submitter or per-event grid prefs. """ if not wor_type or not WOR_PRIMARY_FILE.is_file(): return try: data = json.loads(WOR_PRIMARY_FILE.read_text()) except (json.JSONDecodeError, OSError): return if not isinstance(data, dict): return emails = data.get(wor_type) if isinstance(emails, list): for e in emails: if isinstance(e, str) and e.strip(): yield e def _policy_enforces(wor_type, event): """True if the company Policy grid enforces (wor_type x event) (DVI-597).""" if not (wor_type and event) or not WOR_POLICY_FILE.is_file(): return False try: data = json.loads(WOR_POLICY_FILE.read_text()) except (json.JSONDecodeError, OSError): return False if not isinstance(data, dict): return False type_prefs = data.get(wor_type) return bool(isinstance(type_prefs, dict) and type_prefs.get(event)) # Key under which per-user SCADA notification prefs live (DVI-465). Shape: # settings["scada_notifications"] = {"stale_data": bool, "out_of_range": bool} SCADA_NOTIFY_KEY = "scada_notifications" def resolve_scada_recipients(pref_key): """Resolve de-duplicated emails of users who opted into a SCADA pref. Scans every ``USER_SETTINGS_DIR/*.json`` for a truthy ``scada_notifications[pref_key]`` (e.g. ``"stale_data"`` / ``"out_of_range"``). Email is read from the stored ``email`` field when present, falling back to the settings filename stem. Returns a list of original-cased email strings, order-stable, no duplicates. """ ordered = [] seen = set() if not pref_key: return ordered try: files = sorted(USER_SETTINGS_DIR.glob("*.json")) except OSError: return ordered for fp in files: try: settings = json.loads(fp.read_text()) except (json.JSONDecodeError, OSError): continue if not isinstance(settings, dict): continue prefs = settings.get(SCADA_NOTIFY_KEY) if not isinstance(prefs, dict) or not prefs.get(pref_key): continue email = settings.get("email") or fp.stem norm = _normalize_email(email) if norm and norm not in seen: seen.add(norm) ordered.append(email.strip()) return ordered def resolve_recipients(event, context): """Resolve the de-duplicated personal recipients for an event. Three independent (OR-merged) sources, de-duplicated by lowercased email: Source A — per-WO opt-in (individual): ``context["notify"]`` (the sidecar ``_notify`` block) of shape ``{"enabled": bool, "email": str}``. Notifies only that submitter, about only this one work order. Source B — admin-wide grid prefs: every user whose User Settings / WO Admin "Settings" grid enables (``context["wor_type"]`` x event), regardless of who submitted the work order. Source C — Primary list: every email flagged Primary for ``context["wor_type"]`` receives all events for that type (DVI-405). Source D — Policy-enforced grid: when the company Policy grid enables (``context["wor_type"]`` x event), the requesting user of the work order is notified as a matter of policy, regardless of opt-in (DVI-597). Returns a list of original-cased email strings, order-stable, no duplicates. """ ordered = [] # preserves first-seen original casing seen = set() # normalized emails already added def _add(addr): norm = _normalize_email(addr) if norm and norm not in seen: seen.add(norm) ordered.append(addr.strip()) # Source A — per-WO opt-in. notify_block = context.get("notify") or {} if isinstance(notify_block, dict) and notify_block.get("enabled"): _add(notify_block.get("email")) # Source B — admin-wide grid prefs: notify every user whose grid enables # this (wor_type x event) cell, regardless of who submitted the WO. wor_type = context.get("wor_type") for email in _iter_grid_recipients(wor_type, event): _add(email) # Source C — Primary list: every Primary email for this WO type gets all # events, regardless of submitter or grid prefs. for email in _iter_primary_recipients(wor_type): _add(email) # Source D — Policy-enforced (DVI-597): when the company Policy grid enables # this (wor_type x event), the requesting user of the work order is notified # as a matter of policy, regardless of their per-WO opt-in. The requesting # user is the signed-in submitter captured in the ``_notify`` sidecar. if _policy_enforces(wor_type, event): notify_block = context.get("notify") or {} requestor = (notify_block.get("requestor_email") or notify_block.get("email") or context.get("submitter_email") or context.get("email")) _add(requestor) # Source E — EVH lifecycle (DVI-1017 / DVI-1011 re-test): explicit EVH # recipients are always notified, independent of the grid/policy config, so # the approval participants reliably receive awaiting/finalized/rejected # mail even when no grid cell or Primary list is configured. Awaiting mail # carries the newly-actionable department's approvers in ``evh_dept_emails``; # finalized/rejected mail carries the full participant set in # ``evh_recipients`` (submitter + every sign-off approver + configured # approvers of the required departments). if event in (EVENT_WO_EVH_AWAITING, EVENT_WO_EVH_FINALIZED, EVENT_WO_EVH_REJECTED): for addr in (context.get("evh_dept_emails") or []): _add(addr) for addr in (context.get("evh_recipients") or []): _add(addr) return ordered # --------------------------------------------------------------------------- # Idempotency log (notification_log.json) # --------------------------------------------------------------------------- def _load_log(): if not NOTIFICATION_LOG_FILE.is_file(): return [] try: data = json.loads(NOTIFICATION_LOG_FILE.read_text()) return data if isinstance(data, list) else [] except (json.JSONDecodeError, OSError): return [] def _record_sent(wor_id, event, recipient): """Append a {wor_id, event, recipient, sent_at} entry to the log.""" entry = { "wor_id": wor_id, "event": event, "recipient": _normalize_email(recipient), "sent_at": datetime.now(timezone.utc).isoformat(), } with _log_lock: log = _load_log() log.append(entry) try: NOTIFICATION_LOG_FILE.write_text(json.dumps(log, indent=2)) except OSError: _log.exception("Failed to persist notification_log.json") def already_sent(wor_id, event, recipient): """True if (wor_id, event, recipient) was previously logged as sent.""" norm = _normalize_email(recipient) with _log_lock: for e in _load_log(): if (e.get("wor_id") == wor_id and e.get("event") == event and e.get("recipient") == norm): return True return False # --------------------------------------------------------------------------- # Templates # --------------------------------------------------------------------------- # Per-event subject + intro copy. ``{ref}`` is the human work-order reference. _EVENT_COPY = { EVENT_WO_CREATED: ( "Work Order {ref} submitted", "Your work order has been submitted and is awaiting review.", ), EVENT_WO_STATUS_IN_PROGRESS: ( "Work Order {ref} is in progress", "Work on your work order has started.", ), EVENT_WO_STATUS_COMPLETED: ( "Work Order {ref} completed", "Your work order has been marked complete.", ), EVENT_WO_STATUS_CANCELLED: ( "Work Order {ref} cancelled", "Your work order has been cancelled.", ), EVENT_WO_RECURRING_DUE_SOON: ( "Recurring Work Order {ref} due soon", "A recurring work order you submitted is due soon.", ), EVENT_WO_RECURRING_OVERDUE: ( "Recurring Work Order {ref} overdue", "A recurring work order you submitted is now overdue.", ), EVENT_WO_EVH_AWAITING: ( "RtS Approval needed — Work Order {ref}", "Your department's sign-off is required before this equipment can be returned to service.", ), EVENT_WO_EVH_FINALIZED: ( "Equipment released to production — Work Order {ref}", "All required departments have approved the Equipment Validation Handoff. The equipment has been released to production.", ), EVENT_WO_EVH_REJECTED: ( "RtS Approval rejected — Work Order {ref}", "A return-to-service approval was rejected. The work order has been reopened for further action.", ), } def _wor_reference(context): """Human-friendly work-order reference for subjects/body.""" num = context.get("wor_number") if num: try: return f"#{int(num):05d}" except (TypeError, ValueError): return f"#{num}" wid = context.get("wor_id") return str(wid) if wid else "" def _esc(value): return (str(value) .replace("&", "&").replace("<", "<").replace(">", ">")) def _base_url(): """Public Togen base URL for building in-app deep links.""" return (os.environ.get("TOGEN_BASE_URL") or "https://togen.icastinc.com").rstrip("/") def _wo_cta_html(context, label): """A deep-link button into WO Admin focused on this work order (DVI-1174). Links to ``?open_wo=`` which the frontend resolves by opening the WO Admin tool and selecting the WO (approver-only users land on the Approvals tab). SSO-gated; recipients without WO Admin access land on the Togen home (graceful). Returns "" when the context carries no work-order number. """ from urllib.parse import quote num = context.get("wor_number") if num in (None, ""): return "" try: token = f"{int(num):05d}" except (TypeError, ValueError): token = str(num) link = f"{_base_url()}/?open_wo={quote(token)}" return ( '
' '{label}' '

' 'Opens this work order in Togen (sign-in required).

' ).format(link=_esc(link), label=_esc(label)) def render_event_email(event, context): """Render (subject, html) for an event given its context. None if unknown.""" copy = _EVENT_COPY.get(event) if not copy: return None subject_tmpl, intro = copy ref = _wor_reference(context) subject = subject_tmpl.format(ref=ref or "").strip() rows = [] for label, key in (("Requestor", "requestor_name"), ("Location", "location"), ("Building", "building"), ("Urgency", "urgency"), ("Desired completion", "desired_completion_date")): val = context.get(key) if val: rows.append( '{lbl}' '{val}' ''.format(lbl=_esc(label), val=_esc(val))) detail_table = ( '' + "".join(rows) + '
' if rows else "") desc = context.get("description") desc_html = ( '

{}

'.format(_esc(desc)) if desc else "") body_html = ( '

{intro}

' '{table}{desc}'.format(intro=_esc(intro), table=detail_table, desc=desc_html)) footer_html = ( '
' '

You are receiving this because you enabled ' 'notifications for this work order.

') # Deep-link CTA back to the WO (DVI-1174): the approval-needed email gets a # "View & approve" button (P4); every other WO event gets a plain "open this # WO" button (P5). Both resolve to ?open_wo=. cta_label = ("View & approve in Togen" if event == EVENT_WO_EVH_AWAITING else "Open this work order in Togen") cta_html = _wo_cta_html(context, cta_label) html = render_email_shell( title=_esc(subject), body_html=body_html, subtitle="Work Order Update", cta_html=cta_html, footer_html=footer_html, ) return subject, html # --------------------------------------------------------------------------- # Dispatch entry point # --------------------------------------------------------------------------- def _dispatch(event, context, recipients, sender): """Send to each recipient once, honoring the idempotency log.""" rendered = render_event_email(event, context) if not rendered: _log.error("notify: no template for event %s", event) return subject, html = rendered wor_id = context.get("wor_id") guard = event in RECURRING_EVENTS for recipient in recipients: if guard and wor_id and already_sent(wor_id, event, recipient): continue ok = sender(recipient, subject, html) if ok and wor_id: _record_sent(wor_id, event, recipient) def notify(event, context, *, background=True, sender=None): """Resolve recipients, render the event template, and send. Args: event: one of the EVENT_* constants. context: dict carrying recipient-resolution inputs (``notify``, ``submitter_email``/``email``, ``wor_type``) and template fields (``wor_id``, ``wor_number``, ``requestor_name``, ``description``, …). background: when True (default) dispatch on a daemon thread (mirrors the existing fire-and-forget pattern); when False, dispatch inline (useful for tests / synchronous callers). sender: callable (to, subject, html) -> bool used to actually send; defaults to :func:`send_html_email`. Injectable for tests. Returns the resolved recipient list (so callers/tests can assert on it). """ if event not in WORKORDER_EVENTS: _log.error("notify: unknown event %r", event) return [] recipients = resolve_recipients(event, context) if not recipients: return [] send = sender or send_html_email if background: t = threading.Thread( target=_dispatch, args=(event, context, recipients, send), daemon=True) t.start() else: _dispatch(event, context, recipients, send) return recipients # --------------------------------------------------------------------------- # SCADA monitoring notifications (DVI-468) # --------------------------------------------------------------------------- def render_scada_email(event, context): """Render (subject, html) for a SCADA event. None for an unknown event. Context keys: stale event — ``stale_after_minutes`` (int), ``last_reading_time`` (str or None), ``age_minutes`` (int or None). out-of-range — ``sensor`` (display name), ``value`` (str), ``low`` / ``high`` (str/num or None), ``unit`` (str, optional). """ if event == EVENT_SCADA_STALE: mins = context.get("stale_after_minutes") subject = "SCADA alert: no new data received" last = context.get("last_reading_time") age = context.get("age_minutes") rows = [] if mins is not None: rows.append(("Stale-after threshold", "{} min".format(_esc(mins)))) if age is not None: rows.append(("Time since last reading", "{} min".format(_esc(age)))) if last: rows.append(("Last reading", _esc(last))) intro = ("The SCADA collector has not received a new reading within the " "configured window. Data may be stale — check the data source.") title = "SCADA data is stale" elif event == EVENT_SCADA_OUT_OF_RANGE: sensor = context.get("sensor") or context.get("label") or "Sensor" unit = context.get("unit") or "" title = "Sensor out of range" if context.get("reason") == "rate_of_change": # Change-over-time component of the out-of-range alert (DVI-1148): # a sensor changed too fast within the configured window. usuf = (" " + _esc(unit)) if unit else "" delta = context.get("rate_delta") window = context.get("rate_window") limit = context.get("rate_max_delta") span = context.get("rate_span") frm = context.get("rate_from") to = context.get("rate_to") subject = "SCADA alert: {} changing too fast".format(sensor) rows = [ ("Sensor", _esc(sensor)), ("Change observed", "{}{} in {} min".format( _esc(delta), usuf, _esc(span))), ("Allowed change", "≤ {}{} within {} min".format( _esc(limit), usuf, _esc(window))), ] if frm is not None and to is not None: rows.append(("Readings", "{}{} → {}{}".format( _esc(frm), usuf, _esc(to), usuf))) intro = ("A monitored SCADA sensor changed faster than its allowed " "rate of change in the most recent data received.") else: value = context.get("value") low = context.get("low") high = context.get("high") if low not in (None, "") and high not in (None, ""): rng = "{} – {}{}".format(_esc(low), _esc(high), (" " + _esc(unit)) if unit else "") elif low not in (None, ""): rng = "≥ {}{}".format(_esc(low), (" " + _esc(unit)) if unit else "") elif high not in (None, ""): rng = "≤ {}{}".format(_esc(high), (" " + _esc(unit)) if unit else "") else: rng = "—" subject = "SCADA alert: {} out of range".format(sensor) rows = [ ("Sensor", _esc(sensor)), ("Current value", "{}{}".format(_esc(value), (" " + _esc(unit)) if unit else "")), ("Configured range", rng), ] intro = ("A monitored SCADA sensor has reported a value outside its " "configured range.") else: return None row_html = "".join( '{lbl}' '{val}' ''.format(lbl=_esc(lbl), val=val) for lbl, val in rows) detail_table = ( '' + row_html + '
' if row_html else "") body_html = ( '

{intro}

{table}' .format(intro=_esc(intro), table=detail_table)) footer_html = ( '
' '

You are receiving this because you enabled SCADA ' 'notifications in your Togen settings.

') html = render_email_shell( title=_esc(title), body_html=body_html, subtitle="SCADA Alert", footer_html=footer_html) return subject, html def notify_scada(event, context, *, recipients=None, background=True, sender=None): """Send a SCADA monitoring email to opted-in users. Recipients default to everyone whose ``scada_notifications`` enables the pref mapped to ``event`` (see :data:`SCADA_EVENT_PREF`); pass an explicit ``recipients`` list to override. Unlike :func:`notify`, idempotency is the caller's responsibility — the SCADA collector only calls this on a state transition (entering staleness / a sensor going out of range), so emails are not re-sent every poll. Returns the resolved recipient list (so callers/tests can assert on it). """ if event not in SCADA_EVENTS: _log.error("notify_scada: unknown event %r", event) return [] if recipients is None: recipients = resolve_scada_recipients(SCADA_EVENT_PREF[event]) if not recipients: return [] rendered = render_scada_email(event, context) if not rendered: return [] subject, html = rendered send = sender or send_html_email def _run(): for recipient in recipients: send(recipient, subject, html) if background: threading.Thread(target=_run, daemon=True).start() else: _run() return recipients # --------------------------------------------------------------------------- # Repair Parts Manager part-lifecycle notifications (DVI-614) # --------------------------------------------------------------------------- # Per-user opt-in grid stored under USER_SETTINGS / .json: # settings["rpm_notifications"] = { # "mobilemachine": {"awaiting": bool, "approved": bool, "ordered": bool, # "delivered": bool, "stocked": bool}, # "preventive": {...same columns...}, # } # The same grid is surfaced in the RPM Settings > Notifications pane and mirrored # into User Settings > Notifications, so any user can subscribe to part events. RPM_NOTIFY_KEY = "rpm_notifications" EVENT_RPM_PART_AWAITING = "rpm.part.awaiting" EVENT_RPM_PART_APPROVED = "rpm.part.approved" EVENT_RPM_PART_ORDERED = "rpm.part.ordered" EVENT_RPM_PART_DELIVERED = "rpm.part.delivered" EVENT_RPM_PART_STOCKED = "rpm.part.stocked" RPM_PART_EVENTS = ( EVENT_RPM_PART_AWAITING, EVENT_RPM_PART_APPROVED, EVENT_RPM_PART_ORDERED, EVENT_RPM_PART_DELIVERED, EVENT_RPM_PART_STOCKED, ) # Grid column key per event (mirrors the RPM notifications grid columns). RPM_EVENT_COL = { EVENT_RPM_PART_AWAITING: "awaiting", EVENT_RPM_PART_APPROVED: "approved", EVENT_RPM_PART_ORDERED: "ordered", EVENT_RPM_PART_DELIVERED: "delivered", EVENT_RPM_PART_STOCKED: "stocked", } # (email subject phrase, body verb) per event. _RPM_EVENT_COPY = { EVENT_RPM_PART_AWAITING: ("part awaiting approval", "is awaiting approval"), EVENT_RPM_PART_APPROVED: ("part approved", "has been approved"), EVENT_RPM_PART_ORDERED: ("part ordered", "has been ordered"), EVENT_RPM_PART_DELIVERED: ("part delivered", "has been delivered"), EVENT_RPM_PART_STOCKED: ("part stocked", "has been stocked"), } _RPM_TYPE_LABEL = { "mobilemachine": "Mobile Machine", "preventive": "Preventative Maintenance", } def _rpm_policy_notify_emails(wor_type): """Company-level policy-enforced RPM recipients for ``wor_type`` (DVI-614 follow-up). A user on this list receives every RPM part event for the type, regardless of their per-user grid. Returns original-cased emails.""" if not wor_type: return [] try: data = json.loads(RPM_APPROVAL_FILE.read_text()) except (json.JSONDecodeError, OSError): return [] if not isinstance(data, dict): return [] pn = data.get("policy_notify") if not isinstance(pn, dict): return [] lst = pn.get(wor_type) return [str(e).strip() for e in lst if str(e).strip()] if isinstance(lst, list) else [] def resolve_rpm_recipients(wor_type, event): """De-duplicated emails of users to notify for (wor_type x event): the per-user opt-in grid (``rpm_notifications[wor_type][col]`` across every USER_SETTINGS_DIR/*.json) OR-merged with the company-level policy-enforced allowlist for the type (which receives all events). Returns original-cased email strings, order-stable, no duplicates.""" ordered, seen = [], set() col = RPM_EVENT_COL.get(event) if not (wor_type and col): return ordered def _add(email): norm = _normalize_email(email) if norm and norm not in seen: seen.add(norm) ordered.append(email.strip()) # Policy-enforced recipients first (all events for the type). for email in _rpm_policy_notify_emails(wor_type): _add(email) try: files = sorted(USER_SETTINGS_DIR.glob("*.json")) except OSError: return ordered for fp in files: try: settings = json.loads(fp.read_text()) except (json.JSONDecodeError, OSError): continue if not isinstance(settings, dict): continue prefs = settings.get(RPM_NOTIFY_KEY) if not isinstance(prefs, dict): continue type_prefs = prefs.get(wor_type) or {} if not (isinstance(type_prefs, dict) and type_prefs.get(col)): continue _add(settings.get("email") or fp.stem) return ordered def render_rpm_email(event, context): """Render (subject, html) for an RPM part event. None for an unknown event. Context keys: ``wor_type``, ``wo_label`` (WO number/label), ``wo_title`` (display title), ``part_name``, ``part_number`` (all strings, optional).""" copy = _RPM_EVENT_COPY.get(event) if not copy: return None subject_phrase, verb = copy wor_type = context.get("wor_type") or "" type_label = _RPM_TYPE_LABEL.get(wor_type, wor_type or "Work Order") part_name = context.get("part_name") or "(unnamed part)" part_number = context.get("part_number") or "" wo_label = context.get("wo_label") or "" wo_title = context.get("wo_title") or "" subject = "Repair Parts: {}".format(subject_phrase) if wo_label: subject += " ({})".format(wo_label) rows = [("Part", _esc(part_name))] if part_number: rows.append(("Part #", _esc(part_number))) rows.append(("Work Order type", _esc(type_label))) if wo_label: rows.append(("Work Order", _esc(wo_label))) if wo_title and wo_title != wo_label: rows.append(("Title", _esc(wo_title))) row_html = "".join( '{lbl}' '{val}' ''.format(lbl=lbl, val=val) for lbl, val in rows) detail_table = ( '' + row_html + '
') intro = "A repair part {verb} on a {type_label} work order.".format( verb=verb, type_label=_esc(type_label)) body_html = ( '

{intro}

{table}' .format(intro=intro, table=detail_table)) footer_html = ( '
' '

You are receiving this because you enabled Repair ' 'Parts notifications in your Togen settings.

') html = render_email_shell( title=_esc(subject_phrase.capitalize()), body_html=body_html, subtitle="Repair Parts Manager", footer_html=footer_html) return subject, html def notify_rpm(event, context, *, recipients=None, background=True, sender=None): """Send a Repair Parts Manager part-event email to opted-in users. Recipients default to everyone whose ``rpm_notifications`` grid enables (``context['wor_type']`` x event). Like :func:`notify_scada`, idempotency is the caller's responsibility — callers fire only on a state transition. Returns the resolved recipient list (so callers/tests can assert on it).""" if event not in RPM_PART_EVENTS: _log.error("notify_rpm: unknown event %r", event) return [] if recipients is None: recipients = resolve_rpm_recipients(context.get("wor_type"), event) if not recipients: return [] rendered = render_rpm_email(event, context) if not rendered: return [] subject, html = rendered send = sender or send_html_email def _run(): for recipient in recipients: send(recipient, subject, html) if background: threading.Thread(target=_run, daemon=True).start() else: _run() return recipients