"""Validated, revision-safe notification routing for workflow outputs.""" from __future__ import annotations import hashlib import hmac import json import os import re import unicodedata import uuid from dataclasses import dataclass from pathlib import Path from types import MappingProxyType from typing import Any, Mapping, Sequence from gyxx_flow.core.artifacts import atomic_write_json from gyxx_flow.core.config import Settings from gyxx_flow.core.locks import LockManager, ResourceBusyError NOTIFICATION_ROUTING_SCHEMA_VERSION = 1 NOTIFICATION_ROUTING_RELATIVE_PATH = Path("state") / "notifications" / "routing.json" NOTIFICATION_ROUTING_CONFIG_NAME = "notification-routing.json" ANALYZER_NOTIFICATION_APP_PROFILE = "hermes-analyzer" ANALYZER_NOTIFICATION_APP_ID = "cli_aa8c4fc918b85cce" ANALYZER_HERMES_PROFILE = "data-analyzer" ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID = ( "ou_7ad5fc8012e2f741afc5346e05ffd447" ) GYXX_NOTIFICATION_ROUTE_MODE = "GYXX_NOTIFICATION_ROUTE_MODE" GYXX_NOTIFICATION_APP_PROFILE = "GYXX_NOTIFICATION_APP_PROFILE" GYXX_NOTIFICATION_HERMES_PROFILE = "GYXX_NOTIFICATION_HERMES_PROFILE" GYXX_NOTIFICATION_RECIPIENTS_JSON = "GYXX_NOTIFICATION_RECIPIENTS_JSON" GYXX_NOTIFICATION_ROUTE_REVISION = "GYXX_NOTIFICATION_ROUTE_REVISION" GYXX_NOTIFICATION_ROUTE_WORKFLOW_ID = "GYXX_NOTIFICATION_ROUTE_WORKFLOW_ID" _ROUTE_MODES = frozenset({"legacy", "enabled", "disabled"}) _SAFE_ID = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$") _SAFE_OPEN_ID = re.compile(r"^ou_[A-Za-z0-9_-]{8,128}$") _SAFE_APP_ID = re.compile(r"^cli_[A-Za-z0-9_-]{6,128}$") _SAFE_SOURCE = frozenset({"provided", "repository", "mobile_lookup", "manual"}) _REVISION = re.compile(r"^[0-9a-f]{64}$") MAX_NOTIFICATION_ROUTING_BYTES = 256 * 1024 MAX_NOTIFICATION_PEOPLE = 500 MAX_NOTIFICATION_APP_PROFILES = 16 MAX_NOTIFICATION_RECIPIENTS = 100 MAX_NOTIFICATION_BINDINGS_PER_PERSON = 16 NOTIFICATION_CAPABILITIES: Mapping[str, str] = MappingProxyType( { "content.marketing_report.daily": "正式执行且启用发送时投递最终营销日报。", "content.relogin.weekly": "刷新期间投递各平台扫码二维码，并在平台失败时投递失败汇总。", "product.alert.daily": "命中销量下滑规则且尚未重复通知时投递告警。", "product.market_rank": "存在本次新生成的平台报告链接且尚未重复通知时投递。", "product.online_stock_alert.daily": "线上商品当前库存低于 3 时向配置接收人发送库存预警。", "supply.purchase_confirmation.daily": "采购确认结果生成完成后投递最终业务通知。", "supply.replenishment.weekly": "补货处理完成后投递最终结果，包括无新增补货项的结果。", "supply.replenishment_alert.daily": "存在命中库存阈值的 SKU 时投递预警。", } ) class NotificationRoutingError(ValueError): """Raised when notification routing is missing, corrupt, or unsafe.""" class NotificationRoutingConflictError(NotificationRoutingError): """Raised when a routing update uses a stale revision or cannot lock.""" class NotificationRoutingPreconditionError(NotificationRoutingError): """Raised when a routing mutation omits its expected revision.""" @dataclass(frozen=True, slots=True) class ResolvedNotificationRoute: """One immutable workflow notification decision.""" configured: bool enabled: bool app_profile: str | None open_ids: tuple[str, ...] revision: str | None = None class NotificationRoutingStore: """Load the project baseline and atomically persist runtime routing state.""" def __init__(self, settings: Settings) -> None: self.settings = settings self.baseline_path = ( settings.project_root / "config" / NOTIFICATION_ROUTING_CONFIG_NAME ) self.path = settings.data_root / NOTIFICATION_ROUTING_RELATIVE_PATH self._locks = LockManager(settings.data_root / "state" / "locks") def snapshot(self) -> tuple[dict[str, Any], str]: """Return one validated snapshot and the digest of the same raw bytes.""" source = self.path if self.path.is_file() else self.baseline_path try: raw = source.read_bytes() except OSError as exc: raise NotificationRoutingError( f"cannot read notification routing: {source.name}" ) from exc if len(raw) > MAX_NOTIFICATION_ROUTING_BYTES: raise NotificationRoutingError("notification routing file is too large") revision = hashlib.sha256(raw).hexdigest() try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise NotificationRoutingError("notification routing JSON is invalid") from exc return _normalize_payload(payload), revision def update( self, payload: Mapping[str, Any], *, expected_revision: str | None, ) -> tuple[dict[str, Any], str]: """Validate and atomically replace runtime state under optimistic locking.""" if expected_revision is None or not expected_revision.strip(): raise NotificationRoutingPreconditionError( "notification routing revision is required" ) expected = expected_revision.strip().strip('"') normalized = _normalize_payload(payload) _require_payload_size(normalized) owner = f"notification-{uuid.uuid4().hex}" try: with self._locks.acquire("config:notification-routing", owner=owner): _current, current_revision = self.snapshot() if not hmac.compare_digest(current_revision, expected): raise NotificationRoutingConflictError( "notification routing was updated; refresh and retry" ) atomic_write_json(self.path, normalized) updated, revision = self.snapshot() return updated, revision except ResourceBusyError as exc: raise NotificationRoutingConflictError( "notification routing is being updated" ) from exc def resolve_notification_route( workflow_id: str, defaults: Sequence[str], environment: Mapping[str, str] | None = None, ) -> ResolvedNotificationRoute: """Resolve a route, preferring the immutable environment snapshot.""" _require_safe_id(workflow_id, field="workflow id") values = os.environ if environment is None else environment mode = values.get(GYXX_NOTIFICATION_ROUTE_MODE) if mode is not None: snapshot_workflow = values.get(GYXX_NOTIFICATION_ROUTE_WORKFLOW_ID, "").strip() if snapshot_workflow != workflow_id: raise NotificationRoutingError( "notification route snapshot belongs to another workflow" ) resolved = _route_from_environment(values, defaults) elif workflow_id not in NOTIFICATION_CAPABILITIES: resolved = ResolvedNotificationRoute( configured=False, enabled=True, app_profile=None, open_ids=_normalize_legacy_defaults(defaults), ) else: settings = Settings.from_env( project_root=values.get("GYXX_PROJECT_ROOT") or None, env=values, ) payload, revision = NotificationRoutingStore(settings).snapshot() route = payload["routes"].get(workflow_id) if route is None: resolved = ResolvedNotificationRoute( configured=False, enabled=True, app_profile=None, open_ids=_normalize_legacy_defaults(defaults), revision=revision, ) elif not route["enabled"]: resolved = ResolvedNotificationRoute( configured=True, enabled=False, app_profile=route["app_profile"], open_ids=(), revision=revision, ) else: open_ids = tuple( payload["people"][person_id]["bindings"][route["app_profile"]][ "open_id" ] for person_id in route["person_ids"] ) resolved = ResolvedNotificationRoute( configured=True, enabled=True, app_profile=route["app_profile"], open_ids=open_ids, revision=revision, ) return _apply_acceptance_recipient(workflow_id, resolved, values) def notification_environment_for_workflow( workflow_id: str, *, project_root: Path | str, data_root: Path | str, environment: Mapping[str, str] | None = None, ) -> dict[str, str]: """Build the stable routing environment injected into one workflow run.""" values = dict(os.environ if environment is None else environment) values["GYXX_PROJECT_ROOT"] = str(Path(project_root).expanduser().resolve()) values["GYXX_DATA_ROOT"] = str(Path(data_root).expanduser().resolve()) resolved = resolve_notification_route(workflow_id, (), values) if resolved.configured: mode = "enabled" if resolved.enabled else "disabled" else: mode = "legacy" return { GYXX_NOTIFICATION_ROUTE_MODE: mode, GYXX_NOTIFICATION_APP_PROFILE: resolved.app_profile or "", GYXX_NOTIFICATION_HERMES_PROFILE: ( ANALYZER_HERMES_PROFILE if resolved.configured else "" ), GYXX_NOTIFICATION_RECIPIENTS_JSON: json.dumps( resolved.open_ids, ensure_ascii=True, separators=(",", ":"), ), GYXX_NOTIFICATION_ROUTE_REVISION: resolved.revision or "", GYXX_NOTIFICATION_ROUTE_WORKFLOW_ID: workflow_id, } def _route_from_environment( values: Mapping[str, str], defaults: Sequence[str], ) -> ResolvedNotificationRoute: mode = values.get(GYXX_NOTIFICATION_ROUTE_MODE, "").strip() if mode not in _ROUTE_MODES: raise NotificationRoutingError("notification route snapshot mode is invalid") revision = values.get(GYXX_NOTIFICATION_ROUTE_REVISION, "").strip() or None if revision is not None and not _REVISION.fullmatch(revision): raise NotificationRoutingError("notification route snapshot revision is invalid") app_profile = values.get(GYXX_NOTIFICATION_APP_PROFILE, "").strip() or None hermes_profile = ( values.get(GYXX_NOTIFICATION_HERMES_PROFILE, "").strip() or None ) raw_recipients = values.get(GYXX_NOTIFICATION_RECIPIENTS_JSON, "[]") try: recipients = json.loads(raw_recipients) except json.JSONDecodeError as exc: raise NotificationRoutingError( "notification route snapshot recipients are invalid" ) from exc if not isinstance(recipients, list) or not all( isinstance(item, str) and _SAFE_OPEN_ID.fullmatch(item) for item in recipients ): raise NotificationRoutingError( "notification route snapshot recipients are invalid" ) open_ids = tuple(dict.fromkeys(recipients)) if len(open_ids) > MAX_NOTIFICATION_RECIPIENTS: raise NotificationRoutingError("notification route has too many recipients") if mode == "legacy": if app_profile is not None or hermes_profile is not None or open_ids: raise NotificationRoutingError("legacy notification snapshot is inconsistent") return ResolvedNotificationRoute( configured=False, enabled=True, app_profile=None, open_ids=_normalize_legacy_defaults(defaults), revision=revision, ) if app_profile is None: raise NotificationRoutingError("notification app profile is required") if hermes_profile != ANALYZER_HERMES_PROFILE: raise NotificationRoutingError( "notification Hermes profile snapshot is invalid" ) _require_safe_id(app_profile, field="app profile") if mode == "disabled": if open_ids: raise NotificationRoutingError("disabled notification snapshot has recipients") return ResolvedNotificationRoute(True, False, app_profile, (), revision) if not open_ids: raise NotificationRoutingError("enabled notification snapshot has no recipients") return ResolvedNotificationRoute(True, True, app_profile, open_ids, revision) def _apply_acceptance_recipient( workflow_id: str, resolved: ResolvedNotificationRoute, values: Mapping[str, str], ) -> ResolvedNotificationRoute: if not resolved.enabled or not _read_bool(values.get("GYXX_WORKFLOW_ACCEPTANCE")): return resolved if workflow_id in NOTIFICATION_CAPABILITIES: # These capabilities are delivered by the analyzer app. The global # acceptance recipient belongs to the collector app and cannot be used # with analyzer-scoped OpenIDs. recipient = ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID else: recipient = values.get("GYXX_NOTIFICATION_RECIPIENT_OPEN_ID", "").strip() if not recipient: return resolved if not _SAFE_OPEN_ID.fullmatch(recipient): raise NotificationRoutingError("acceptance notification recipient is invalid") return ResolvedNotificationRoute( configured=resolved.configured, enabled=True, app_profile=resolved.app_profile, open_ids=(recipient,), revision=resolved.revision, ) def _normalize_payload(payload: object) -> dict[str, Any]: if not isinstance(payload, Mapping): raise NotificationRoutingError("notification routing must be an object") required = { "schema_version", "app_profile", "app_profiles", "people", "capabilities", "routes", } if set(payload) != required: raise NotificationRoutingError("notification routing fields are invalid") schema_version = payload.get("schema_version") if ( isinstance(schema_version, bool) or schema_version != NOTIFICATION_ROUTING_SCHEMA_VERSION ): raise NotificationRoutingError("unsupported notification routing schema") app_profiles = _normalize_app_profiles(payload.get("app_profiles")) app_profile = payload.get("app_profile") if not isinstance(app_profile, str) or app_profile not in app_profiles: raise NotificationRoutingError("default notification app profile is invalid") _require_analyzer_delivery_profile(app_profile, app_profiles) capabilities = _normalize_capabilities(payload.get("capabilities")) people = _normalize_people(payload.get("people"), app_profiles) routes = _normalize_routes( payload.get("routes"), app_profiles=app_profiles, people=people, ) normalized = { "schema_version": NOTIFICATION_ROUTING_SCHEMA_VERSION, "app_profile": app_profile, "app_profiles": app_profiles, "people": people, "capabilities": capabilities, "routes": routes, } _require_payload_size(normalized) return normalized def _normalize_app_profiles(value: object) -> dict[str, dict[str, str]]: if not isinstance(value, Mapping) or not value: raise NotificationRoutingError("notification app profiles are invalid") if len(value) > MAX_NOTIFICATION_APP_PROFILES: raise NotificationRoutingError("too many notification app profiles") profiles: dict[str, dict[str, str]] = {} for profile_id, raw in value.items(): _require_safe_id(profile_id, field="app profile") if not isinstance(raw, Mapping) or set(raw) != { "label", "app_id", "hermes_profile", }: raise NotificationRoutingError(f"invalid app profile: {profile_id}") app_id = raw.get("app_id") if not isinstance(app_id, str) or not _SAFE_APP_ID.fullmatch(app_id): raise NotificationRoutingError(f"invalid app id for profile: {profile_id}") hermes_profile = raw.get("hermes_profile") if not isinstance(hermes_profile, str): raise NotificationRoutingError( f"invalid Hermes profile for app profile: {profile_id}" ) _require_safe_id(hermes_profile, field="Hermes profile") profiles[profile_id] = { "label": _normalize_text(raw.get("label"), field="app profile label"), "app_id": app_id, "hermes_profile": hermes_profile, } return profiles def _normalize_capabilities(value: object) -> dict[str, dict[str, str]]: if not isinstance(value, Mapping) or set(value) != set(NOTIFICATION_CAPABILITIES): raise NotificationRoutingError("notification capabilities do not match the catalog") capabilities: dict[str, dict[str, str]] = {} for workflow_id, expected_condition in NOTIFICATION_CAPABILITIES.items(): raw = value.get(workflow_id) if not isinstance(raw, Mapping) or set(raw) != {"condition"}: raise NotificationRoutingError( f"invalid notification capability: {workflow_id}" ) if raw.get("condition") != expected_condition: raise NotificationRoutingError( f"notification capability condition changed: {workflow_id}" ) capabilities[workflow_id] = {"condition": expected_condition} return capabilities def _normalize_people( value: object, app_profiles: Mapping[str, object], ) -> dict[str, dict[str, Any]]: if not isinstance(value, Mapping): raise NotificationRoutingError("notification people must be an object") if len(value) > MAX_NOTIFICATION_PEOPLE: raise NotificationRoutingError("too many notification people") people: dict[str, dict[str, Any]] = {} identities: dict[tuple[str, str], str] = {} for person_id, raw in value.items(): _require_safe_id(person_id, field="person id") if not isinstance(raw, Mapping) or set(raw) != {"name", "mobile", "bindings"}: raise NotificationRoutingError(f"invalid person: {person_id}") bindings_value = raw.get("bindings") if not isinstance(bindings_value, Mapping): raise NotificationRoutingError(f"invalid bindings for person: {person_id}") if len(bindings_value) > MAX_NOTIFICATION_BINDINGS_PER_PERSON: raise NotificationRoutingError(f"too many bindings for person: {person_id}") bindings: dict[str, dict[str, Any]] = {} for profile_id, binding in bindings_value.items(): _require_safe_id(profile_id, field="app profile") if profile_id not in app_profiles: raise NotificationRoutingError( f"unknown app profile for person {person_id}: {profile_id}" ) if not isinstance(binding, Mapping) or set(binding) != { "open_id", "verified", "source", }: raise NotificationRoutingError( f"invalid binding for person {person_id}: {profile_id}" ) open_id = binding.get("open_id") if not isinstance(open_id, str) or not _SAFE_OPEN_ID.fullmatch(open_id): raise NotificationRoutingError( f"invalid open_id for person {person_id}: {profile_id}" ) verified = binding.get("verified") source = binding.get("source") if ( not isinstance(verified, bool) or not isinstance(source, str) or source not in _SAFE_SOURCE ): raise NotificationRoutingError( f"invalid binding metadata for person {person_id}: {profile_id}" ) identity_key = (profile_id, open_id) previous = identities.get(identity_key) if previous is not None and previous != person_id: raise NotificationRoutingError( f"open_id is bound to multiple people in {profile_id}" ) identities[identity_key] = person_id bindings[profile_id] = { "open_id": open_id, "verified": verified, "source": source, } mobile = _normalize_mobile(raw.get("mobile")) if not bindings and mobile is None: raise NotificationRoutingError( f"person requires a mobile or app binding: {person_id}" ) people[person_id] = { "name": _normalize_text(raw.get("name"), field="person name"), "mobile": mobile, "bindings": bindings, } return people def _normalize_routes( value: object, *, app_profiles: Mapping[str, object], people: Mapping[str, Mapping[str, Any]], ) -> dict[str, dict[str, Any]]: if not isinstance(value, Mapping): raise NotificationRoutingError("notification routes must be an object") routes: dict[str, dict[str, Any]] = {} for workflow_id, raw in value.items(): _require_safe_id(workflow_id, field="workflow id") if workflow_id not in NOTIFICATION_CAPABILITIES: raise NotificationRoutingError( f"workflow does not support notification routing: {workflow_id}" ) if not isinstance(raw, Mapping) or set(raw) != { "enabled", "app_profile", "person_ids", }: raise NotificationRoutingError(f"invalid notification route: {workflow_id}") enabled = raw.get("enabled") profile_id = raw.get("app_profile") person_ids_value = raw.get("person_ids") if not isinstance(enabled, bool): raise NotificationRoutingError(f"invalid enabled flag: {workflow_id}") if not isinstance(profile_id, str) or profile_id not in app_profiles: raise NotificationRoutingError(f"invalid app profile: {workflow_id}") _require_analyzer_delivery_profile(profile_id, app_profiles) if not isinstance(person_ids_value, list) or not all( isinstance(person_id, str) for person_id in person_ids_value ): raise NotificationRoutingError(f"invalid person list: {workflow_id}") person_ids = list(dict.fromkeys(person_ids_value)) if len(person_ids) > MAX_NOTIFICATION_RECIPIENTS: raise NotificationRoutingError(f"too many route recipients: {workflow_id}") if enabled and not person_ids: raise NotificationRoutingError( f"enabled notification route has no recipients: {workflow_id}" ) for person_id in person_ids: _require_safe_id(person_id, field="person id") person = people.get(person_id) if person is None: raise NotificationRoutingError( f"notification route references unknown person: {person_id}" ) if enabled and profile_id not in person["bindings"]: raise NotificationRoutingError( f"person {person_id} has no binding for {profile_id}" ) routes[workflow_id] = { "enabled": enabled, "app_profile": profile_id, "person_ids": person_ids, } return routes def _normalize_legacy_defaults(defaults: Sequence[str]) -> tuple[str, ...]: normalized: list[str] = [] for value in defaults: if not isinstance(value, str): raise NotificationRoutingError("legacy notification recipient must be a string") recipient = value.strip() if not recipient or len(recipient) > 256 or _has_unsafe_characters(recipient): raise NotificationRoutingError("legacy notification recipient is unsafe") if recipient not in normalized: normalized.append(recipient) if len(normalized) > MAX_NOTIFICATION_RECIPIENTS: raise NotificationRoutingError("too many legacy notification recipients") return tuple(normalized) def _normalize_mobile(value: object) -> str | None: if value is None: return None if not isinstance(value, str): raise NotificationRoutingError("person mobile must be a string or null") mobile = re.sub(r"[\s()-]", "", value) if re.fullmatch(r"1\d{10}", mobile): mobile = f"+86{mobile}" elif re.fullmatch(r"86\d{11}", mobile): mobile = f"+{mobile}" if not re.fullmatch(r"\+[1-9]\d{6,14}", mobile): raise NotificationRoutingError("person mobile must use E.164 format") return mobile def _normalize_text(value: object, *, field: str) -> str: if not isinstance(value, str): raise NotificationRoutingError(f"{field} must be a string") normalized = unicodedata.normalize("NFC", value).strip() if ( not normalized or len(normalized) > 64 or len(normalized.encode("utf-8")) > 256 or _has_unsafe_characters(normalized) ): raise NotificationRoutingError(f"{field} is invalid") return normalized def _has_unsafe_characters(value: str) -> bool: return any( unicodedata.category(character) in {"Cc", "Cf", "Cs"} for character in value ) def _require_safe_id(value: object, *, field: str) -> None: if not isinstance(value, str) or not _SAFE_ID.fullmatch(value): raise NotificationRoutingError(f"invalid {field}: {value!r}") def _require_analyzer_delivery_profile( profile_id: str, app_profiles: Mapping[str, Mapping[str, str]], ) -> None: profile = app_profiles.get(profile_id) if profile is None or profile.get("hermes_profile") != ANALYZER_HERMES_PROFILE: raise NotificationRoutingError( "business notification routes must use an analyzer Hermes app profile" ) def _require_payload_size(payload: Mapping[str, Any]) -> None: encoded = json.dumps( payload, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") if len(encoded) > MAX_NOTIFICATION_ROUTING_BYTES: raise NotificationRoutingError("notification routing payload is too large") def _read_bool(value: str | None) -> bool: return bool(value and value.strip().casefold() in {"1", "true", "yes", "on"}) __all__ = [ "ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID", "ANALYZER_HERMES_PROFILE", "ANALYZER_NOTIFICATION_APP_ID", "ANALYZER_NOTIFICATION_APP_PROFILE", "GYXX_NOTIFICATION_APP_PROFILE", "GYXX_NOTIFICATION_HERMES_PROFILE", "GYXX_NOTIFICATION_RECIPIENTS_JSON", "GYXX_NOTIFICATION_ROUTE_MODE", "GYXX_NOTIFICATION_ROUTE_REVISION", "GYXX_NOTIFICATION_ROUTE_WORKFLOW_ID", "MAX_NOTIFICATION_ROUTING_BYTES", "NOTIFICATION_CAPABILITIES", "NOTIFICATION_ROUTING_SCHEMA_VERSION", "NotificationRoutingConflictError", "NotificationRoutingError", "NotificationRoutingPreconditionError", "NotificationRoutingStore", "ResolvedNotificationRoute", "notification_environment_for_workflow", "resolve_notification_route", ]