审批引擎设计文档

**版本**: v1.0

**最后更新**: 2026-06-04

**状态**: 设计草案

---

1. 概述

审批引擎（Approval Engine）是 TopCentral Mall 系统的核心业务组件，负责处理各类审批流程的创建、流转、审批、通知和归档。本文档描述审批引擎的整体架构设计、多级审批流程、自动升级机制以及与 WeChat Work 的集成方案。

---

2. 架构设计

┌─────────────────────────────────────────────────────────┐
│                    Approval Engine                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌────────┐  │
│  │ 创建服务  │  │ 流转引擎  │  │ 审批执行  │  │ 归档器 │  │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └───┬────┘  │
│       │             │             │             │       │
│  ┌────┴─────────────┴─────────────┴─────────────┴────┐  │
│  │               Approval DB (PostgreSQL)              │  │
│  └──────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘
         │                 │                  │
         ▼                 ▼                  ▼
  ┌──────────┐    ┌──────────────┐    ┌──────────────┐
  │  Notifier │    │  Escalator   │    │  Audit Trail │
  └──────────┘    └──────────────┘    └──────────────┘

2.1 核心模块

---

3. 多级审批流程（金额驱动）

3.1 审批层级定义

审批层级根据 **申请金额** 自动路由到不同的审批链：

3.2 审批链（Approval Chain）

销售 (提交)
  │
  ▼
┌─────────────────────────────────────────┐
│  销售经理 (Level 1)                     │
│  - 金额权限: ¥0 - ¥100,000              │
│  - 处理超时: 24h → 自动升级到上级       │
└────────────┬────────────────────────────┘
             │
             ▼ (金额 > ¥10,000 时)
┌─────────────────────────────────────────┐
│  部门总监 / 销售总监 (Level 2)          │
│  - 金额权限: ¥100,000 - ¥500,000        │
│  - 处理超时: 24h → 自动升级到 CEO       │
└────────────┬────────────────────────────┘
             │
             ▼ (金额 > ¥100,000 时)
┌─────────────────────────────────────────┐
│  CEO / 总经理 (Level 3)                 │
│  - 金额权限: ¥500,000 以上              │
│  - 需法务会签 (超 ¥500,000)             │
└─────────────────────────────────────────┘

3.3 金额判定规则

def determine_approval_chain(amount: float) -> list:
    """根据金额确定审批链"""
    chain = [
        {"level": 1, "role": "sales_manager", "name": "销售经理"},
    ]
    if amount > 10000:
        chain.append({"level": 2, "role": "sales_director", "name": "销售总监"})
    if amount > 100000:
        chain.append({"level": 3, "role": "ceo", "name": "CEO / 总经理"})
    if amount > 500000:
        chain.append({"level": 4, "role": "legal", "name": "法务会签", "type": "countersign"})
    return chain

---

4. 自动升级机制（Auto-Escalation）

4.1 升级规则

4.2 升级流程

申请提交（等待审批）
     │
     ▼
┌──────────────────────┐
│  等待经理审批         │  ← 倒计时 24h
│  [⏰ 剩余 23:59:59]  │
└──────┬───────────────┘
       │ 超时
       ▼
┌──────────────────────┐
│  自动升级到总监       │  ← 原经理被跳过
│  发送通知给总监+经理  │
│  [⏰ 剩余 23:59:59]  │
└──────┬───────────────┘
       │ 再次超时
       ▼
┌──────────────────────┐
│  紧急状态 - 通知 CEO  │  ← 标记为 "urgent"
│  全管理层广播通知     │
└──────────────────────┘

4.3 定时任务实现

# cron_service.py
ESCLATION_CRON = {
    "name": "approval_escalation_check",
    "schedule": "*/30 * * * *",       # 每30分钟检查一次
    "description": "审批超时自动升级检查",
    "handler": "check_escalations",
}

async def check_escalations():
    """查询所有超时的待审批项，执行升级"""
    overdue = await ApprovalModel.find_overdue(hours=24)
    for item in overdue:
        if item.escalation_count == 0:
            # 首次超时: 升级到上级
            await escalate_to_supervisor(item)
        elif item.escalation_count == 1:
            # 二次超时: 升级到CEO
            await escalate_to_ceo(item)
        else:
            # 三次超时: 紧急标记
            await mark_as_urgent(item)
        item.escalation_count += 1
        item.escalated_at = datetime.utcnow()
        await item.save()

---

5. WeChat Work 通知集成

5.1 通知类型

5.2 消息卡片模板（Markdown）

