feat(approval): 钉钉审批接成产线审批关卡
复用引擎既有 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
This commit is contained in:
ymq
parent
8862d72a60
commit
1d743df4e3
@ -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,
|
||||
|
||||
149
pipeline_service/dingtalk_approval.py
Normal file
149
pipeline_service/dingtalk_approval.py
Normal file
@ -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
|
||||
@ -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."""
|
||||
|
||||
@ -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()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user