yumoqing/dingdingflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 钉钉审批工作流模块从 cms 独立

Browse Source 
- 表名统一 dda_ 前缀(dda_approvals/dda_approval_configs)，跨应用复用防冲突
- 解耦关键改动：原回调硬编码 biz_type=='content_publish' + 写 cms_content 表，
  改为 register_biz_handler(biz_type, handler) 钩子分派，模块内零业务表引用
- 含 models(四段式)/json(CRUD)/9个dspy端点/load_path(any回调+9 logined)/i18n四语言
- README 覆盖凭据配置、审批模板配置、回调地址、审批节点接法(产线步骤)
This commit is contained in:
ymq 2026-08-25 18:15:52 +08:00
parent e3d5d8b694
commit a3f2ea8616
23 changed files with 1654 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

198
README.md
View File

@ -1,2 +1,198 @@
# dingdingflow
# dingdingflow — 钉钉审批工作流模块
从 cms 模块独立而来。**本模块只负责钉钉审批本身**（发起 / 查询 / 回调 / 模板配置），
不认识任何业务表。审批通过或驳回后要做什么，由宿主按 `biz_type` 注册钩子决定。
## 数据表（统一 `dda_` 前缀，避免跨应用复用时与业务表冲突）
| 表名 | 用途 |
|------|------|
| `dda_approvals` | 审批记录（每次发起一条，含钉钉实例 ID、状态、意见） |
| `dda_approval_configs` | 审批流程配置（biz_type → 钉钉审批模板 process_code） |
## 安装与集成
### 1. 安装
```bash
pip install -e /path/to/dingdingflow # 开发
# 或宿主 build.sh 里：pip install pkgs/dingdingflow/
```
### 2. 宿主入口注册（例：pipeline-app 的 `app/pipeline_app.py`）
```python
from dingdingflow.init import load_dingdingflow, register_biz_handler
load_dingdingflow() # 注册所有函数到 ServerEnv
```
### 3. 建表
`models/*.json` 是标准四段式表定义，用 `json2ddl` 生成 DDL：
```bash
cd pkgs/dingdingflow/models && json2ddl mysql . > /tmp/dda_ddl.sql
mysql -h <host> -u<user> -p<pwd> <db> < /tmp/dda_ddl.sql
```
### 4. RBAC 权限
```bash
py3/bin/python pkgs/dingdingflow/scripts/load_path.py
```
注册 9 个 logined 端点 + 1 个 any 端点（`dingtalk_callback.dspy` 必须 any，
因为钉钉服务器回调没有登录态；安全性靠回调内部用 `processInstanceId`
匹配本地记录，匹配不到直接拒绝）。
### 5. wwwroot 软链 + 菜单
```bash
ln -sf ../pkgs/dingdingflow/wwwroot wwwroot/dingdingflow
```
---
## 配置在哪里设
### A. 钉钉应用凭据 → 环境变量
放宿主的 `init/.dingdingflow` 文件（不入库，`start.sh` 里 source）：
```bash
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'` 的步骤：
```python
# 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)
```
再注册回调钩子，把审批结果写回产线：
```python
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` 表插一行（或产线定义页面加一步）：
```sql
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`，就实现「审批通过才继续」
### 接法二：任意业务动作前置审批
不走产线引擎，直接在业务代码里调：
```python
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 签名：
```python
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_` 前缀**：模块跨应用复用时不与业务表冲突

11
dingdingflow/__init__.py Normal file
View File

@ -0,0 +1,11 @@
"""dingdingflow — 钉钉审批工作流模块。"""
from .init import (
load_dingdingflow,
register_biz_handler,
list_biz_handlers,
submit_approval,
get_approval_status,
handle_dingtalk_callback,
)
__version__ = '1.0.0'

240
dingdingflow/dingtalk_client.py Normal file
View File

@ -0,0 +1,240 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
DingTalk API Client for approval workflow integration.
Reads credentials from environment variables (DINGTALK_APP_KEY, DINGTALK_APP_SECRET, DINGTALK_AGENT_ID).
Gracefully handles missing credentials by returning mock responses in dev mode.
"""
import os
import json
import logging
import time
import urllib.request
import urllib.error
logger = logging.getLogger(__name__)
class DingTalkClient:
"""Client for DingTalk Open API - approval workflow operations."""
BASE_URL = "https://oapi.dingtalk.com"
def __init__(self):
self.app_key = os.environ.get("DINGTALK_APP_KEY", "")
self.app_secret = os.environ.get("DINGTALK_APP_SECRET", "")
self.agent_id = os.environ.get("DINGTALK_AGENT_ID", "")
self.callback_token = os.environ.get("DINGTALK_CALLBACK_TOKEN", "")
self._access_token = None
self._token_expires_at = 0
if not self.app_key or not self.app_secret:
logger.warning(
"DingTalk credentials not configured (DINGTALK_APP_KEY / DINGTALK_APP_SECRET). "
"Running in dev/mock mode - API calls will return mock responses."
)
@property
def is_dev_mode(self):
"""Return True if credentials are missing (dev/mock mode)."""
return not self.app_key or not self.app_secret
def _http_post(self, url, data=None, params=None):
"""Make HTTP POST request and return parsed JSON response."""
if params:
query = "&".join(f"{k}={v}" for k, v in params.items())
url = f"{url}?{query}"
body = json.dumps(data).encode("utf-8") if data else b""
req = urllib.request.Request(
url,
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
resp_data = resp.read().decode("utf-8")
return json.loads(resp_data)
except urllib.error.HTTPError as e:
error_body = e.read().decode("utf-8", errors="replace")
logger.error("DingTalk API HTTP error %s: %s", e.code, error_body)
return {"errcode": e.code, "errmsg": f"HTTP {e.code}: {error_body}"}
except urllib.error.URLError as e:
logger.error("DingTalk API connection error: %s", str(e))
return {"errcode": -1, "errmsg": f"Connection error: {str(e)}"}
except Exception as e:
logger.error("DingTalk API unexpected error: %s", str(e))
return {"errcode": -1, "errmsg": str(e)}
def _http_get(self, url, params=None):
"""Make HTTP GET request and return parsed JSON response."""
if params:
query = "&".join(f"{k}={v}" for k, v in params.items())
url = f"{url}?{query}"
req = urllib.request.Request(url, headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=30) as resp:
resp_data = resp.read().decode("utf-8")
return json.loads(resp_data)
except urllib.error.HTTPError as e:
error_body = e.read().decode("utf-8", errors="replace")
logger.error("DingTalk API HTTP error %s: %s", e.code, error_body)
return {"errcode": e.code, "errmsg": f"HTTP {e.code}: {error_body}"}
except Exception as e:
logger.error("DingTalk API unexpected error: %s", str(e))
return {"errcode": -1, "errmsg": str(e)}
def get_access_token(self):
"""
Get DingTalk access token. Caches token until expiry.
Returns token string or empty string on failure.
"""
if self.is_dev_mode:
logger.info("Dev mode: returning mock access token")
return "mock_access_token_dev"
# Return cached token if still valid (with 5-min buffer)
now = time.time()
if self._access_token and now < self._token_expires_at - 300:
return self._access_token
url = f"{self.BASE_URL}/gettoken"
params = {"appkey": self.app_key, "appsecret": self.app_secret}
result = self._http_get(url, params=params)
if result.get("errcode") == 0:
self._access_token = result.get("access_token", "")
expires_in = result.get("expires_in", 7200)
self._token_expires_at = now + expires_in
logger.info("DingTalk access token obtained, expires in %ss", expires_in)
return self._access_token
else:
logger.error(
"Failed to get DingTalk access token: %s",
result.get("errmsg", "unknown error"),
)
return ""
def create_approval_instance(self, process_code, form_data, originator_user_id):
"""
Create a DingTalk approval instance.
Args:
process_code: DingTalk approval template code (from dda_approval_configs)
form_data: List of form component values, e.g.:
[{"name": "审批类型", "value": "内容发布"}, ...]
originator_user_id: DingTalk user ID of the applicant
Returns:
dict with keys:
- success (bool)
- instance_id (str) - DingTalk process instance ID
- errmsg (str) - error message if failed
"""
if self.is_dev_mode:
mock_instance_id = f"mock_instance_{int(time.time())}"
logger.info(
"Dev mode: mock approval instance created: %s (process_code=%s)",
mock_instance_id,
process_code,
)
return {
"success": True,
"instance_id": mock_instance_id,
"errmsg": "",
}
token = self.get_access_token()
if not token:
return {"success": False, "instance_id": "", "errmsg": "Failed to get access token"}
url = f"{self.BASE_URL}/topapi/processinstance/create"
payload = {
"agent_id": int(self.agent_id) if self.agent_id else 0,
"process_code": process_code,
"originator_user_id": originator_user_id,
"dept_id": -1,
"form_component_values": form_data,
}
result = self._http_post(url, data=payload, params={"access_token": token})
if result.get("errcode") == 0:
instance_id = result.get("process_instance_id", "")
logger.info("DingTalk approval instance created: %s", instance_id)
return {"success": True, "instance_id": instance_id, "errmsg": ""}
else:
errmsg = result.get("errmsg", "unknown error")
logger.error("Failed to create DingTalk approval: %s", errmsg)
return {"success": False, "instance_id": "", "errmsg": errmsg}
def get_approval_instance(self, instance_id):
"""
Get DingTalk approval instance details.
Args:
instance_id: DingTalk process instance ID
Returns:
dict with keys:
- success (bool)
- status (str) - NEW/RUNNING/COMPLETED/TERMINATED
- result (str) - agree/refuse (only when status=COMPLETED)
- data (dict) - full instance data
- errmsg (str) - error message if failed
"""
if self.is_dev_mode:
logger.info("Dev mode: mock get approval instance: %s", instance_id)
return {
"success": True,
"status": "COMPLETED",
"result": "agree",
"data": {
"process_instance_id": instance_id,
"status": "COMPLETED",
"result": "agree",
},
"errmsg": "",
}
token = self.get_access_token()
if not token:
return {"success": False, "status": "", "result": "", "data": {}, "errmsg": "Failed to get access token"}
url = f"{self.BASE_URL}/topapi/processinstance/get"
payload = {"process_instance_id": instance_id}
result = self._http_post(url, data=payload, params={"access_token": token})
if result.get("errcode") == 0:
pi = result.get("process_instance", {})
status = pi.get("status", "")
pi_result = pi.get("result", "")
return {
"success": True,
"status": status,
"result": pi_result,
"data": pi,
"errmsg": "",
}
else:
errmsg = result.get("errmsg", "unknown error")
logger.error("Failed to get DingTalk approval instance: %s", errmsg)
return {"success": False, "status": "", "result": "", "data": {}, "errmsg": errmsg}
# Module-level singleton
_client_instance = None
def get_dingtalk_client():
"""Get or create the DingTalkClient singleton."""
global _client_instance
if _client_instance is None:
_client_instance = DingTalkClient()
return _client_instance

449
dingdingflow/init.py Normal file
View File

@ -0,0 +1,449 @@
"""dingdingflow — 钉钉审批工作流独立模块
从 cms 模块独立而来。本模块只负责钉钉审批本身（发起/查询/回调/配置），
不认识任何业务表：审批通过/驳回后的业务动作由宿主通过 register_biz_handler()
按 biz_type 注册钩子，模块回调时分派调用。
宿主集成:
from dingdingflow.init import load_dingdingflow, register_biz_handler
load_dingdingflow()
register_biz_handler('pipeline_step', my_async_handler)
环境变量（缺失时 dingtalk_client 自动走 mock，便于本地开发）:
DINGTALK_APP_KEY / DINGTALK_APP_SECRET / DINGTALK_AGENT_ID
"""
import json
import logging
import datetime
from ahserver.serverenv import ServerEnv
from appPublic.uniqueID import getID
from sqlor.dbpools import DBPools
from .dingtalk_client import get_dingtalk_client
MODULE_NAME = 'dingdingflow'
MODULE_VERSION = '1.0.0'
logger = logging.getLogger(__name__)
# biz_type -> async handler(biz_id, status, approval_id, comment)
# 宿主注册；本模块不内置任何业务表知识
_BIZ_HANDLERS = {}
def register_biz_handler(biz_type, handler):
"""注册某个 biz_type 的审批结果处理器。
handler 签名: async def handler(biz_id, status, approval_id, comment)
status ∈ ('approved', 'rejected', 'cancelled')
"""
_BIZ_HANDLERS[biz_type] = handler
logger.info('[dingdingflow] biz handler registered: %s', biz_type)
def list_biz_handlers():
"""已注册的 biz_type 列表（排障用）。"""
return {k: getattr(v, '__name__', str(v)) for k, v in _BIZ_HANDLERS.items()}
def _get_dbname():
"""动态取库名，禁硬编码（宿主通过 get_module_dbname 映射）。"""
env = ServerEnv()
return env.get_module_dbname(MODULE_NAME)
# ═══════════════════════════════════════════════════════════════════════════════
# DD Approvals CRUD (原 dingdingflow)
# ═══════════════════════════════════════════════════════════════════════════════
async def dd_approvals_create(data):
"""创建审批记录"""
dbname = _get_dbname()
db = DBPools()
data['id'] = getID()
if 'org_id' not in data:
data['org_id'] = '0'
if 'status' not in data:
data['status'] = 'pending'
if 'created_at' not in data:
data['created_at'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
async with db.sqlorContext(dbname) as sor:
await sor.C('dda_approvals', data)
return {'id': data['id']}
async def dd_approvals_update(data):
"""更新审批记录"""
dbname = _get_dbname()
db = DBPools()
record_id = data.get('id')
if not record_id:
raise ValueError('id is required for update')
async with db.sqlorContext(dbname) as sor:
await sor.U('dda_approvals', data)
return {'id': record_id}
async def dd_approvals_delete(data):
"""删除审批记录"""
dbname = _get_dbname()
db = DBPools()
async with db.sqlorContext(dbname) as sor:
await sor.D('dda_approvals', data)
return data
async def dd_approvals_list(ns=None):
"""查询审批记录列表"""
dbname = _get_dbname()
db = DBPools()
async with db.sqlorContext(dbname) as sor:
ns = ns or {}
ns.setdefault('sort', 'created_at desc')
rows = await sor.R('dda_approvals', ns)
return {'rows': rows, 'total': len(rows)}
# ═══════════════════════════════════════════════════════════════════════════════
# DD Approval Configs CRUD (原 dingdingflow)
# ═══════════════════════════════════════════════════════════════════════════════
async def dd_approval_configs_create(data):
"""创建审批配置"""
dbname = _get_dbname()
db = DBPools()
data['id'] = getID()
if 'org_id' not in data:
data['org_id'] = '0'
if 'is_active' not in data:
data['is_active'] = '1'
now_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
if 'created_at' not in data:
data['created_at'] = now_str
data['updated_at'] = now_str
async with db.sqlorContext(dbname) as sor:
await sor.C('dda_approval_configs', data)
return {'id': data['id']}
async def dd_approval_configs_update(data):
"""更新审批配置"""
dbname = _get_dbname()
db = DBPools()
record_id = data.get('id')
if not record_id:
raise ValueError('id is required for update')
data['updated_at'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
async with db.sqlorContext(dbname) as sor:
await sor.U('dda_approval_configs', data)
return {'id': record_id}
async def dd_approval_configs_delete(data):
"""删除审批配置"""
dbname = _get_dbname()
db = DBPools()
async with db.sqlorContext(dbname) as sor:
await sor.D('dda_approval_configs', data)
return data
async def dd_approval_configs_list(ns=None):
"""查询审批配置列表"""
dbname = _get_dbname()
db = DBPools()
async with db.sqlorContext(dbname) as sor:
ns = ns or {}
ns.setdefault('sort', 'biz_type')
rows = await sor.R('dda_approval_configs', ns)
return {'rows': rows, 'total': len(rows)}
async def get_approval_config_by_type(org_id, biz_type):
"""根据org_id和biz_type获取审批配置"""
dbname = _get_dbname()
db = DBPools()
async with db.sqlorContext(dbname) as sor:
rows = await sor.R('dda_approval_configs', {'org_id': org_id, 'biz_type': biz_type})
if rows:
return rows[0]
return None
# ═══════════════════════════════════════════════════════════════════════════════
# Approval Workflow Business Logic (原 dingdingflow)
# ═══════════════════════════════════════════════════════════════════════════════
async def submit_approval(biz_type, biz_id, title, applicant_id, org_id='0'):
"""
提交审批请求:
1. 查找审批配置
2. 创建审批记录
3. 调用钉钉API创建审批实例
4. 保存钉钉实例ID
"""
client = get_dingtalk_client()
# 查找审批配置
config = await get_approval_config_by_type(org_id, biz_type)
if not config:
logger.error('No approval config found for org_id=%s, biz_type=%s', org_id, biz_type)
return {'success': False, 'message': f'No approval config found for biz_type={biz_type}'}
process_code = config.get('process_code', '') if isinstance(config, dict) else getattr(config, 'process_code', '') or ''
form_config_raw = config.get('form_config', '') if isinstance(config, dict) else getattr(config, 'form_config', '') or ''
# 从form_config构建表单数据
form_data = []
if form_config_raw:
try:
form_config = json.loads(form_config_raw) if isinstance(form_config_raw, str) else form_config_raw
if isinstance(form_config, list):
form_data = form_config
except (json.JSONDecodeError, TypeError) as e:
logger.warning('Failed to parse form_config: %s', str(e))
# 无表单数据时创建最小表单
if not form_data:
form_data = [
{'name': '审批标题', 'value': title},
{'name': '业务类型', 'value': biz_type},
]
# 调用钉钉API
result = client.create_approval_instance(process_code, form_data, applicant_id)
if not result['success']:
# API失败仍然创建记录
approval_data = {
'biz_type': biz_type,
'biz_id': biz_id,
'title': title,
'applicant_id': applicant_id,
'org_id': org_id,
'status': 'pending',
'dingtalk_instance_id': '',
'comment': f"DingTalk API error: {result.get('errmsg', '')}",
}
approval = await dd_approvals_create(approval_data)
return {
'success': False,
'message': f"DingTalk API failed: {result.get('errmsg', '')}",
'approval_id': approval['id'],
}
# 创建审批记录
approval_data = {
'biz_type': biz_type,
'biz_id': biz_id,
'title': title,
'applicant_id': applicant_id,
'org_id': org_id,
'status': 'pending',
'dingtalk_instance_id': result['instance_id'],
}
approval = await dd_approvals_create(approval_data)
logger.info(
'Approval submitted: id=%s, instance=%s, biz=%s/%s',
approval['id'], result['instance_id'], biz_type, biz_id,
)
return {
'success': True,
'message': 'Approval submitted successfully',
'approval_id': approval['id'],
'instance_id': result['instance_id'],
}
async def get_approval_status(approval_id):
"""查询钉钉审批最新状态并同步到本地"""
dbname = _get_dbname()
db = DBPools()
# 获取本地记录
async with db.sqlorContext(dbname) as sor:
rows = await sor.R('dda_approvals', {'id': approval_id})
if not rows:
return {'success': False, 'message': 'Approval record not found'}
record = rows[0]
instance_id = record.get('dingtalk_instance_id', '') if isinstance(record, dict) else getattr(record, 'dingtalk_instance_id', '')
current_status = record.get('status', '') if isinstance(record, dict) else getattr(record, 'status', '')
# 已完成无需再查
if current_status in ('approved', 'rejected', 'cancelled'):
return {
'success': True,
'status': current_status,
'approval_id': approval_id,
'instance_id': instance_id,
}
if not instance_id:
return {
'success': True,
'status': current_status,
'approval_id': approval_id,
'instance_id': '',
'message': 'No DingTalk instance ID, cannot sync',
}
# 查询钉钉
client = get_dingtalk_client()
dt_result = client.get_approval_instance(instance_id)
if not dt_result['success']:
return {
'success': False,
'message': f"DingTalk query failed: {dt_result.get('errmsg', '')}",
'status': current_status,
}
# 映射钉钉状态
dt_status = dt_result.get('status', '')
dt_result_val = dt_result.get('result', '')
new_status = current_status
if dt_status == 'COMPLETED':
if dt_result_val == 'agree':
new_status = 'approved'
elif dt_result_val == 'refuse':
new_status = 'rejected'
elif dt_status == 'TERMINATED':
new_status = 'cancelled'
# 更新本地记录
if new_status != current_status:
update_data = {'id': approval_id, 'status': new_status}
if new_status in ('approved', 'rejected', 'cancelled'):
update_data['completed_at'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
await dd_approvals_update(update_data)
logger.info('Approval %s status synced: %s -> %s', approval_id, current_status, new_status)
return {
'success': True,
'status': new_status,
'approval_id': approval_id,
'instance_id': instance_id,
}
async def handle_dingtalk_callback(data):
"""
处理钉钉webhook回调。
钉钉在审批状态变化时发送回调。
"""
logger.info('DingTalk callback received: %s', json.dumps(data, ensure_ascii=False))
instance_id = data.get('processInstanceId', '')
if not instance_id:
return {'success': False, 'message': 'Missing processInstanceId'}
callback_type = data.get('type', '')
if callback_type != 'bpms_instance_change':
logger.info('Ignoring callback type: %s', callback_type)
return {'success': True, 'message': f'Ignored callback type: {callback_type}'}
# 查找本地审批记录
dbname = _get_dbname()
db = DBPools()
async with db.sqlorContext(dbname) as sor:
rows = await sor.R('dda_approvals', {'dingtalk_instance_id': instance_id})
if not rows:
logger.warning('No local approval found for instance_id=%s', instance_id)
return {'success': False, 'message': f'No approval found for instance {instance_id}'}
record = rows[0]
record_id = record.get('id', '') if isinstance(record, dict) else getattr(record, 'id', '')
current_status = record.get('status', '') if isinstance(record, dict) else getattr(record, 'status', '')
# 映射回调状态
dt_result = data.get('result', '')
new_status = current_status
if dt_result == 'agree':
new_status = 'approved'
elif dt_result == 'refuse':
new_status = 'rejected'
elif callback_type == 'terminate':
new_status = 'cancelled'
# 更新记录
if new_status != current_status:
update_data = {
'id': record_id,
'status': new_status,
'comment': data.get('remark', ''),
}
if new_status in ('approved', 'rejected', 'cancelled'):
update_data['completed_at'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
await dd_approvals_update(update_data)
logger.info('Callback: approval %s updated to %s', record_id, new_status)
# 按 biz_type 分派给宿主注册的回调钩子（本模块不认识任何业务表）
biz_type = record.get('biz_type', '') if isinstance(record, dict) else getattr(record, 'biz_type', '')
biz_id = record.get('biz_id', '') if isinstance(record, dict) else getattr(record, 'biz_id', '')
handler = _BIZ_HANDLERS.get(biz_type)
if handler and biz_id:
try:
await handler(biz_id=biz_id, status=new_status, approval_id=record_id,
comment=data.get('remark', ''))
logger.info('Callback: biz handler %s done for %s -> %s',
biz_type, biz_id, new_status)
except Exception as e:
logger.error('Callback: biz handler %s failed for %s: %s',
biz_type, biz_id, e)
elif biz_id:
logger.warning('Callback: no handler registered for biz_type=%s '
'(use register_biz_handler)', biz_type)
return {
'success': True,
'message': f'Approval {record_id} updated to {new_status}',
'approval_id': record_id,
'status': new_status,
}
# ═══════════════════════════════════════════════════════════════════════════════
# Module Loader
# ═══════════════════════════════════════════════════════════════════════════════
# ═══════════════════════════════════════════════════════════════════════════════
# Module Loader
# ═══════════════════════════════════════════════════════════════════════════════
def load_dingdingflow():
"""注册钉钉审批模块函数到 ServerEnv。"""
env = ServerEnv()
# 审批记录 CRUD
env.dd_approvals_create = dd_approvals_create
env.dd_approvals_update = dd_approvals_update
env.dd_approvals_delete = dd_approvals_delete
env.dd_approvals_list = dd_approvals_list
# 审批流程配置 CRUD
env.dd_approval_configs_create = dd_approval_configs_create
env.dd_approval_configs_update = dd_approval_configs_update
env.dd_approval_configs_delete = dd_approval_configs_delete
env.dd_approval_configs_list = dd_approval_configs_list
env.get_approval_config_by_type = get_approval_config_by_type
# 审批业务逻辑
env.submit_approval = submit_approval
env.get_approval_status = get_approval_status
env.handle_dingtalk_callback = handle_dingtalk_callback
# 回调钩子注册（宿主用）
env.register_biz_handler = register_biz_handler
env.list_biz_handlers = list_biz_handlers
# 钉钉客户端
env.get_dingtalk_client = get_dingtalk_client
logger.info('dingdingflow module loaded (v%s)', MODULE_VERSION)
return True

10
i18n/en/msg.txt Normal file
View File

@ -0,0 +1,10 @@
钉钉审批: DingTalk Approval
审批记录: Approval Records
审批流程配置: Approval Flow Config
审批标题: Approval Title
审批意见: Approval Comment
业务类型: Business Type
钉钉审批模板编码: DingTalk Process Code
待审批: Pending
已通过: Approved
已驳回: Rejected

10
i18n/jp/msg.txt Normal file
View File

@ -0,0 +1,10 @@
钉钉审批: DingTalk承認
审批记录: 承認記録
审批流程配置: 承認フロー設定
审批标题: 承認タイトル
审批意见: 承認コメント
业务类型: 業務タイプ
钉钉审批模板编码: DingTalkプロセスコード
待审批: 承認待ち
已通过: 承認済み
已驳回: 却下

10
i18n/ko/msg.txt Normal file
View File

@ -0,0 +1,10 @@
钉钉审批: 딩톡 결재
审批记录: 결재 기록
审批流程配置: 결재 흐름 설정
审批标题: 결재 제목
审批意见: 결재 의견
业务类型: 업무 유형
钉钉审批模板编码: 딩톡 프로세스 코드
待审批: 대기 중
已通过: 승인됨
已驳回: 반려됨

10
i18n/zh/msg.txt Normal file
View File

@ -0,0 +1,10 @@
钉钉审批: 钉钉审批
审批记录: 审批记录
审批流程配置: 审批流程配置
审批标题: 审批标题
审批意见: 审批意见
业务类型: 业务类型
钉钉审批模板编码: 钉钉审批模板编码
待审批: 待审批
已通过: 已通过
已驳回: 已驳回

36
json/dda_approval_configs.json Normal file
View File

@ -0,0 +1,36 @@
{
"tblname": "dda_approval_configs",
"alias": "dda_approval_configs",
"title": "审批流程配置",
"params": {
"sortby": [
"biz_type"
],
"browserfields": {
"exclouded": [
"id",
"form_config"
],
"alters": {
"is_active": {
"uitype": "code",
"data": [
{
"value": "1",
"text": "启用"
},
{
"value": "0",
"text": "停用"
}
]
}
}
},
"editable": {
"new_data_url": "{{entire_url('../api/dingdingflow/dd_approval_configs_create.dspy')}}",
"update_data_url": "{{entire_url('../api/dingdingflow/dd_approval_configs_update.dspy')}}",
"delete_data_url": "{{entire_url('../api/dingdingflow/dd_approval_configs_delete.dspy')}}"
}
}
}

79
json/dda_approvals.json Normal file
View File

@ -0,0 +1,79 @@
{
"tblname": "dda_approvals",
"alias": "dda_approvals",
"title": "审批记录",
"params": {
"sortby": [
"created_at desc"
],
"data_filter": {
"AND": [
{
"field": "status",
"op": "=",
"var": "status_filter"
},
{
"field": "biz_type",
"op": "=",
"var": "biz_type_filter"
},
{
"field": "title",
"op": "LIKE",
"var": "title_filter"
}
]
},
"browserfields": {
"exclouded": [
"id"
],
"alters": {
"status": {
"uitype": "code",
"data": [
{
"value": "pending",
"text": "待审批"
},
{
"value": "approved",
"text": "已通过"
},
{
"value": "rejected",
"text": "已拒绝"
},
{
"value": "cancelled",
"text": "已取消"
}
]
},
"biz_type": {
"uitype": "code",
"data": [
{
"value": "content_publish",
"text": "内容发布"
},
{
"value": "content_update",
"text": "内容修改"
},
{
"value": "content_delete",
"text": "内容删除"
}
]
}
}
},
"editable": {
"new_data_url": "{{entire_url('../api/dingdingflow/dd_approvals_create.dspy')}}",
"update_data_url": "{{entire_url('../api/dingdingflow/dd_approvals_update.dspy')}}",
"delete_data_url": "{{entire_url('../api/dingdingflow/dd_approvals_delete.dspy')}}"
}
}
}

85
models/dda_approval_configs.json Normal file
View File

@ -0,0 +1,85 @@
{
"summary": [
{
"name": "dda_approval_configs",
"title": "审批流程配置表",
"primary": [
"id"
]
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "org_id",
"title": "组织ID",
"type": "str",
"length": 32,
"default": "0"
},
{
"name": "biz_type",
"title": "业务类型",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "biz_type_title",
"title": "业务类型名称",
"type": "str",
"length": 100
},
{
"name": "process_code",
"title": "钉钉审批模板编码",
"type": "str",
"length": 100
},
{
"name": "agent_id",
"title": "钉钉应用AgentId",
"type": "str",
"length": 100
},
{
"name": "form_config",
"title": "表单字段配置JSON",
"type": "text"
},
{
"name": "is_active",
"title": "是否启用(1/0)",
"type": "str",
"length": 1,
"default": "1"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"indexes": [
{
"name": "idx_apvcfg_org_type",
"idxtype": "unique",
"idxfields": [
"org_id",
"biz_type"
]
}
],
"codes": []
}

114
models/dda_approvals.json Normal file
View File

@ -0,0 +1,114 @@
{
"summary": [
{
"name": "dda_approvals",
"title": "审批记录表",
"primary": [
"id"
]
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "org_id",
"title": "组织ID",
"type": "str",
"length": 32,
"default": "0"
},
{
"name": "biz_type",
"title": "业务类型(content_publish等)",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "biz_id",
"title": "业务数据ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "title",
"title": "审批标题",
"type": "str",
"length": 255,
"nullable": "no"
},
{
"name": "applicant_id",
"title": "申请人ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "approver_id",
"title": "审批人ID",
"type": "str",
"length": 32
},
{
"name": "dingtalk_instance_id",
"title": "钉钉审批实例ID",
"type": "str",
"length": 100
},
{
"name": "status",
"title": "状态(pending/approved/rejected/cancelled)",
"type": "str",
"length": 32,
"default": "pending"
},
{
"name": "comment",
"title": "审批意见",
"type": "text"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp"
},
{
"name": "completed_at",
"title": "完成时间",
"type": "datetime"
}
],
"indexes": [
{
"name": "idx_approval_biz",
"idxtype": "index",
"idxfields": [
"biz_type",
"biz_id"
]
},
{
"name": "idx_approval_status",
"idxtype": "index",
"idxfields": [
"org_id",
"status"
]
},
{
"name": "idx_approval_applicant",
"idxtype": "index",
"idxfields": [
"applicant_id"
]
}
]
}

18
pyproject.toml Normal file
View File

@ -0,0 +1,18 @@
[build-system]
requires = ["setuptools>=45", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "dingdingflow"
version = "1.0.0"
description = "钉钉审批工作流模块 - 发起/查询/回调，业务动作由宿主按 biz_type 注册钩子"
requires-python = ">=3.8"
dependencies = [
"sqlor",
"bricks_for_python",
"requests",
]
[tool.setuptools.packages.find]
where = ["."]
include = ["dingdingflow*"]

42
scripts/load_path.py Normal file
View File

@ -0,0 +1,42 @@
#!/usr/bin/env python3
"""RBAC path registration for dingdingflow module.
宿主根目录执行：py3/bin/python pkgs/dingdingflow/scripts/load_path.py
（pipeline-app 的 load_path.sh 会自动扫描 pkgs/*/scripts/load_path.py）
"""
import subprocess
MOD = "dingdingflow"
# 钉钉服务器回调：外部 POST 无登录态，必须 any
# ⚠️ 安全：回调处理内部靠 processInstanceId 匹配本地记录，无匹配即拒绝
PATHS_ANY = [
f"/{MOD}/api/dingtalk_callback.dspy",
]
PATHS_LOGINED = [
f"/{MOD}",
f"/{MOD}/api/dd_approval_configs_create.dspy",
f"/{MOD}/api/dd_approval_configs_delete.dspy",
f"/{MOD}/api/dd_approval_configs_list.dspy",
f"/{MOD}/api/dd_approval_configs_update.dspy",
f"/{MOD}/api/dd_approvals_create.dspy",
f"/{MOD}/api/dd_approvals_delete.dspy",
f"/{MOD}/api/dd_approvals_list.dspy",
f"/{MOD}/api/dd_approvals_update.dspy",
]
def register_paths():
for path in PATHS_ANY:
subprocess.run(["py3/bin/python", "set_role_perm.py", "any", path])
print(f" any: {path}")
for path in PATHS_LOGINED:
subprocess.run(["py3/bin/python", "set_role_perm.py", "logined", path])
print(f" logined: {path}")
if __name__ == "__main__":
print(f"=== {MOD} RBAC registration ===")
register_paths()
print(f"Done. any={len(PATHS_ANY)} logined={len(PATHS_LOGINED)}")

37
wwwroot/api/dd_approval_configs_create.dspy Normal file
View File

@ -0,0 +1,37 @@
result = {'widgettype': 'Message', 'options': {'title': 'Error', 'message': 'Invalid request', 'type': 'error'}}
try:
biz_type = params_kw.get('biz_type', '')
biz_type_title = params_kw.get('biz_type_title', '')
process_code = params_kw.get('process_code', '')
if not biz_type:
result['options'] = {'title': 'Error', 'message': 'biz_type is required', 'type': 'error'}
else:
new_id = getID()
org_id = (await get_userorgid()) or '0'
now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
dbname = get_module_dbname('dingdingflow')
async with DBPools().sqlorContext(dbname) as sor:
await sor.sqlExe(
"INSERT INTO dda_approval_configs (id, org_id, biz_type, biz_type_title, process_code, agent_id, form_config, is_active, created_at, updated_at) VALUES (${id}$, ${org_id}$, ${biz_type}$, ${biz_type_title}$, ${process_code}$, ${agent_id}$, ${form_config}$, ${is_active}$, ${created_at}$, ${updated_at}$)",
{
'id': new_id,
'org_id': org_id,
'biz_type': biz_type,
'biz_type_title': biz_type_title,
'process_code': process_code,
'agent_id': params_kw.get('agent_id', ''),
'form_config': params_kw.get('form_config', ''),
'is_active': params_kw.get('is_active', '1'),
'created_at': now_str,
'updated_at': now_str
}
)
result = {'widgettype': 'Message', 'options': {'title': 'Success', 'message': '审批配置创建成功', 'type': 'success'}}
except Exception as e:
result['options'] = {'title': 'Error', 'message': f'创建失败: {str(e)}', 'type': 'error'}
return result

16
wwwroot/api/dd_approval_configs_delete.dspy Normal file
View File

@ -0,0 +1,16 @@
result = {'widgettype': 'Message', 'options': {'title': 'Error', 'message': 'Invalid request', 'type': 'error'}}
try:
record_id = params_kw.get('id', '')
if not record_id:
result['options'] = {'title': 'Error', 'message': 'ID is required', 'type': 'error'}
else:
dbname = get_module_dbname('dingdingflow')
async with DBPools().sqlorContext(dbname) as sor:
await sor.sqlExe("DELETE FROM dda_approval_configs WHERE id=${id}$", {'id': record_id})
result = {'widgettype': 'Message', 'options': {'title': 'Success', 'message': '审批配置删除成功', 'type': 'success'}}
except Exception as e:
result['options'] = {'title': 'Error', 'message': f'删除失败: {str(e)}', 'type': 'error'}
return result

54
wwwroot/api/dd_approval_configs_list.dspy Normal file
View File

@ -0,0 +1,54 @@
result = {'success': False, 'rows': [], 'total': 0}
try:
dbname = get_module_dbname('dingdingflow')
async with DBPools().sqlorContext(dbname) as sor:
where_clauses = []
where_ns = {}
# Optional filtering
is_active = params_kw.get('is_active', '')
if is_active:
where_clauses.append("is_active=${is_active}$")
where_ns['is_active'] = is_active
biz_type = params_kw.get('biz_type', '')
if biz_type:
where_clauses.append("biz_type=${biz_type}$")
where_ns['biz_type'] = biz_type
where_sql = " AND ".join(where_clauses)
where_prefix = " WHERE " if where_clauses else ""
# Count query
count_sql = "SELECT count(*) rcnt FROM dda_approval_configs" + where_prefix + where_sql
count_rows = await sor.sqlExe(count_sql, where_ns)
total = 0
if count_rows and len(count_rows) > 0:
r = count_rows[0]
total = getattr(r, 'rcnt', 0)
if total > 0:
ns = {
'page': int(params_kw.get('page', 1)),
'rows': int(params_kw.get('rows', 20)),
'sort': params_kw.get('sort', 'biz_type')
}
sql = "SELECT id, org_id, biz_type, biz_type_title, process_code, agent_id, form_config, is_active, created_at, updated_at FROM dda_approval_configs" + where_prefix + where_sql
query_ns = dict(list(ns.items()) + list(where_ns.items()))
rows = await sor.sqlExe(sql, query_ns)
if isinstance(rows, dict):
result['rows'] = rows.get('rows', [])
result['total'] = rows.get('total', total)
elif rows:
result['rows'] = [dict(r) if hasattr(r, 'keys') else r for r in rows]
result['total'] = total
else:
result['total'] = 0
result['success'] = True
except Exception as e:
result['error'] = str(e)
return result

32
wwwroot/api/dd_approval_configs_update.dspy Normal file
View File

@ -0,0 +1,32 @@
result = {'widgettype': 'Message', 'options': {'title': 'Error', 'message': 'Invalid request', 'type': 'error'}}
try:
record_id = params_kw.get('id', '')
if not record_id:
result['options'] = {'title': 'Error', 'message': 'ID is required', 'type': 'error'}
else:
dbname = get_module_dbname('dingdingflow')
async with DBPools().sqlorContext(dbname) as sor:
update_fields = []
update_ns = {'id': record_id}
for field in ['biz_type', 'biz_type_title', 'process_code', 'agent_id', 'form_config', 'is_active']:
val = params_kw.get(field)
if val is not None:
update_fields.append(f"{field}=${field}$")
update_ns[field] = val
# Always update updated_at
now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
update_fields.append("updated_at=${updated_at}$")
update_ns['updated_at'] = now_str
if update_fields:
set_clause = ", ".join(update_fields)
await sor.sqlExe(f"UPDATE dda_approval_configs SET {set_clause} WHERE id=${id}$", update_ns)
result = {'widgettype': 'Message', 'options': {'title': 'Success', 'message': '审批配置更新成功', 'type': 'success'}}
except Exception as e:
result['options'] = {'title': 'Error', 'message': f'更新失败: {str(e)}', 'type': 'error'}
return result

39
wwwroot/api/dd_approvals_create.dspy Normal file
View File

@ -0,0 +1,39 @@
result = {'widgettype': 'Message', 'options': {'title': 'Error', 'message': 'Invalid request', 'type': 'error'}}
try:
biz_type = params_kw.get('biz_type', '')
biz_id = params_kw.get('biz_id', '')
title = params_kw.get('title', '')
applicant_id = params_kw.get('applicant_id', '')
if not biz_type or not title:
result['options'] = {'title': 'Error', 'message': 'biz_type and title are required', 'type': 'error'}
else:
new_id = getID()
org_id = (await get_userorgid()) or '0'
now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
dbname = get_module_dbname('dingdingflow')
async with DBPools().sqlorContext(dbname) as sor:
await sor.sqlExe(
"INSERT INTO dda_approvals (id, org_id, biz_type, biz_id, title, applicant_id, approver_id, dingtalk_instance_id, status, comment, created_at) VALUES (${id}$, ${org_id}$, ${biz_type}$, ${biz_id}$, ${title}$, ${applicant_id}$, ${approver_id}$, ${dingtalk_instance_id}$, ${status}$, ${comment}$, ${created_at}$)",
{
'id': new_id,
'org_id': org_id,
'biz_type': biz_type,
'biz_id': biz_id,
'title': title,
'applicant_id': applicant_id,
'approver_id': params_kw.get('approver_id', ''),
'dingtalk_instance_id': params_kw.get('dingtalk_instance_id', ''),
'status': params_kw.get('status', 'pending'),
'comment': params_kw.get('comment', ''),
'created_at': now_str
}
)
result = {'widgettype': 'Message', 'options': {'title': 'Success', 'message': '审批记录创建成功', 'type': 'success'}}
except Exception as e:
result['options'] = {'title': 'Error', 'message': f'创建失败: {str(e)}', 'type': 'error'}
return result

16
wwwroot/api/dd_approvals_delete.dspy Normal file
View File

@ -0,0 +1,16 @@
result = {'widgettype': 'Message', 'options': {'title': 'Error', 'message': 'Invalid request', 'type': 'error'}}
try:
record_id = params_kw.get('id', '')
if not record_id:
result['options'] = {'title': 'Error', 'message': 'ID is required', 'type': 'error'}
else:
dbname = get_module_dbname('dingdingflow')
async with DBPools().sqlorContext(dbname) as sor:
await sor.sqlExe("DELETE FROM dda_approvals WHERE id=${id}$", {'id': record_id})
result = {'widgettype': 'Message', 'options': {'title': 'Success', 'message': '审批记录删除成功', 'type': 'success'}}
except Exception as e:
result['options'] = {'title': 'Error', 'message': f'删除失败: {str(e)}', 'type': 'error'}
return result

72
wwwroot/api/dd_approvals_list.dspy Normal file
View File

@ -0,0 +1,72 @@
result = {'success': False, 'rows': [], 'total': 0}
try:
dbname = get_module_dbname('dingdingflow')
async with DBPools().sqlorContext(dbname) as sor:
# Build WHERE clause from data_filter or direct params
where_clauses = []
where_ns = {}
# Support data_filter from CRUD search popup
data_filter_str = params_kw.get('data_filter', '')
if data_filter_str:
# Individual filter values passed alongside data_filter
status_filter = params_kw.get('status_filter', '')
biz_type_filter = params_kw.get('biz_type_filter', '')
title_filter = params_kw.get('title_filter', '')
if status_filter:
where_clauses.append("status=${status_filter}$")
where_ns['status_filter'] = status_filter
if biz_type_filter:
where_clauses.append("biz_type=${biz_type_filter}$")
where_ns['biz_type_filter'] = biz_type_filter
if title_filter:
where_clauses.append("title LIKE ${title_filter}$")
where_ns['title_filter'] = f'%{title_filter}%'
else:
# Direct param filtering
status = params_kw.get('status', '')
if status:
where_clauses.append("status=${status}$")
where_ns['status'] = status
biz_type = params_kw.get('biz_type', '')
if biz_type:
where_clauses.append("biz_type=${biz_type}$")
where_ns['biz_type'] = biz_type
where_sql = " AND ".join(where_clauses)
where_prefix = " WHERE " if where_clauses else ""
# Count query
count_sql = "SELECT count(*) rcnt FROM dda_approvals" + where_prefix + where_sql
count_rows = await sor.sqlExe(count_sql, where_ns)
total = 0
if count_rows and len(count_rows) > 0:
r = count_rows[0]
total = getattr(r, 'rcnt', 0)
if total > 0:
ns = {
'page': int(params_kw.get('page', 1)),
'rows': int(params_kw.get('rows', 20)),
'sort': params_kw.get('sort', 'created_at desc')
}
sql = "SELECT id, org_id, biz_type, biz_id, title, applicant_id, approver_id, dingtalk_instance_id, status, comment, created_at, completed_at FROM dda_approvals" + where_prefix + where_sql
query_ns = dict(list(ns.items()) + list(where_ns.items()))
rows = await sor.sqlExe(sql, query_ns)
if isinstance(rows, dict):
result['rows'] = rows.get('rows', [])
result['total'] = rows.get('total', total)
elif rows:
result['rows'] = [dict(r) if hasattr(r, 'keys') else r for r in rows]
result['total'] = total
else:
result['total'] = 0
result['success'] = True
except Exception as e:
result['error'] = str(e)
return result

32
wwwroot/api/dd_approvals_update.dspy Normal file
View File

@ -0,0 +1,32 @@
result = {'widgettype': 'Message', 'options': {'title': 'Error', 'message': 'Invalid request', 'type': 'error'}}
try:
record_id = params_kw.get('id', '')
if not record_id:
result['options'] = {'title': 'Error', 'message': 'ID is required', 'type': 'error'}
else:
dbname = get_module_dbname('dingdingflow')
async with DBPools().sqlorContext(dbname) as sor:
update_fields = []
update_ns = {'id': record_id}
for field in ['biz_type', 'biz_id', 'title', 'applicant_id', 'approver_id', 'dingtalk_instance_id', 'status', 'comment']:
val = params_kw.get(field)
if val is not None:
update_fields.append(f"{field}=${field}$")
update_ns[field] = val
completed_at = params_kw.get('completed_at')
if completed_at:
update_fields.append("completed_at=${completed_at}$")
update_ns['completed_at'] = completed_at
if update_fields:
set_clause = ", ".join(update_fields)
await sor.sqlExe(f"UPDATE dda_approvals SET {set_clause} WHERE id=${id}$", update_ns)
result = {'widgettype': 'Message', 'options': {'title': 'Success', 'message': '审批记录更新成功', 'type': 'success'}}
except Exception as e:
result['options'] = {'title': 'Error', 'message': f'更新失败: {str(e)}', 'type': 'error'}
return result

45
wwwroot/api/dingtalk_callback.dspy Normal file
View File

@ -0,0 +1,45 @@
result = {'success': False, 'message': 'Invalid callback'}
try:
# DingTalk callback data comes via params_kw (POST body auto-parsed)
callback_data = {}
# params_kw contains the parsed POST body fields
process_instance_id = params_kw.get('processInstanceId', '')
callback_type = params_kw.get('type', '')
callback_result = params_kw.get('result', '')
staff_id = params_kw.get('staffId', '')
process_code = params_kw.get('processCode', '')
remark = params_kw.get('remark', '')
# Also handle nested JSON body case where entire body is under a key
if not process_instance_id:
body = params_kw.get('body', None)
if isinstance(body, dict):
process_instance_id = body.get('processInstanceId', '')
callback_type = body.get('type', '')
callback_result = body.get('result', '')
staff_id = body.get('staffId', '')
process_code = body.get('processCode', '')
remark = body.get('remark', '')
callback_data = {
'processInstanceId': process_instance_id,
'type': callback_type,
'result': callback_result,
'staffId': staff_id,
'processCode': process_code,
'remark': remark,
}
if not process_instance_id:
result = {'success': False, 'message': 'Missing processInstanceId'}
else:
# Call the handle_dingtalk_callback function registered via load_dingdingflow()
callback_result_data = await handle_dingtalk_callback(callback_data)
result = callback_result_data
except Exception as e:
result = {'success': False, 'message': f'Callback processing error: {str(e)}'}
return result