{
  "touser": "user_id",
  "msgtype": "template_card",
  "template_card": {
    "card_type": "news_notice",
    "source": {
      "icon_url": "https://mall.topcentral.com/static/logo.png",
      "desc": "TopCentral Mall 审批系统"
    },
    "main_title": {
      "title": "📋 新的审批申请",
      "desc": "请及时处理"
    },
    "card_image": { "url": "" },
    "vertical_content_list": [
      {
        "title": "审批类型",
        "desc": "客户信用额度申请"
      },
      {
        "title": "申请人",
        "desc": "张三 (sales@topcentral.com)"
      },
      {
        "title": "金额",
        "desc": "¥ 150,000.00"
      },
      {
        "title": "申请时间",
        "desc": "2026-06-04 10:30:00"
      }
    ],
    "horizontal_content_list": [
      { "keyname": "当前步骤", "value": "销售经理审批" },
      { "keyname": "超时时间", "value": "2026-06-05 10:30 (剩余 23h)" }
    ],
    "card_action": {
      "type": 1,
      "url": "https://mall.topcentral.com/approval/detail?id=APP20260604001"
    },
    "jump_list": [
      { "type": 1, "title": "查看详情", "url": "https://mall.topcentral.com/approval/detail?id=APP20260604001" },
      { "type": 1, "title": "快速审批", "url": "https://mall.topcentral.com/approval/quick-approve?id=APP20260604001" }
    ]
  }
}

5.3 通知适配器接口

class WeChatWorkNotifier:
    """企业微信通知适配器"""
    
    def __init__(self):
        self.corp_id = os.getenv("WECHAT_CORP_ID")
        self.agent_id = os.getenv("WECHAT_AGENT_ID")
        self.secret = os.getenv("WECHAT_SECRET")
        self.token_cache = {}
    
    async def get_access_token(self) -> str:
        """获取 access_token（缓存 + 自动刷新）"""
        if self.token_cache.get("expires_at", 0) > time.time():
            return self.token_cache["token"]
        # 调用企业微信API获取新token
        ...
    
    async def send_approval_notification(self, approval: ApprovalModel, 
                                          notification_type: str):
        """发送审批通知"""
        template = self._build_template(approval, notification_type)
        await self._send_message(approval.approver_id, template)
    
    async def send_escalation_notice(self, approval: ApprovalModel):
        """发送升级通知给上级"""
        supervisor = await UserService.get_supervisor(approval.approver_id)
        template = self._build_escalation_template(approval)
        await self._send_message(supervisor.wechat_id, template)
        # 同时通知原审批人
        await self._send_message(
            approval.approver_id,
            self._build_escalation_warning_template(approval)
        )

---

6. 数据库设计

6.1 核心表结构

-- 审批单主表
CREATE TABLE approvals (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    approval_no     VARCHAR(32) UNIQUE NOT NULL,       -- 审批编号, e.g. APP20260604001
    type            VARCHAR(32) NOT NULL,                -- 审批类型
    title           VARCHAR(200) NOT NULL,               -- 审批标题
    applicant_id    UUID NOT NULL,                       -- 申请人
    department_id   UUID,                                -- 部门
    amount          DECIMAL(15,2) DEFAULT 0,             -- 金额
    status          VARCHAR(20) DEFAULT 'pending',       -- pending/approved/rejected/cancelled
    current_step    INT DEFAULT 1,                       -- 当前步骤
    total_steps     INT NOT NULL,                        -- 总步骤数
    escalation_count INT DEFAULT 0,                      -- 升级次数
    form_data       JSONB,                               -- 申请表单数据
    created_at      TIMESTAMP DEFAULT NOW(),
    updated_at      TIMESTAMP DEFAULT NOW(),
    completed_at    TIMESTAMP
);

-- 审批节点表
CREATE TABLE approval_nodes (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    approval_id     UUID NOT NULL REFERENCES approvals(id),
    step            INT NOT NULL,                        -- 步骤序号
    role            VARCHAR(32) NOT NULL,                -- 角色
    approver_id     UUID,                                -- 审批人
    status          VARCHAR(20) DEFAULT 'pending',       -- pending/approved/rejected/skipped
    comment         TEXT,                                -- 审批意见
    deadline        TIMESTAMP,                           -- 处理截止时间
    escalated       BOOLEAN DEFAULT FALSE,               -- 是否已升级
    processed_at    TIMESTAMP,
    created_at      TIMESTAMP DEFAULT NOW()
);

-- 审批操作日志
CREATE TABLE approval_logs (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    approval_id     UUID NOT NULL REFERENCES approvals(id),
    action          VARCHAR(32) NOT NULL,                -- submit/approve/reject/escalate/comment
    operator_id     UUID NOT NULL,                       -- 操作人
    operator_name   VARCHAR(100),
    detail          JSONB,                               -- 操作详情
    created_at      TIMESTAMP DEFAULT NOW()
);

-- 索引
CREATE INDEX idx_approvals_status ON approvals(status);
CREATE INDEX idx_approvals_applicant ON approvals(applicant_id);
CREATE INDEX idx_approvals_amount ON approvals(amount);
CREATE INDEX idx_approval_nodes_approver ON approval_nodes(approver_id);
CREATE INDEX idx_approval_nodes_deadline ON approval_nodes(deadline) WHERE status = 'pending';

---

7. API 设计

---

8. 与现有系统的集成

8.1 与现有 `approval_service.py` 的关系

现有的 backend/app/services/approval_service.py 提供了：

新引擎将：

8.2 与 RBAC 系统的集成

Approval Engine → RBAC Service → User Roles
     │                │
     ▼                ▼
审批节点角色映射     用户权限校验

---

9. 安全与审计

---

10. 实施路线图

---

11. 附录

11.1 关键决策记录

11.2 相关文件