"""企微机器人异常报警:任意 flow run 失败/崩溃时由 Prefect 事件触发器自动调用。 对应 wecom-alert deployment(prefect.yaml 里的 triggers 声明): - match: prefect.resource.id = prefect.flow-run.* - expect: prefect.flow-run.Failed / Crashed - parameters.event = "{{ event }}" flow 体全容错——本 flow 自身永不进入 Failed 状态,避免自我触发循环。 event 可能以 dict 或 str(jinja 渲染结果)传入,均做防御性解析; 解析出 flow run id 后再查 Prefect API 补全权威信息(名称/错误消息/时间)。 """ from __future__ import annotations import json import os import re from datetime import datetime from typing import Any import httpx from prefect import flow from flows.shared import send_markdown UI_BASE = os.environ.get("PREFECT_UI_URL", "http://192.168.124.92:4200") def _parse_event(event: Any) -> dict: if isinstance(event, dict): return event if isinstance(event, str) and event.strip(): # jinja "{{ event }}" 可能渲染成 JSON 串或 Python repr,统一抠除关键字段 rid = re.search(r"prefect\.flow-run\.[0-9a-fA-F-]{36}", event) ev = re.search(r"prefect\.flow-run\.(?:Failed|Crashed)", event) name = re.search(r"prefect\.resource\.name['\"]?\s*[:=]\s*['\"]([^'\"]+)", event) occurred = re.search(r"occurred['\"]?\s*[:=]\s*['\"]([^'\"]+)", event) return { "event": ev.group(0) if ev else "", "resource": { "prefect.resource.id": rid.group(0) if rid else "", "prefect.resource.name": name.group(1) if name else "", }, "occurred": occurred.group(1) if occurred else "", } return {} def _flow_run_id(event: dict) -> str | None: rid = (event.get("resource") or {}).get("prefect.resource.id", "") m = re.match(r"^prefect\.flow-run\.(.+)$", rid) return m.group(1) if m else None def _fetch_run_detail(run_id: str) -> dict: """查 Prefect API 补全 flow run 权威信息(含 flow 名),失败返回空 dict。""" api_url = os.environ.get("PREFECT_API_URL", "http://localhost:4200/api") try: resp = httpx.get(f"{api_url}/flow_runs/{run_id}", timeout=10) resp.raise_for_status() detail = resp.json() flow_id = detail.get("flow_id") if flow_id: try: fresp = httpx.get(f"{api_url}/flows/{flow_id}", timeout=10) fresp.raise_for_status() detail["_flowName"] = fresp.json().get("name", "") except Exception: # noqa: BLE001 detail["_flowName"] = "" return detail except Exception as err: # noqa: BLE001 print(f"[wecom-alert] 查询 flow run {run_id} 失败: {err}") return {} def _fmt_ts(value: Any) -> str: if not value: return "—" try: return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone().strftime("%m-%d %H:%M") except (ValueError, TypeError): return str(value)[:19] @flow(name="wecom-alert", log_prints=True) def wecom_alert_failure(event: dict | str | None = None) -> dict: """Prefect 事件触发:flow run 失败/崩溃 → 发企微群报警。永不抛异常。""" result = {"sent": False, "flowName": None, "runId": None} try: ev = _parse_event(event) event_name = ev.get("event", "prefect.flow-run.?") resource_name = (ev.get("resource") or {}).get("prefect.resource.name", "") run_id = _flow_run_id(ev) flow_name, run_name, state_msg, occurred = "", resource_name, "", ev.get("occurred", "") if run_id: detail = _fetch_run_detail(run_id) flow_name = detail.get("_flowName") or resource_name run_name = detail.get("name", "") state = detail.get("state") or {} state_msg = (state.get("message") or "").strip() ended = detail.get("end_time") or detail.get("expected_start_time") occurred = occurred or ended lines = [f"## 流程失败:{flow_name or resource_name or '未知'}"] if run_name: lines.append(f"Run {run_name} · {event_name.split('.')[-1]} · {_fmt_ts(occurred)}") if state_msg: lines.append(f'错误:{state_msg[:150].replace(chr(10), " ")}') if run_id: lines.append(f"[查看详情]({UI_BASE}/runs/flow-run/{run_id})") else: raw = json.dumps(ev, ensure_ascii=False)[:300] if ev else "(事件内容解析失败)" lines.append(f"原始事件:{raw}") result["sent"] = send_markdown("\n".join(lines)) result.update({"flowName": flow_name or None, "runId": run_id}) if not result["sent"]: print(f"[wecom-alert] 报警内容(发送失败,留档): {' '.join(lines)[:400]}") print(f"[wecom-alert] 已处理失败报警: {flow_name or resource_name} ({event_name})") except Exception as err: # noqa: BLE001 print(f"[wecom-alert] 报警流程异常(已吞掉,避免自我触发): {err}") return result if __name__ == "__main__": # 本地手测:模拟一条 Failed 事件 wecom_alert_failure({ "event": "prefect.flow-run.Failed", "resource": { "prefect.resource.id": "prefect.flow-run.00000000-0000-0000-0000-000000000000", "prefect.resource.name": "riin-order-sync/manual", }, "occurred": datetime.now().astimezone().isoformat(), })