From 1d743df4e3de1bb7a1f498fab16749f44579f408 Mon Sep 17 00:00:00 2001 From: ymq Date: Tue, 25 Aug 2026 18:22:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(approval):=20=E9=92=89=E9=92=89=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=E6=8E=A5=E6=88=90=E4=BA=A7=E7=BA=BF=E5=AE=A1=E6=89=B9?= =?UTF-8?q?=E5=85=B3=E5=8D=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 复用引擎既有 approval_gate 机制，不重造状态机： - 新增 dingtalk_approval.py：注册 step_type='dingtalk_approval'(interactive)， 进入步骤自动向钉钉发起审批；注册 biz_type='pipeline_approval' 回调钩子， 审批结果回来调 approval_approve/reject 推进产线(内部已 save_artifact+resume_task) - executor._handle_interactive_step 加外部通知钩子(按 step_type meta.notifier 分派)， 并把 tenant_id 从 _execute_step 传下来(原 step_info 里无此字段，会永远取空) - biz_id 编码 tenant_id:task_id:step_name，回调据此定位步骤 - 软依赖 dingdingflow：未加载时步骤仍挂起等人工审批，不推钉钉、不报错 - 通知失败只记日志不改步骤状态，避免外部系统故障导致步骤 FAILED --- pipeline_service/__init__.py | 5 + pipeline_service/dingtalk_approval.py | 149 ++++++++++++++++++++++++++ pipeline_service/executor.py | 21 +++- pipeline_service/init.py | 8 ++ 4 files changed, 181 insertions(+), 2 deletions(-) create mode 100644 pipeline_service/dingtalk_approval.py diff --git a/pipeline_service/__init__.py b/pipeline_service/__init__.py index e337807..7d3064b 100644 --- a/pipeline_service/__init__.py +++ b/pipeline_service/__init__.py @@ -18,6 +18,11 @@ from .init import ( from .handler import register_handler, list_handlers, register_default_handler from .step_registry import register_step_type, get_step_type, list_step_types, load_builtin_types from .human import human_complete, approval_approve, approval_reject, human_list +from .dingtalk_approval import ( + load_dingtalk_approval, + notify_dingtalk_approval, + on_dingtalk_approval_done, +) from .state import ( STATE_PENDING, STATE_RUNNING, STATE_COMPLETED, STATE_FAILED, STATE_SKIPPED, STATE_WAITING, STATE_REJECTED, diff --git a/pipeline_service/dingtalk_approval.py b/pipeline_service/dingtalk_approval.py new file mode 100644 index 0000000..f4b813f --- /dev/null +++ b/pipeline_service/dingtalk_approval.py @@ -0,0 +1,149 @@ +"""钉钉审批适配器 — 把 dingdingflow 模块接成产线的审批关卡。 + +设计：不重造审批状态机。产线引擎已内置 approval_gate 交互步骤 +（executor 遇到即置 WAITING 挂起，approval_approve/reject 会 +save_artifact + STATE_COMPLETED + resume_task 自动推进）。 +本适配器只做两件事： + +1. 注册 step_type='dingtalk_approval'（interactive），进入该步骤时 + 自动向钉钉发起审批实例； +2. 注册 dingdingflow 的 biz_type 回调钩子，钉钉审批结果回来时 + 调 approval_approve / approval_reject 推进产线。 + +biz_id 编码为 "{tenant_id}:{task_id}:{step_name}"，回调据此定位步骤。 + +宿主集成（在 load_pipeline_service() 之后调用）： + from pipeline_service.dingtalk_approval import load_dingtalk_approval + load_dingtalk_approval() + +未安装 dingdingflow 时本模块自动降级：step_type 仍注册（步骤照常挂起 +等人工在产线页面审批），只是不向钉钉推送。 +""" + +import json +import logging + +from ahserver.serverenv import ServerEnv + +from .step_registry import register_step_type +from .human import approval_approve, approval_reject + +logger = logging.getLogger("pipeline.dingtalk_approval") + +STEP_TYPE = "dingtalk_approval" +BIZ_TYPE = "pipeline_approval" +BIZ_ID_SEP = ":" + + +def _encode_biz_id(tenant_id, task_id, step_name): + """审批记录里存的业务 ID，回调靠它定位产线步骤。""" + return BIZ_ID_SEP.join([str(tenant_id), str(task_id), str(step_name)]) + + +def _decode_biz_id(biz_id): + """反解 biz_id。step_name 可能含分隔符，故只切前两段。""" + parts = str(biz_id).split(BIZ_ID_SEP, 2) + if len(parts) != 3: + raise ValueError("bad biz_id: %s" % biz_id) + return parts[0], parts[1], parts[2] + + +async def notify_dingtalk_approval(task_id, step_name, step_type, version, + input_data, step_config, tenant_id=""): + """进入审批步骤时向钉钉发起审批实例。 + + 由 executor 的交互步骤钩子调用（见 _INTERACTIVE_NOTIFIERS）。 + 失败不抛异常：钉钉推送失败不应让产线步骤变 FAILED， + 步骤仍停在 WAITING，可在产线页面人工审批兜底。 + """ + env = ServerEnv() + submit_approval = getattr(env, "submit_approval", None) + if submit_approval is None: + logger.warning("dingdingflow 未加载，步骤 %s 仅挂起等人工审批（未推钉钉）", step_name) + return None + + if not isinstance(step_config, dict): + step_config = {} + + title = step_config.get("approval_title") or "产线审批：%s" % step_name + applicant_id = (step_config.get("applicant_id") + or (input_data or {}).get("applicant_id", "") + or "") + biz_type = step_config.get("biz_type") or BIZ_TYPE + biz_id = _encode_biz_id(tenant_id, task_id, step_name) + + try: + result = await submit_approval(biz_type, biz_id, title, applicant_id, + str(tenant_id) or "0") + approval_id = (result or {}).get("approval_id", "") + logger.info("钉钉审批已发起: step=%s approval_id=%s biz_id=%s", + step_name, approval_id, biz_id) + return approval_id + except Exception as e: + logger.error("钉钉审批发起失败(步骤仍可人工审批): step=%s err=%s", step_name, e) + return None + + +async def on_dingtalk_approval_done(biz_id, status, approval_id, comment): + """钉钉审批结果回调 → 推进产线步骤。 + + dingdingflow 的 register_biz_handler 钩子签名。 + status ∈ ('approved', 'rejected', 'cancelled') + """ + try: + tenant_id, task_id, step_name = _decode_biz_id(biz_id) + except ValueError as e: + logger.error("回调 biz_id 解析失败: %s", e) + return + + reviewer_id = "dingtalk:%s" % (approval_id or "") + + if status == "approved": + raw = await approval_approve(tenant_id, task_id, step_name, + reviewer_id, comment) + elif status in ("rejected", "cancelled"): + # cancelled 也走 reject：产线不能停在 WAITING 永久卡死 + raw = await approval_reject(tenant_id, task_id, step_name, + reviewer_id, comment) + else: + logger.warning("未知审批状态，忽略: %s", status) + return + + try: + parsed = json.loads(raw) if isinstance(raw, str) else (raw or {}) + except (ValueError, TypeError): + parsed = {} + + if parsed.get("success"): + logger.info("产线步骤已推进: task=%s step=%s status=%s", + task_id, step_name, status) + else: + logger.error("产线步骤推进失败: task=%s step=%s status=%s msg=%s", + task_id, step_name, status, parsed.get("message")) + + +def load_dingtalk_approval(): + """注册钉钉审批步骤类型 + 回调钩子。""" + env = ServerEnv() + + # 1. 注册为交互步骤：executor 遇到即置 WAITING 挂起，不需要 handler + register_step_type(STEP_TYPE, { + "display_name": "钉钉审批", + "category": "interactive", + "is_interactive": True, + "description": "向钉钉发起审批，审批通过后产线继续，驳回则中止", + "notifier": "notify_dingtalk_approval", + }) + + # 2. 暴露通知函数（executor 交互步骤钩子 / 手工重推都能用） + env.notify_dingtalk_approval = notify_dingtalk_approval + + # 3. 注册审批结果回调钩子到 dingdingflow + register_biz_handler = getattr(env, "register_biz_handler", None) + if register_biz_handler: + register_biz_handler(BIZ_TYPE, on_dingtalk_approval_done) + logger.info("dingtalk_approval loaded — biz_type=%s 钩子已注册", BIZ_TYPE) + else: + logger.warning("dingdingflow 未加载：step_type=%s 已注册但审批结果无法自动回流，" + "请先 load_dingdingflow()", STEP_TYPE) + return True diff --git a/pipeline_service/executor.py b/pipeline_service/executor.py index b845a31..0dbdfae 100644 --- a/pipeline_service/executor.py +++ b/pipeline_service/executor.py @@ -180,7 +180,8 @@ async def _execute_step(task_id: str, step_name: str, step_graph: dict, task_inf # Check if this is an interactive step type if is_interactive(step_type): await _handle_interactive_step( - task_id, step_name, step_type, version, input_data, step_info + task_id, step_name, step_type, version, input_data, step_info, + tenant_id=tenant_id ) return @@ -219,7 +220,7 @@ async def _execute_step(task_id: str, step_name: str, step_graph: dict, task_inf await update_step_state(task_id, step_name, STATE_FAILED, error_msg) -async def _handle_interactive_step(task_id, step_name, step_type, version, input_data, step_info): +async def _handle_interactive_step(task_id, step_name, step_type, version, input_data, step_info, tenant_id=""): """Handle an interactive step — create human task record and enter WAITING.""" # Get step type metadata meta = get_step_type(step_type) or {} @@ -253,6 +254,22 @@ async def _handle_interactive_step(task_id, step_name, step_type, version, input await update_step_state(task_id, step_name, STATE_WAITING) logger.info(f"Step {step_name} waiting for human input (type={step_type}, role={assignee_role})") + # 外部通知钩子：交互步骤挂起后，按 step_type 推送到外部审批系统（如钉钉）。 + # 钩子由适配器注册到 ServerEnv（见 dingtalk_approval.load_dingtalk_approval）。 + # 失败只记日志，不改步骤状态——外部系统不可用时仍可在产线页面人工审批。 + notifier_name = (meta or {}).get("notifier") + if notifier_name: + try: + from ahserver.serverenv import ServerEnv + notifier = getattr(ServerEnv(), notifier_name, None) + if notifier: + await notifier(task_id, step_name, step_type, version, + input_data, step_config, tenant_id=tenant_id) + else: + logger.warning(f"notifier {notifier_name} not registered in ServerEnv") + except Exception as e: + logger.error(f"notifier {notifier_name} failed for step {step_name}: {e}") + async def _gather_inputs(task_id: str, version: int, deps: list) -> dict: """Gather input data from dependency step outputs.""" diff --git a/pipeline_service/init.py b/pipeline_service/init.py index bb2e4e8..135a973 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -532,6 +532,14 @@ def load_pipeline_service(): # Load built-in interactive step types load_builtin_types() + # 钉钉审批适配器：注册 step_type='dingtalk_approval' + 审批结果回调钩子 + # 软依赖 dingdingflow：未安装时步骤仍可挂起等人工审批，只是不推钉钉 + try: + from .dingtalk_approval import load_dingtalk_approval + load_dingtalk_approval() + except Exception as e: + debug(f'dingtalk_approval 加载失败(不影响产线其他功能): {e}') + # Register workspace file-management functions (shared by workspace_*.dspy) from .workspace import load_workspace load_workspace()