inkreach-prefect/flows/monitor/alert_failure.py

"""企微机器人异常报警:任意 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(),
    })