shanxia2.0/backend/app/services/workflow_callback.py

"""工作流回调服务 —— 流程状态变更时通知业务系统"""
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional

import requests

if TYPE_CHECKING:
    from app.models.models import WfProcessInstance

logger = logging.getLogger("sqxp.workflow_callback")

# ---------------------------------------------------------------------------
# 本地回调注册表：内部路径（以 / 开头）直接在进程内调用，不走 HTTP
# ---------------------------------------------------------------------------

_LocalCallbackFn = Callable[[Any, str, str], None]  # (db, business_id, event) -> None

_LOCAL_CALLBACKS: Dict[str, _LocalCallbackFn] = {}


def register_local_callback(path: str, fn: _LocalCallbackFn) -> None:
    """注册本地回调函数。当 callback_url 以 path 开头时直接调用 fn。"""
    _LOCAL_CALLBACKS[path] = fn


def _dispatch_local(db: Any, instance: WfProcessInstance, event: str) -> bool:
    """尝试本地直调。返回 True 表示已处理，False 表示无匹配的本地回调。"""
    url = instance.callback_url or ""
    if not url.startswith("/"):
        return False
    for prefix, fn in _LOCAL_CALLBACKS.items():
        if url.startswith(prefix):
            try:
                fn(db, instance.business_id or "", event)
                logger.info(
                    "工作流本地回调成功: event=%s instance=%s path=%s",
                    event,
                    instance.id,
                    url,
                )
                return True
            except Exception:
                logger.exception(
                    "工作流本地回调失败: event=%s instance=%s path=%s",
                    event,
                    instance.id,
                    url,
                )
                return True  # 已匹配但执行失败，不再走 HTTP
    return False


# ---------------------------------------------------------------------------
# trigger_callback（对外入口）
# ---------------------------------------------------------------------------


def trigger_callback(db: Any, instance: WfProcessInstance, event: str) -> None:
    """流程实例状态变更后，向业务系统发送回调通知。

    优先级：
    1. 如果 callback_url 是内部路径（以 / 开头），优先尝试本地直调
    2. 本地无匹配时，回退到 HTTP 回调（需完整 URL）

    Args:
        db: 数据库会话。
        instance: 流程实例 ORM 对象。
        event: 事件类型，如 approved / rejected / completed。
    """
    if not instance.callback_url:
        return

    # 尝试本地直调
    if _dispatch_local(db, instance, event):
        return

    # 回退到 HTTP 回调
    payload: Dict[str, Any] = {
        "event": event,
        "instanceId": instance.id,
        "businessType": instance.business_type,
        "businessId": instance.business_id,
        "status": instance.status,
    }
    try:
        resp = requests.post(
            instance.callback_url,
            json=payload,
            timeout=10,
            headers={"Content-Type": "application/json"},
        )
        logger.info(
            "工作流HTTP回调成功: event=%s instance=%s status=%d",
            event,
            instance.id,
            resp.status_code,
        )
    except Exception:
        logger.exception(
            "工作流HTTP回调失败: event=%s instance=%s url=%s",
            event,
            instance.id,
            instance.callback_url,
        )