表名	用途
`dda_approvals`	审批记录（每次发起一条，含钉钉实例 ID、状态、意见）
`dda_approval_configs`	审批流程配置（biz_type → 钉钉审批模板 process_code）

安装与集成

1. 安装

pip install -e /path/to/dingdingflow          # 开发
# 或宿主 build.sh 里：pip install pkgs/dingdingflow/

2. 宿主入口注册（例：pipeline-app 的 `app/pipeline_app.py`）

from dingdingflow.init import load_dingdingflow, register_biz_handler

load_dingdingflow()        # 注册所有函数到 ServerEnv

3. 建表

models/*.json 是标准四段式表定义，用 json2ddl 生成 DDL：

cd pkgs/dingdingflow/models && json2ddl mysql . > /tmp/dda_ddl.sql
mysql -h <host> -u<user> -p<pwd> <db> < /tmp/dda_ddl.sql

4. RBAC 权限

py3/bin/python pkgs/dingdingflow/scripts/load_path.py

注册 9 个 logined 端点 + 1 个 any 端点（dingtalk_callback.dspy 必须 any，因为钉钉服务器回调没有登录态；安全性靠回调内部用 processInstanceId 匹配本地记录，匹配不到直接拒绝）。

5. wwwroot 软链 + 菜单

ln -sf ../pkgs/dingdingflow/wwwroot wwwroot/dingdingflow

配置在哪里设

A. 钉钉应用凭据 → 环境变量

放宿主的 init/.dingdingflow 文件（不入库，start.sh 里 source）：

export DINGTALK_APP_KEY=your_app_key
export DINGTALK_APP_SECRET=your_app_secret
export DINGTALK_AGENT_ID=your_agent_id

缺失这三个变量时 dingtalk_client 自动走 mock 模式（返回假的 access_token 和实例 ID），可在没有钉钉环境时先把流程跑通。

去哪拿：钉钉开放平台 → 应用开发 → 企业内部应用 → 凭据与基础信息。需要的权限：审批实例创建、审批实例读取、审批回调订阅。

B. 审批模板映射 → `dda_approval_configs` 表

每种业务类型配一条，把 biz_type 映射到钉钉审批模板：

字段	说明	示例
`biz_type`	业务类型标识（代码里 `submit_approval` 的第一个参数）	`pipeline_deploy`
`biz_type_title`	业务类型显示名	`产线部署审批`
`process_code`	钉钉审批模板编码	`PROC-XXXX-XXXX`
`agent_id`	钉钉应用 AgentId	`1234567`
`form_config`	表单字段映射 JSON（本地字段 → 钉钉表单控件）	见下
`is_active`	是否启用 `1`/`0`	`1`

process_code 去哪拿：钉钉管理后台 → 工作台 → 审批 → 选中模板 → URL 里的 processCode，或用钉钉 API /topapi/process/get_by_name 查。

配置入口（两种都行）：

页面：/dingdingflow/api/dd_approval_configs_list.dspy（CRUD 界面）
直接 SQL / 模块 init/data.json 种子

C. 钉钉回调地址

钉钉开放平台 → 事件订阅 → 填：

https://<你的域名>/dingdingflow/api/dingtalk_callback.dspy

订阅事件类型：bpms_instance_change（审批实例状态变化）。

审批节点在哪里设

本模块提供「审批能力」，审批节点挂在哪由宿主决定。两种接法：

接法一：产线步骤即审批节点（推荐给 pipeline-app）

产线的步骤定义在 pipeline_steps 表，step_type 决定用哪个 handler。所以审批节点 = 一个 step_type='dingtalk_approval' 的步骤：

# pipeline_service/handlers_approval.py
from dingdingflow.init import submit_approval, register_biz_handler

async def handle_dingtalk_approval(tenant_id, task_id, step_name, input_data, config):
    """审批步骤 handler：发起钉钉审批后挂起，等回调推进。"""
    r = await submit_approval(
        biz_type=config.get('biz_type', 'pipeline_step'),
        biz_id=f"{task_id}:{step_name}",          # 回调靠这个定位任务和步骤
        title=config.get('title', f'{step_name} 审批'),
        applicant_id=input_data.get('applicant_id', ''),
    )
    # 返回后由引擎把步骤置为等待人工状态，不阻塞 worker
    return {'approval_id': r.get('approval_id'), 'status': 'waiting_approval'}

register_handler('dingtalk_approval', handle_dingtalk_approval)

再注册回调钩子，把审批结果写回产线：

async def on_pipeline_approval(biz_id, status, approval_id, comment):
    task_id, step_name = biz_id.split(':', 1)
    env = ServerEnv()
    if status == 'approved':
        await env.approval_approve(tenant_id, task_id, step_name, 'dingtalk', comment)
    else:
        await env.approval_reject(tenant_id, task_id, step_name, 'dingtalk', comment)

register_biz_handler('pipeline_step', on_pipeline_approval)

节点具体位置：在 pipeline_steps 表插一行（或产线定义页面加一步）：

INSERT INTO pipeline_steps (id, pipeline_id, step_name, step_type, step_order, step_config)
VALUES (<id>, <产线id>, 'deploy_approval', 'dingtalk_approval', 30,
        '{"deps": ["build"], "biz_type": "pipeline_step", "title": "上线审批"}');

step_type 必须是 dingtalk_approval（匹配上面注册的 handler 名）
deps 控制它卡在哪个步骤之后（DAG 依赖）
后续步骤把它写进自己的 deps，就实现「审批通过才继续」

接法二：任意业务动作前置审批

不走产线引擎，直接在业务代码里调：

r = await submit_approval('content_publish', content_id, '发布审批', user_id)
# ...钉钉审批中...
# 回调时你注册的 handler 被调用，在里面做真正的业务动作
register_biz_handler('content_publish', my_publish_handler)

对外函数（`load_dingdingflow()` 注册到 ServerEnv）

函数	用途
`submit_approval(biz_type, biz_id, title, applicant_id, org_id)`	发起审批
`get_approval_status(approval_id)`	查审批状态（会同步钉钉最新状态）
`handle_dingtalk_callback(data)`	处理钉钉回调（回调 dspy 调它）
`register_biz_handler(biz_type, handler)`	注册审批结果处理器
`list_biz_handlers()`	查已注册钩子（排障）
`dd_approvals_` / `dd_approval_configs_`	两张表的 CRUD
`get_approval_config_by_type(org_id, biz_type)`	查审批模板配置

register_biz_handler 的 handler 签名：

async def handler(biz_id, status, approval_id, comment): ...
# status ∈ ('approved', 'rejected', 'cancelled')

设计约束

不硬编码库名：走宿主的 get_module_dbname('dingdingflow') 映射
不认识业务表：审批结果只通过 _BIZ_HANDLERS 分派，模块内零业务表引用（这是从 cms 独立时改掉的：原代码把 biz_type=='content_publish' 和写 cms_content 表硬编码在回调里，导致无法复用）
表名统一 dda_ 前缀：模块跨应用复用时不与业务表冲突

README.md Unescape Escape

dingdingflow — 钉钉审批工作流模块

数据表（统一 dda_ 前缀，避免跨应用复用时与业务表冲突）