# 工作流模块（2.1.11 流程管理）实施计划 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 实现完整的自研工作流系统——包含流程配置管理、AntV X6图形化设计器、待办中心、已办中心、后端图遍历引擎。 **Architecture:** 后端图遍历引擎读取 wf_node + wf_transition 关系表执行节点分发。前端 AntV X6 设计器保存 graph_json（设计器快照）+ 同步写入 wf_node/wf_transition（运行时数据）。业务模块通过 API 调用发起流程，引擎通过回调通知业务状态变更。 **Tech Stack:** FastAPI, SQLAlchemy, Pydantic v2, Vue 3, Element Plus, AntV X6, TypeScript **Design Spec:** `docs/superpowers/specs/2026-05-13-workflow-module-design.md` --- ## File Structure ### Backend（新建文件） - `backend/app/models/models.py` — 新增6个wf_*模型（追加到现有文件） - `backend/app/schemas/workflow.py` — Pydantic请求/响应schemas - `backend/app/crud/workflow.py` — CRUD操作 - `backend/app/services/workflow_engine.py` — 核心引擎（节点分发、条件评估、审批人解析） - `backend/app/services/workflow_callback.py` — 回调服务 - `backend/app/api/workflow.py` — API路由 - `backend/app/api/__init__.py` — 注册workflow路由（修改） - `backend/app/main.py` — 注册workflow路由（修改） - `docs/sql/zkxw.sql` — 数据库全量导出基线；增量 schema 可先落库再重导或另附 `docs/sql/` 补丁（如 `callback_url` 字段） ### Frontend（新建文件） - `web-admin/src/api/workflow.ts` — API调用 - `web-admin/src/types/api.d.ts` — 新增Workflow命名空间（追加） - `web-admin/src/views/system/workflow/process/index.vue` — 流程配置管理 - `web-admin/src/views/system/workflow/designer/FlowDesigner.vue` — AntV X6设计器 - `web-admin/src/views/system/workflow/task-pending/index.vue` — 待办中心 - `web-admin/src/views/system/workflow/task-completed/index.vue` — 已办中心 - `web-admin/src/views/system/workflow/components/ApprovalDialog.vue` — 审批弹窗 - `web-admin/src/views/system/workflow/components/ProcessStatusTag.vue` — 状态标签 - `web-admin/src/router/modules/sqxp.ts` — 新增3个路由（修改） - `web-admin/src/App.vue` — 标题白名单（修改） ### SQL数据 - `docs/sql/zkxw.sql` — 全库导出（含 `sys_menu` 等）；若仅需菜单补丁可在 `docs/sql/` 另增增量脚本 --- ## Phase 1: Backend Foundation ### Task 1: SQLAlchemy Models（6个wf_*模型） **Files:** - Modify: `backend/app/models/models.py`（追加到文件末尾） - [ ] **Step 1: 在 models.py 末尾追加6个wf_*模型** 在文件末尾追加以下代码。字段顺序遵循规范：id → 业务字段 → tenant_no → order_num → remark → create_by → create_dept_id → create_time → update_by → update_time → del_flag。 ```python # ============================================================================= # 工作流（2.1.11 流程管理） # ============================================================================= class WfProcessDef(Base): """流程定义 wf_process_def""" __tablename__ = "wf_process_def" id = Column(String(64), primary_key=True, comment="主键ID") process_code = Column(String(64), nullable=False, comment="流程编码") process_name = Column(String(200), nullable=False, comment="流程名称") status = Column(CHAR(1), nullable=False, default="1", comment="1启用 0停用") tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号") order_num = Column(Integer, nullable=True, comment="排序号") remark = Column(String(512), nullable=True, comment="备注") create_by = Column(String(64), nullable=True, comment="创建人id") create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID") create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间") update_by = Column(String(64), nullable=True, comment="更新人id") update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间") del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除") class WfProcessVersion(Base): """流程定义版本 wf_process_version""" __tablename__ = "wf_process_version" id = Column(String(64), primary_key=True, comment="主键ID") process_def_id = Column(String(64), nullable=False, comment="流程定义ID") version_no = Column(Integer, nullable=False, comment="版本号，递增") graph_json = Column(Text, nullable=True, comment="节点与连线等配置（JSON字符串）") is_latest = Column(CHAR(1), default="0", comment="是否最新版本 1是 0否") tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号") order_num = Column(Integer, nullable=True, comment="排序号") remark = Column(String(512), nullable=True, comment="备注") create_by = Column(String(64), nullable=True, comment="创建人id") create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID") create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间") update_by = Column(String(64), nullable=True, comment="更新人id") update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间") del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除") class WfNode(Base): """流程节点 wf_node""" __tablename__ = "wf_node" id = Column(String(64), primary_key=True, comment="主键ID") process_version_id = Column(String(64), nullable=False, comment="流程版本ID") node_key = Column(String(64), nullable=False, comment="节点唯一键（画布内）") node_type = Column(String(32), nullable=False, comment="start/end/approval/condition等") node_name = Column(String(200), nullable=True, comment="节点名称") config_json = Column(Text, nullable=True, comment="审批人/SQL/会签策略等配置") tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号") order_num = Column(Integer, nullable=True, comment="排序号") remark = Column(String(512), nullable=True, comment="备注") create_by = Column(String(64), nullable=True, comment="创建人id") create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID") create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间") update_by = Column(String(64), nullable=True, comment="更新人id") update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间") del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除") class WfTransition(Base): """流程连线/转移 wf_transition""" __tablename__ = "wf_transition" id = Column(String(64), primary_key=True, comment="主键ID") process_version_id = Column(String(64), nullable=False, comment="流程版本ID") from_node_key = Column(String(64), nullable=False, comment="起始节点key") to_node_key = Column(String(64), nullable=False, comment="目标节点key") condition_expr = Column(String(512), nullable=True, comment="条件表达式JSON") tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号") order_num = Column(Integer, nullable=True, comment="排序号") remark = Column(String(512), nullable=True, comment="备注") create_by = Column(String(64), nullable=True, comment="创建人id") create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID") create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间") update_by = Column(String(64), nullable=True, comment="更新人id") update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间") del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除") class WfProcessInstance(Base): """流程实例 wf_process_instance""" __tablename__ = "wf_process_instance" id = Column(String(64), primary_key=True, comment="主键ID") process_version_id = Column(String(64), nullable=False, comment="发起时锁定的流程版本ID") business_type = Column(String(64), nullable=True, comment="业务类型") business_id = Column(String(64), nullable=True, comment="业务主键") starter_user_id = Column(String(64), nullable=True, comment="发起人") variables_json = Column(Text, nullable=True, comment="流程变量JSON") status = Column(String(32), nullable=True, comment="running/completed/rejected/cancelled") callback_url = Column(String(512), nullable=True, comment="业务回调地址") tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号") order_num = Column(Integer, nullable=True, comment="排序号") remark = Column(String(512), nullable=True, comment="备注") create_by = Column(String(64), nullable=True, comment="创建人id") create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID") create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间") update_by = Column(String(64), nullable=True, comment="更新人id") update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间") del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除") class WfTask(Base): """流程任务/待办 wf_task""" __tablename__ = "wf_task" id = Column(String(64), primary_key=True, comment="主键ID") instance_id = Column(String(64), nullable=False, comment="流程实例ID") node_key = Column(String(64), nullable=True, comment="当前节点key") assignee_user_id = Column(String(64), nullable=True, comment="待办人") status = Column(String(32), nullable=True, comment="pending/approved/rejected/withdrawn") comment_text = Column(String(512), nullable=True, comment="审批意见") tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号") order_num = Column(Integer, nullable=True, comment="排序号") remark = Column(String(512), nullable=True, comment="备注") create_by = Column(String(64), nullable=True, comment="创建人id") create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID") create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间") update_by = Column(String(64), nullable=True, comment="更新人id") update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间") del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除") ``` - [ ] **Step 2: 创建Schema变更SQL** 在目标库执行下列 `ALTER TABLE` 后，将变更纳入 `docs/sql/zkxw.sql`（重导）或 `docs/sql/` 下增量补丁： ```sql -- 为 wf_process_instance 新增 callback_url 字段 ALTER TABLE `wf_process_instance` ADD COLUMN `callback_url` VARCHAR(512) DEFAULT NULL COMMENT '业务回调地址' AFTER `status`; ``` - [ ] **Step 3: 验证模型可导入** Run: `cd backend && python -c "from app.models.models import WfProcessDef, WfProcessVersion, WfNode, WfTransition, WfProcessInstance, WfTask; print('OK')"` Expected: OK --- ### Task 2: Pydantic Schemas **Files:** - Create: `backend/app/schemas/workflow.py` - [ ] **Step 1: 创建workflow schemas文件** ```python """Pydantic schemas - 工作流管理""" from pydantic import BaseModel, Field from typing import Optional, List, Any # --------------------------------------------------------------------------- # 流程定义 # --------------------------------------------------------------------------- class ProcessDefCreate(BaseModel): processCode: str = Field(..., min_length=1, max_length=64, description="流程编码") processName: str = Field(..., min_length=1, max_length=200, description="流程名称") status: Optional[str] = Field("1", description="1启用 0停用") orderNum: Optional[int] = Field(0, description="排序号") remark: Optional[str] = Field(None, description="备注") class ProcessDefUpdate(BaseModel): processCode: Optional[str] = Field(None, min_length=1, max_length=64) processName: Optional[str] = Field(None, min_length=1, max_length=200) status: Optional[str] = None orderNum: Optional[int] = None remark: Optional[str] = None class ProcessDefResponse(BaseModel): id: str processCode: str processName: str status: str latestVersion: Optional[int] = None tenantNo: str orderNum: int remark: Optional[str] = None createTime: str updateTime: str # --------------------------------------------------------------------------- # 发布版本 # --------------------------------------------------------------------------- class NodeData(BaseModel): nodeKey: str = Field(..., description="节点唯一键") nodeType: str = Field(..., description="start/end/approval/condition") nodeName: Optional[str] = None configJson: Optional[str] = None class TransitionData(BaseModel): fromNodeKey: str toNodeKey: str conditionExpr: Optional[str] = None class PublishVersionRequest(BaseModel): graphJson: str = Field(..., description="AntV X6画布JSON快照") nodes: List[NodeData] = Field(default_factory=list) transitions: List[TransitionData] = Field(default_factory=list) remark: Optional[str] = None class VersionResponse(BaseModel): id: str versionNo: int isLatest: str remark: Optional[str] = None createTime: str updateTime: str # --------------------------------------------------------------------------- # 发起流程 # --------------------------------------------------------------------------- class StartInstanceRequest(BaseModel): processCode: str = Field(..., description="流程编码") businessType: Optional[str] = Field(None, description="业务类型") businessId: Optional[str] = Field(None, description="业务主键") variables: Optional[dict] = Field(default_factory=dict, description="流程变量") callbackUrl: Optional[str] = Field(None, description="业务回调地址") class InstanceResponse(BaseModel): id: str processVersionId: str businessType: Optional[str] = None businessId: Optional[str] = None starterUserId: Optional[str] = None status: Optional[str] = None createTime: str updateTime: str tasks: Optional[List[dict]] = None # --------------------------------------------------------------------------- # 任务 # --------------------------------------------------------------------------- class ApproveRequest(BaseModel): commentText: Optional[str] = Field(None, description="审批意见") class TaskResponse(BaseModel): id: str instanceId: str nodeKey: Optional[str] = None assigneeUserId: Optional[str] = None status: Optional[str] = None commentText: Optional[str] = None createTime: str updateTime: str # 关联信息（列表页用） processName: Optional[str] = None businessType: Optional[str] = None businessId: Optional[str] = None starterUserId: Optional[str] = None instanceStatus: Optional[str] = None instanceCreateTime: Optional[str] = None ``` - [ ] **Step 2: 验证schemas可导入** Run: `cd backend && python -c "from app.schemas.workflow import ProcessDefCreate, StartInstanceRequest; print('OK')"` Expected: OK --- ### Task 3: CRUD Operations **Files:** - Create: `backend/app/crud/workflow.py` - [ ] **Step 1: 创建workflow CRUD文件** 遵循项目现有CRUD模式（独立函数、`_row_to_dict`、`generate_uuid()`、手动字段映射）。 ```python """工作流 CRUD 操作""" from __future__ import annotations import json import math from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import and_, func, or_ from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.models.models import ( WfProcessDef, WfProcessVersion, WfNode, WfTransition, WfProcessInstance, WfTask, SysUser, ) # --------------------------------------------------------------------------- # 流程定义 # --------------------------------------------------------------------------- def _def_to_dict(t: WfProcessDef, latest_version: Optional[int] = None) -> Dict[str, Any]: return { "id": t.id, "processCode": t.process_code, "processName": t.process_name, "status": t.status, "latestVersion": latest_version, "tenantNo": t.tenant_no, "orderNum": t.order_num if t.order_num is not None else 0, "remark": t.remark or "", "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "", "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "", } def list_process_defs( db: Session, *, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, status: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: def _filtered(): q0 = db.query(WfProcessDef).filter(WfProcessDef.del_flag == 0) if tenant_no: q0 = q0.filter(WfProcessDef.tenant_no == tenant_no) if status: q0 = q0.filter(WfProcessDef.status == status) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter( or_(WfProcessDef.process_name.like(like), WfProcessDef.process_code.like(like)) ) return q0 total = _filtered().order_by(None).with_entities(func.count(WfProcessDef.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(WfProcessDef.order_num.asc(), WfProcessDef.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) result = [] for r in rows: latest = ( db.query(func.max(WfProcessVersion.version_no)) .filter( WfProcessVersion.process_def_id == r.id, WfProcessVersion.del_flag == 0, ) .scalar() ) result.append(_def_to_dict(r, latest_version=latest)) return result, int(total) def get_process_def(db: Session, def_id: str) -> Optional[WfProcessDef]: return ( db.query(WfProcessDef) .filter(and_(WfProcessDef.id == def_id, WfProcessDef.del_flag == 0)) .first() ) def process_code_exists( db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None ) -> bool: q = db.query(WfProcessDef).filter( and_(WfProcessDef.process_code == code.strip(), WfProcessDef.tenant_no == tenant_no, WfProcessDef.del_flag == 0) ) if exclude_id: q = q.filter(WfProcessDef.id != exclude_id) return q.first() is not None def create_process_def( db: Session, *, process_code: str, process_name: str, tenant_no: str, status: str = "1", order_num: int = 0, remark: Optional[str] = None, user_id: Optional[str] = None, ) -> WfProcessDef: code = (process_code or "").strip() if not code: raise ValueError("流程编码不能为空") if process_code_exists(db, code, tenant_no): raise ValueError("流程编码已存在") obj = WfProcessDef( id=generate_uuid(), process_code=code, process_name=(process_name or "").strip(), status=status, tenant_no=tenant_no, order_num=order_num, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def update_process_def( db: Session, def_id: str, *, process_code: Optional[str] = None, process_name: Optional[str] = None, status: Optional[str] = None, order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, tenant_no: Optional[str] = None, ) -> WfProcessDef: obj = get_process_def(db, def_id) if not obj: raise ValueError("流程定义不存在") if process_code is not None: new_code = process_code.strip() if not new_code: raise ValueError("流程编码不能为空") tn = tenant_no or obj.tenant_no if new_code != obj.process_code and process_code_exists(db, new_code, tn, exclude_id=def_id): raise ValueError("流程编码已存在") obj.process_code = new_code if process_name is not None: obj.process_name = (process_name or "").strip() if status is not None: obj.status = status if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = (remark or "").strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def soft_delete_process_def(db: Session, def_id: str) -> None: obj = get_process_def(db, def_id) if not obj: raise ValueError("流程定义不存在") obj.del_flag = 1 obj.update_time = datetime.now() db.add(obj) db.commit() # --------------------------------------------------------------------------- # 发布版本 # --------------------------------------------------------------------------- def publish_version( db: Session, *, process_def_id: str, graph_json: str, nodes: List[Dict[str, Any]], transitions: List[Dict[str, Any]], tenant_no: str, remark: Optional[str] = None, user_id: Optional[str] = None, ) -> WfProcessVersion: # 将旧版本 is_latest 置 0 db.query(WfProcessVersion).filter( WfProcessVersion.process_def_id == process_def_id, WfProcessVersion.is_latest == "1", WfProcessVersion.del_flag == 0, ).update({"is_latest": "0"}, synchronize_session=False) # 计算新版本号 max_ver = ( db.query(func.max(WfProcessVersion.version_no)) .filter(WfProcessVersion.process_def_id == process_def_id, WfProcessVersion.del_flag == 0) .scalar() ) or 0 new_ver = max_ver + 1 # 创建新版本 version_id = generate_uuid() version = WfProcessVersion( id=version_id, process_def_id=process_def_id, version_no=new_ver, graph_json=graph_json, is_latest="1", tenant_no=tenant_no, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(version) # 写入节点 for n in nodes: node = WfNode( id=generate_uuid(), process_version_id=version_id, node_key=n.get("nodeKey", ""), node_type=n.get("nodeType", ""), node_name=n.get("nodeName"), config_json=n.get("configJson"), tenant_no=tenant_no, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(node) # 写入连线 for t in transitions: trans = WfTransition( id=generate_uuid(), process_version_id=version_id, from_node_key=t.get("fromNodeKey", ""), to_node_key=t.get("toNodeKey", ""), condition_expr=t.get("conditionExpr"), tenant_no=tenant_no, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(trans) db.commit() db.refresh(version) return version def get_latest_version(db: Session, process_def_id: str) -> Optional[WfProcessVersion]: return ( db.query(WfProcessVersion) .filter( WfProcessVersion.process_def_id == process_def_id, WfProcessVersion.is_latest == "1", WfProcessVersion.del_flag == 0, ) .first() ) def list_versions( db: Session, process_def_id: str, tenant_no: Optional[str] = None ) -> List[Dict[str, Any]]: q = db.query(WfProcessVersion).filter( WfProcessVersion.process_def_id == process_def_id, WfProcessVersion.del_flag == 0, ) if tenant_no: q = q.filter(WfProcessVersion.tenant_no == tenant_no) rows = q.order_by(WfProcessVersion.version_no.desc()).all() return [ { "id": v.id, "versionNo": v.version_no, "isLatest": v.is_latest, "remark": v.remark or "", "createTime": v.create_time.strftime("%Y-%m-%d %H:%M:%S") if v.create_time else "", "updateTime": v.update_time.strftime("%Y-%m-%d %H:%M:%S") if v.update_time else "", } for v in rows ] # --------------------------------------------------------------------------- # 流程实例 # --------------------------------------------------------------------------- def create_instance( db: Session, *, process_version_id: str, business_type: Optional[str] = None, business_id: Optional[str] = None, starter_user_id: Optional[str] = None, variables: Optional[dict] = None, callback_url: Optional[str] = None, tenant_no: str = "zkxw", user_id: Optional[str] = None, ) -> WfProcessInstance: obj = WfProcessInstance( id=generate_uuid(), process_version_id=process_version_id, business_type=business_type, business_id=business_id, starter_user_id=starter_user_id, variables_json=json.dumps(variables or {}, ensure_ascii=False), status="running", callback_url=callback_url, tenant_no=tenant_no, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def get_instance(db: Session, instance_id: str) -> Optional[WfProcessInstance]: return ( db.query(WfProcessInstance) .filter(and_(WfProcessInstance.id == instance_id, WfProcessInstance.del_flag == 0)) .first() ) def update_instance_status(db: Session, instance_id: str, status: str) -> None: db.query(WfProcessInstance).filter(WfProcessInstance.id == instance_id).update( {"status": status, "update_time": datetime.now()}, synchronize_session=False ) db.commit() # --------------------------------------------------------------------------- # 任务 # --------------------------------------------------------------------------- def create_task( db: Session, *, instance_id: str, node_key: str, assignee_user_id: str, tenant_no: str = "zkxw", user_id: Optional[str] = None, ) -> WfTask: obj = WfTask( id=generate_uuid(), instance_id=instance_id, node_key=node_key, assignee_user_id=assignee_user_id, status="pending", tenant_no=tenant_no, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def get_task(db: Session, task_id: str) -> Optional[WfTask]: return ( db.query(WfTask) .filter(and_(WfTask.id == task_id, WfTask.del_flag == 0)) .first() ) def get_pending_tasks_by_node( db: Session, instance_id: str, node_key: str ) -> List[WfTask]: return ( db.query(WfTask) .filter( WfTask.instance_id == instance_id, WfTask.node_key == node_key, WfTask.status == "pending", WfTask.del_flag == 0, ) .all() ) def list_pending_tasks( db: Session, *, user_id: str, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: """我的待办列表""" q = ( db.query(WfTask, WfProcessInstance, WfProcessVersion, WfProcessDef) .join(WfProcessInstance, WfTask.instance_id == WfProcessInstance.id) .join(WfProcessVersion, WfProcessInstance.process_version_id == WfProcessVersion.id) .join(WfProcessDef, WfProcessVersion.process_def_id == WfProcessDef.id) .filter( WfTask.assignee_user_id == user_id, WfTask.status == "pending", WfTask.del_flag == 0, WfProcessInstance.del_flag == 0, ) ) if tenant_no: q = q.filter(WfTask.tenant_no == tenant_no) if keyword and keyword.strip(): like = f"%{keyword.strip()}%" q = q.filter(WfProcessDef.process_name.like(like)) total = q.count() page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q.order_by(WfTask.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) result = [] for task, inst, ver, pdef in rows: result.append({ "id": task.id, "instanceId": task.instance_id, "nodeKey": task.node_key, "assigneeUserId": task.assignee_user_id, "status": task.status, "createTime": task.create_time.strftime("%Y-%m-%d %H:%M:%S") if task.create_time else "", "updateTime": task.update_time.strftime("%Y-%m-%d %H:%M:%S") if task.update_time else "", "processName": pdef.process_name, "businessType": inst.business_type, "businessId": inst.business_id, "starterUserId": inst.starter_user_id, "instanceStatus": inst.status, "instanceCreateTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "", }) return result, int(total) def list_completed_tasks( db: Session, *, user_id: str, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: """我的已办列表""" q = ( db.query(WfTask, WfProcessInstance, WfProcessVersion, WfProcessDef) .join(WfProcessInstance, WfTask.instance_id == WfProcessInstance.id) .join(WfProcessVersion, WfProcessInstance.process_version_id == WfProcessVersion.id) .join(WfProcessDef, WfProcessVersion.process_def_id == WfProcessDef.id) .filter( WfTask.assignee_user_id == user_id, WfTask.status.in_(["approved", "rejected"]), WfTask.del_flag == 0, WfProcessInstance.del_flag == 0, ) ) if tenant_no: q = q.filter(WfTask.tenant_no == tenant_no) if keyword and keyword.strip(): like = f"%{keyword.strip()}%" q = q.filter(WfProcessDef.process_name.like(like)) total = q.count() page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q.order_by(WfTask.update_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) result = [] for task, inst, ver, pdef in rows: result.append({ "id": task.id, "instanceId": task.instance_id, "nodeKey": task.node_key, "assigneeUserId": task.assignee_user_id, "status": task.status, "commentText": task.comment_text, "createTime": task.create_time.strftime("%Y-%m-%d %H:%M:%S") if task.create_time else "", "updateTime": task.update_time.strftime("%Y-%m-%d %H:%M:%S") if task.update_time else "", "processName": pdef.process_name, "businessType": inst.business_type, "businessId": inst.business_id, "starterUserId": inst.starter_user_id, "instanceStatus": inst.status, "instanceCreateTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "", }) return result, int(total) # --------------------------------------------------------------------------- # 节点/连线查询 # --------------------------------------------------------------------------- def get_nodes_by_version(db: Session, version_id: str) -> List[WfNode]: return ( db.query(WfNode) .filter(WfNode.process_version_id == version_id, WfNode.del_flag == 0) .all() ) def get_transitions_by_version(db: Session, version_id: str) -> List[WfTransition]: return ( db.query(WfTransition) .filter(WfTransition.process_version_id == version_id, WfTransition.del_flag == 0) .all() ) def total_pages(total: int, page_size: int) -> int: if total <= 0: return 0 return int(math.ceil(total / float(page_size))) ``` - [ ] **Step 2: 验证CRUD可导入** Run: `cd backend && python -c "from app.crud.workflow import list_process_defs, publish_version; print('OK')"` Expected: OK --- ## Phase 2: Workflow Engine ### Task 4: Core Engine（节点分发 + Start/End Handler） **Files:** - Create: `backend/app/services/workflow_engine.py` - [ ] **Step 1: 创建工作流引擎** ```python """工作流引擎 - 图遍历模式""" from __future__ import annotations import json import logging import re from datetime import datetime from typing import Any, Dict, List, Optional from sqlalchemy.orm import Session from app.crud import workflow as wf_crud from app.models.models import WfNode, WfTransition, WfTask logger = logging.getLogger(__name__) class WorkflowEngine: """图遍历工作流引擎""" def __init__(self, db: Session): self.db = db # ------------------------------------------------------------------ # 公共入口 # ------------------------------------------------------------------ def start_process( self, *, process_version_id: str, business_type: Optional[str] = None, business_id: Optional[str] = None, starter_user_id: Optional[str] = None, variables: Optional[dict] = None, callback_url: Optional[str] = None, tenant_no: str = "zkxw", user_id: Optional[str] = None, ) -> str: """发起流程，返回实例ID""" instance = wf_crud.create_instance( self.db, process_version_id=process_version_id, business_type=business_type, business_id=business_id, starter_user_id=starter_user_id, variables=variables, callback_url=callback_url, tenant_no=tenant_no, user_id=user_id, ) # 找到 start 节点并推进 nodes = wf_crud.get_nodes_by_version(self.db, process_version_id) start_node = next((n for n in nodes if n.node_type == "start"), None) if start_node: self._advance(instance.id, start_node.node_key, tenant_no) return instance.id def approve_task(self, task_id: str, comment: Optional[str] = None, tenant_no: str = "zkxw") -> None: """审批通过任务""" task = wf_crud.get_task(self.db, task_id) if not task or task.status != "pending": raise ValueError("任务不存在或已处理") task.status = "approved" task.comment_text = comment task.update_time = datetime.now() self.db.add(task) self.db.commit() # 检查同节点是否全部通过 instance = wf_crud.get_instance(self.db, task.instance_id) if not instance: return nodes = wf_crud.get_nodes_by_version(self.db, instance.process_version_id) current_node = next((n for n in nodes if n.node_key == task.node_key), None) if not current_node: return config = json.loads(current_node.config_json or "{}") pass_type = config.get("pass_type", "any") pending_tasks = wf_crud.get_pending_tasks_by_node(self.db, task.instance_id, task.node_key) should_advance = False if pass_type == "any": # 任一通过即可推进，取消其余待办 should_advance = True for pt in pending_tasks: pt.status = "withdrawn" pt.update_time = datetime.now() self.db.add(pt) self.db.commit() else: # all: 全部通过才推进 if not pending_tasks: should_advance = True if should_advance: self._advance(task.instance_id, task.node_key, tenant_no) def reject_task(self, task_id: str, comment: Optional[str] = None, tenant_no: str = "zkxw") -> None: """审批驳回任务""" task = wf_crud.get_task(self.db, task_id) if not task or task.status != "pending": raise ValueError("任务不存在或已处理") task.status = "rejected" task.comment_text = comment task.update_time = datetime.now() self.db.add(task) self.db.commit() # 驳回 → 取消同节点其他待办 → 实例标记 rejected pending_tasks = wf_crud.get_pending_tasks_by_node(self.db, task.instance_id, task.node_key) for pt in pending_tasks: pt.status = "withdrawn" pt.update_time = datetime.now() self.db.add(pt) self.db.commit() wf_crud.update_instance_status(self.db, task.instance_id, "rejected") # 触发回调 instance = wf_crud.get_instance(self.db, task.instance_id) if instance: from app.services.workflow_callback import trigger_callback trigger_callback(self.db, instance, event="rejected") # ------------------------------------------------------------------ # 内部推进 # ------------------------------------------------------------------ def _advance(self, instance_id: str, from_node_key: str, tenant_no: str) -> None: """从指定节点推进到下一个节点""" instance = wf_crud.get_instance(self.db, instance_id) if not instance or instance.status != "running": return nodes = wf_crud.get_nodes_by_version(self.db, instance.process_version_id) transitions = wf_crud.get_transitions_by_version(self.db, instance.process_version_id) variables = json.loads(instance.variables_json or "{}") current_node = next((n for n in nodes if n.node_key == from_node_key), None) if not current_node: return # 找到出边 out_edges = [t for t in transitions if t.from_node_key == from_node_key] if not out_edges: return # 根据节点类型决定下一条边 next_node_key = None if current_node.node_type == "start": next_node_key = out_edges[0].to_node_key elif current_node.node_type == "approval": next_node_key = out_edges[0].to_node_key if out_edges else None elif current_node.node_type == "condition": next_node_key = self._evaluate_condition(out_edges, variables) else: return if not next_node_key: return # 执行目标节点 next_node = next((n for n in nodes if n.node_key == next_node_key), None) if not next_node: return self._execute_node(instance_id, next_node, variables, tenant_no) def _execute_node(self, instance_id: str, node: WfNode, variables: dict, tenant_no: str) -> None: """执行节点逻辑""" if node.node_type == "end": wf_crud.update_instance_status(self.db, instance_id, "completed") instance = wf_crud.get_instance(self.db, instance_id) if instance: from app.services.workflow_callback import trigger_callback trigger_callback(self.db, instance, event="completed") elif node.node_type == "approval": self._create_approval_tasks(instance_id, node, variables, tenant_no) elif node.node_type == "condition": self._advance(instance_id, node.node_key, tenant_no) else: # start 等其他类型直接推进 self._advance(instance_id, node.node_key, tenant_no) def _create_approval_tasks( self, instance_id: str, node: WfNode, variables: dict, tenant_no: str ) -> None: """审批节点：解析审批人并创建任务""" config = json.loads(node.config_json or "{}") user_ids = self._resolve_approvers(config, variables, tenant_no) for uid in user_ids: wf_crud.create_task( self.db, instance_id=instance_id, node_key=node.node_key, assignee_user_id=uid, tenant_no=tenant_no, ) def _resolve_approvers(self, config: dict, variables: dict, tenant_no: str) -> List[str]: """解析审批人列表""" strategy = config.get("strategy", "users") if strategy == "users": return config.get("user_ids", []) elif strategy == "roles": role_ids = config.get("role_ids", []) if not role_ids: return [] from app.models.models import SysUserRole, SysUser rows = ( self.db.query(SysUser.id) .join(SysUserRole, SysUser.id == SysUserRole.user_id) .filter( SysUserRole.role_id.in_(role_ids), SysUser.del_flag == 0, SysUser.status == "1", ) .distinct() .all() ) return [r[0] for r in rows] elif strategy == "sql": sql_template = config.get("sql", "") if not sql_template: return [] return self._execute_sql_approvers(sql_template, variables, tenant_no) return [] def _execute_sql_approvers(self, sql_template: str, variables: dict, tenant_no: str) -> List[str]: """执行SQL解析审批人（仅允许SELECT，参数化查询）""" # 安全检查：只允许 SELECT normalized = sql_template.strip().upper() for forbidden in ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "TRUNCATE"]: if forbidden in normalized: logger.warning(f"SQL approver resolution blocked: forbidden keyword '{forbidden}'") return [] if not normalized.startswith("SELECT"): return [] # 参数替换 params = dict(variables) params["tenant_no"] = tenant_no try: result = self.db.execute(sql_template, params) rows = result.fetchall() return [str(r[0]) for r in rows] except Exception as e: logger.error(f"SQL approver resolution failed: {e}") return [] def _evaluate_condition(self, out_edges: List[WfTransition], variables: dict) -> Optional[str]: """评估条件分支，返回目标节点key""" for edge in out_edges: if not edge.condition_expr: # 无条件 = 默认分支（最后一条） continue try: rules = json.loads(edge.condition_expr) if self._check_rules(rules, variables): return edge.to_node_key except (json.JSONDecodeError, KeyError): continue # 兜底：最后一条无条件边 for edge in reversed(out_edges): if not edge.condition_expr: return edge.to_node_key return out_edges[-1].to_node_key if out_edges else None def _check_rules(self, rules: Any, variables: dict) -> bool: """检查条件规则""" if isinstance(rules, dict): if "rules" in rules: # 顶层 rules 数组 return self._check_rules(rules["rules"], variables) if "and" in rules: return all(self._check_rules(r, variables) for r in rules["and"]) if "or" in rules: return any(self._check_rules(r, variables) for r in rules["or"]) # 单条规则 field = rules.get("field", "") op = rules.get("op", "==") value = rules.get("value") actual = self._get_variable(field, variables) return self._compare(actual, op, value) elif isinstance(rules, list): # 如果是数组，第一条匹配即返回（按顺序评估） for rule in rules: if self._check_rules(rule, variables): return True return False return False def _get_variable(self, field: str, variables: dict) -> Any: """从变量字典取值，支持点号路径""" parts = field.split(".") val = variables for p in parts: if isinstance(val, dict): val = val.get(p) else: return None return val def _compare(self, actual: Any, op: str, expected: Any) -> bool: """比较操作""" try: if op == "==": return actual == expected elif op == "!=": return actual != expected elif op == ">": return float(actual) > float(expected) elif op == "<": return float(actual) < float(expected) elif op == ">=": return float(actual) >= float(expected) elif op == "<=": return float(actual) <= float(expected) elif op == "in": return actual in expected if isinstance(expected, list) else False elif op == "not_in": return actual not in expected if isinstance(expected, list) else True elif op == "contains": return str(expected) in str(actual) if actual else False except (TypeError, ValueError): return False return False ``` - [ ] **Step 2: 验证引擎可导入** Run: `cd backend && python -c "from app.services.workflow_engine import WorkflowEngine; print('OK')"` Expected: OK --- ### Task 5: Callback Service **Files:** - Create: `backend/app/services/workflow_callback.py` - [ ] **Step 1: 创建回调服务** ```python """工作流回调服务""" from __future__ import annotations import json import logging from typing import Optional import requests from sqlalchemy.orm import Session from app.models.models import WfProcessInstance logger = logging.getLogger(__name__) CALLBACK_TIMEOUT_SECONDS = 10 def trigger_callback(db: Session, instance: WfProcessInstance, event: str) -> None: """触发业务回调""" if not instance.callback_url: return payload = { "event": event, "instanceId": instance.id, "businessType": instance.business_type, "businessId": instance.business_id, "status": instance.status, } try: resp = requests.post( instance.callback_url, json=payload, timeout=CALLBACK_TIMEOUT_SECONDS, headers={"Content-Type": "application/json"}, ) logger.info( f"Workflow callback triggered: {instance.callback_url} event={event} status={resp.status_code}" ) except Exception as e: logger.error(f"Workflow callback failed: {instance.callback_url} error={e}") ``` - [ ] **Step 2: 添加 requests 到 requirements.txt（如果缺失）** Run: `cd backend && grep -q "requests" requirements.txt && echo "EXISTS" || echo "requests" >> requirements.txt` --- ## Phase 3: API Routes ### Task 6: Workflow API Routes + Router Registration **Files:** - Create: `backend/app/api/workflow.py` - Modify: `backend/app/api/__init__.py` - Modify: `backend/app/main.py` - [ ] **Step 1: 创建workflow API路由文件** ```python """工作流管理 API 路由""" from typing import Optional from fastapi import APIRouter, Depends, Query from pydantic import BaseModel, ConfigDict, Field from sqlalchemy.orm import Session from app.core.database import get_db from app.core.menu_guard import require_admin_or_super_admin from app.core.response import success_response, error_response from app.crud import workflow as wf_crud from app.schemas.workflow import ( ProcessDefCreate, ProcessDefUpdate, PublishVersionRequest, StartInstanceRequest, ApproveRequest, ) from app.services.workflow_engine import WorkflowEngine router = APIRouter(prefix="/web/admin/workflow", tags=["工作流管理"]) # --------------------------------------------------------------------------- # 流程定义 # --------------------------------------------------------------------------- @router.get("/defs") async def list_defs( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100), keyword: Optional[str] = None, status: Optional[str] = None, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _user.get("tenant_no", "zkxw") rows, total = wf_crud.list_process_defs( db, page=page, page_size=pageSize, keyword=keyword, status=status, tenant_no=tenant_no ) return success_response({ "list": rows, "total": total, "page": page, "pageSize": pageSize, "totalPages": wf_crud.total_pages(total, pageSize), }) class _DefBody(BaseModel): model_config = ConfigDict(populate_by_name=True) processCode: str = Field(..., alias="processCode") processName: str = Field(..., alias="processName") status: Optional[str] = Field("1") orderNum: Optional[int] = Field(0) remark: Optional[str] = None @router.post("/defs") async def create_def( body: _DefBody, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _user.get("tenant_no", "zkxw") user_id = _user.get("user_id") or _user.get("id") try: obj = wf_crud.create_process_def( db, process_code=body.processCode, process_name=body.processName, tenant_no=tenant_no, status=body.status or "1", order_num=body.orderNum or 0, remark=body.remark, user_id=user_id, ) return success_response(wf_crud._def_to_dict(obj)) except ValueError as e: return error_response(str(e)) class _DefUpdateBody(BaseModel): model_config = ConfigDict(populate_by_name=True) processCode: Optional[str] = None processName: Optional[str] = None status: Optional[str] = None orderNum: Optional[int] = None remark: Optional[str] = None @router.put("/defs/{def_id}") async def update_def( def_id: str, body: _DefUpdateBody, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _user.get("tenant_no", "zkxw") user_id = _user.get("user_id") or _user.get("id") try: obj = wf_crud.update_process_def( db, def_id, process_code=body.processCode, process_name=body.processName, status=body.status, order_num=body.orderNum, remark=body.remark, user_id=user_id, tenant_no=tenant_no, ) return success_response(wf_crud._def_to_dict(obj)) except ValueError as e: return error_response(str(e)) @router.delete("/defs/{def_id}") async def delete_def( def_id: str, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): try: wf_crud.soft_delete_process_def(db, def_id) return success_response(None, msg="删除成功") except ValueError as e: return error_response(str(e)) # --------------------------------------------------------------------------- # 发布版本 # --------------------------------------------------------------------------- @router.post("/defs/{def_id}/publish") async def publish_version( def_id: str, body: PublishVersionRequest, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _user.get("tenant_no", "zkxw") user_id = _user.get("user_id") or _user.get("id") def_obj = wf_crud.get_process_def(db, def_id) if not def_obj: return error_response("流程定义不存在") nodes = [n.model_dump() for n in body.nodes] transitions = [t.model_dump() for t in body.transitions] version = wf_crud.publish_version( db, process_def_id=def_id, graph_json=body.graphJson, nodes=nodes, transitions=transitions, tenant_no=tenant_no, remark=body.remark, user_id=user_id, ) return success_response({ "id": version.id, "versionNo": version.version_no, "isLatest": version.is_latest, }) @router.get("/defs/{def_id}/versions") async def list_versions( def_id: str, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _user.get("tenant_no", "zkxw") rows = wf_crud.list_versions(db, def_id, tenant_no=tenant_no) return success_response(rows) # --------------------------------------------------------------------------- # 获取设计器数据（最新版本的 graph_json + nodes + transitions） # --------------------------------------------------------------------------- @router.get("/defs/{def_id}/designer") async def get_designer_data( def_id: str, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): version = wf_crud.get_latest_version(db, def_id) if not version: return success_response({"graphJson": None, "nodes": [], "transitions": []}) nodes = wf_crud.get_nodes_by_version(db, version.id) transitions = wf_crud.get_transitions_by_version(db, version.id) return success_response({ "graphJson": version.graph_json, "versionNo": version.version_no, "nodes": [ { "nodeKey": n.node_key, "nodeType": n.node_type, "nodeName": n.node_name, "configJson": n.config_json, } for n in nodes ], "transitions": [ { "fromNodeKey": t.from_node_key, "toNodeKey": t.to_node_key, "conditionExpr": t.condition_expr, } for t in transitions ], }) # --------------------------------------------------------------------------- # 流程实例 # --------------------------------------------------------------------------- @router.post("/instances") async def start_instance( body: StartInstanceRequest, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _user.get("tenant_no", "zkxw") user_id = _user.get("user_id") or _user.get("id") # 查找流程定义 all_defs = db.query(wf_crud.WfProcessDef).filter( wf_crud.WfProcessDef.process_code == body.processCode, wf_crud.WfProcessDef.del_flag == 0, wf_crud.WfProcessDef.tenant_no == tenant_no, ).first() if not all_defs: return error_response("流程定义不存在") if all_defs.status != "1": return error_response("流程已停用") # 查找最新版本 version = wf_crud.get_latest_version(db, all_defs.id) if not version: return error_response("流程尚未发布版本") engine = WorkflowEngine(db) instance_id = engine.start_process( process_version_id=version.id, business_type=body.businessType, business_id=body.businessId, starter_user_id=user_id, variables=body.variables, callback_url=body.callbackUrl, tenant_no=tenant_no, user_id=user_id, ) return success_response({"instanceId": instance_id}) @router.get("/instances/{instance_id}") async def get_instance_detail( instance_id: str, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): instance = wf_crud.get_instance(db, instance_id) if not instance: return error_response("流程实例不存在") tasks = ( db.query(wf_crud.WfTask) .filter(wf_crud.WfTask.instance_id == instance_id, wf_crud.WfTask.del_flag == 0) .order_by(wf_crud.WfTask.create_time.asc()) .all() ) return success_response({ "id": instance.id, "processVersionId": instance.process_version_id, "businessType": instance.business_type, "businessId": instance.business_id, "starterUserId": instance.starter_user_id, "status": instance.status, "variables": instance.variables_json, "createTime": instance.create_time.strftime("%Y-%m-%d %H:%M:%S") if instance.create_time else "", "tasks": [ { "id": t.id, "nodeKey": t.node_key, "assigneeUserId": t.assignee_user_id, "status": t.status, "commentText": t.comment_text, "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "", "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "", } for t in tasks ], }) # --------------------------------------------------------------------------- # 任务办理 # --------------------------------------------------------------------------- @router.get("/tasks/pending") async def list_pending( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100), keyword: Optional[str] = None, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): user_id = _user.get("user_id") or _user.get("id") tenant_no = _user.get("tenant_no", "zkxw") rows, total = wf_crud.list_pending_tasks( db, user_id=user_id, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no ) return success_response({ "list": rows, "total": total, "page": page, "pageSize": pageSize, "totalPages": wf_crud.total_pages(total, pageSize), }) @router.get("/tasks/completed") async def list_completed( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100), keyword: Optional[str] = None, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): user_id = _user.get("user_id") or _user.get("id") tenant_no = _user.get("tenant_no", "zkxw") rows, total = wf_crud.list_completed_tasks( db, user_id=user_id, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no ) return success_response({ "list": rows, "total": total, "page": page, "pageSize": pageSize, "totalPages": wf_crud.total_pages(total, pageSize), }) @router.post("/tasks/{task_id}/approve") async def approve_task( task_id: str, body: ApproveRequest, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _user.get("tenant_no", "zkxw") try: engine = WorkflowEngine(db) engine.approve_task(task_id, comment=body.commentText, tenant_no=tenant_no) return success_response(None, msg="审批通过") except ValueError as e: return error_response(str(e)) @router.post("/tasks/{task_id}/reject") async def reject_task( task_id: str, body: ApproveRequest, _user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _user.get("tenant_no", "zkxw") try: engine = WorkflowEngine(db) engine.reject_task(task_id, comment=body.commentText, tenant_no=tenant_no) return success_response(None, msg="审批驳回") except ValueError as e: return error_response(str(e)) ``` 注意：`workflow.py` 路由中 `start_instance` 端点直接引用了 `wf_crud.WfProcessDef` 和 `wf_crud.WfTask`，这是因为 CRUD 模块没有导出模型。更干净的做法是在路由文件顶部直接 import 模型： ```python from app.models.models import WfProcessDef, WfTask ``` 并将 `wf_crud.WfProcessDef` 改为 `WfProcessDef`，`wf_crud.WfTask` 改为 `WfTask`。 - [ ] **Step 2: 注册路由到 __init__.py** 在 `backend/app/api/__init__.py` 追加： ```python from app.api.workflow import router as workflow_router ``` 并在 `__all__` 中添加 `"workflow_router"`。 - [ ] **Step 3: 注册路由到 main.py** 在 `backend/app/main.py` 中找到 `app.include_router(admin_router)` 等注册行，追加： ```python from app.api.workflow import router as workflow_router app.include_router(workflow_router) ``` 或者如果 main.py 从 `__init__.py` 导入，则追加对应 include_router 调用。 - [ ] **Step 4: 验证路由加载** Run: `cd backend && python -c "from app.api.workflow import router; print(f'Routes: {len(router.routes)}')"` Expected: Routes: 13 --- ## Phase 4: Frontend Foundation ### Task 7: TypeScript Types + API Module **Files:** - Modify: `web-admin/src/types/api.d.ts`（追加 Workflow 命名空间） - Create: `web-admin/src/api/workflow.ts` - [ ] **Step 1: 在 api.d.ts 末尾追加 Workflow 类型** 在 `declare namespace Api {` 的最后一个 namespace 之后（即文件最后的 `}` 之前）追加： ```typescript namespace Workflow { interface ProcessDefItem { id: string processCode: string processName: string status: string latestVersion: number | null tenantNo: string orderNum: number remark: string createTime: string updateTime: string } interface VersionItem { id: string versionNo: number isLatest: string remark: string createTime: string updateTime: string } interface DesignerData { graphJson: string | null versionNo: number nodes: NodeData[] transitions: TransitionData[] } interface NodeData { nodeKey: string nodeType: string nodeName: string | null configJson: string | null } interface TransitionData { fromNodeKey: string toNodeKey: string conditionExpr: string | null } interface TaskItem { id: string instanceId: string nodeKey: string | null assigneeUserId: string | null status: string | null commentText: string | null createTime: string updateTime: string processName: string | null businessType: string | null businessId: string | null starterUserId: string | null instanceStatus: string | null instanceCreateTime: string | null } interface InstanceDetail { id: string processVersionId: string businessType: string | null businessId: string | null starterUserId: string | null status: string | null variables: string createTime: string tasks: TaskDetail[] } interface TaskDetail { id: string nodeKey: string | null assigneeUserId: string | null status: string | null commentText: string | null createTime: string updateTime: string } } ``` - [ ] **Step 2: 创建 workflow API 模块** Create: `web-admin/src/api/workflow.ts` ```typescript import request from '@/utils/http' // --------------------------------------------------------------------------- // 流程定义 // --------------------------------------------------------------------------- export function fetchProcessDefList(params?: { page?: number pageSize?: number keyword?: string status?: string }) { return request.get<{ list: Api.Workflow.ProcessDefItem[] total: number page: number pageSize: number totalPages: number }>({ url: '/web/admin/workflow/defs', params, }) } export function fetchCreateProcessDef(data: { processCode: string processName: string status?: string orderNum?: number remark?: string }) { return request.post({ url: '/web/admin/workflow/defs', data, showSuccessMessage: true, }) } export function fetchUpdateProcessDef( id: string, data: Partial<{ processCode: string processName: string status: string orderNum: number remark: string }>, ) { return request.put({ url: `/web/admin/workflow/defs/${id}`, data, showSuccessMessage: true, }) } export function fetchDeleteProcessDef(id: string) { return request.delete({ url: `/web/admin/workflow/defs/${id}`, showSuccessMessage: true, }) } // --------------------------------------------------------------------------- // 发布版本 // --------------------------------------------------------------------------- export function fetchPublishVersion( defId: string, data: { graphJson: string nodes: Array<{ nodeKey: string nodeType: string nodeName?: string | null configJson?: string | null }> transitions: Array<{ fromNodeKey: string toNodeKey: string conditionExpr?: string | null }> remark?: string }, ) { return request.post({ url: `/web/admin/workflow/defs/${defId}/publish`, data, showSuccessMessage: true, }) } export function fetchVersionList(defId: string) { return request.get({ url: `/web/admin/workflow/defs/${defId}/versions`, }) } // --------------------------------------------------------------------------- // 设计器数据 // --------------------------------------------------------------------------- export function fetchDesignerData(defId: string) { return request.get({ url: `/web/admin/workflow/defs/${defId}/designer`, }) } // --------------------------------------------------------------------------- // 流程实例 // --------------------------------------------------------------------------- export function fetchStartInstance(data: { processCode: string businessType?: string businessId?: string variables?: Record callbackUrl?: string }) { return request.post<{ instanceId: string }>({ url: '/web/admin/workflow/instances', data, }) } export function fetchInstanceDetail(instanceId: string) { return request.get({ url: `/web/admin/workflow/instances/${instanceId}`, }) } // --------------------------------------------------------------------------- // 任务办理 // --------------------------------------------------------------------------- export function fetchPendingTasks(params?: { page?: number pageSize?: number keyword?: string }) { return request.get<{ list: Api.Workflow.TaskItem[] total: number page: number pageSize: number totalPages: number }>({ url: '/web/admin/workflow/tasks/pending', params, }) } export function fetchCompletedTasks(params?: { page?: number pageSize?: number keyword?: string }) { return request.get<{ list: Api.Workflow.TaskItem[] total: number page: number pageSize: number totalPages: number }>({ url: '/web/admin/workflow/tasks/completed', params, }) } export function fetchApproveTask(taskId: string, data?: { commentText?: string }) { return request.post({ url: `/web/admin/workflow/tasks/${taskId}/approve`, data, showSuccessMessage: true, }) } export function fetchRejectTask(taskId: string, data?: { commentText?: string }) { return request.post({ url: `/web/admin/workflow/tasks/${taskId}/reject`, data, showSuccessMessage: true, }) } ``` - [ ] **Step 3: 验证前端编译** Run: `cd web-admin && npx vue-tsc --noEmit 2>&1 | head -20` Expected: 无新增类型错误（可能有预存错误） --- ### Task 8: Router Entries **Files:** - Modify: `web-admin/src/router/modules/sqxp.ts` - 菜单种子：见 `docs/sql/zkxw.sql` 内数据或 `docs/sql/` 增量脚本 - [ ] **Step 1: 在 sqxp.ts 中新增3个路由** 在 `sqxpAdminRoutes` 数组中，找到系统管理相关的 children，追加3个路由条目： ```typescript { id: '10000000000000000000000000000wf01', name: 'WorkflowProcess', path: 'workflow/process', component: '/system/workflow/process/index', meta: { title: '流程配置', keepAlive: true, icon: 'workflow' }, }, { id: '10000000000000000000000000000wf02', name: 'WorkflowPending', path: 'workflow/pending', component: '/system/workflow/task-pending/index', meta: { title: '待办中心', keepAlive: true }, }, { id: '10000000000000000000000000000wf03', name: 'WorkflowCompleted', path: 'workflow/completed', component: '/system/workflow/task-completed/index', meta: { title: '已办中心', keepAlive: true }, }, ``` - [ ] **Step 2: 创建菜单初始化SQL** 在 `docs/sql/zkxw.sql` 或增量脚本中维护菜单数据（示例）： ```sql -- 工作流管理菜单数据 -- 将流程配置、待办中心、已办中心添加到系统管理菜单下 -- 流程配置 INSERT INTO sys_menu (id, menu_name, parent_id, order_num, path, component, menu_type, visible, status, perms, icon, menu_code, tenant_no, create_time, update_time, del_flag) VALUES ('10000000000000000000000000000wf01', '流程配置', '0', 15, 'workflow/process', 'system/workflow/process/index', 'C', '0', '1', '', 'workflow', 'workflow:process', 'zkxw', NOW(), NOW(), 0); -- 待办中心 INSERT INTO sys_menu (id, menu_name, parent_id, order_num, path, component, menu_type, visible, status, perms, icon, menu_code, tenant_no, create_time, update_time, del_flag) VALUES ('10000000000000000000000000000wf02', '待办中心', '0', 16, 'workflow/pending', 'system/workflow/task-pending/index', 'C', '0', '1', '', 'bell', 'workflow:pending', 'zkxw', NOW(), NOW(), 0); -- 已办中心 INSERT INTO sys_menu (id, menu_name, parent_id, order_num, path, component, menu_type, visible, status, perms, icon, menu_code, tenant_no, create_time, update_time, del_flag) VALUES ('10000000000000000000000000000wf03', '已办中心', '0', 17, 'workflow/completed', 'system/workflow/task-completed/index', 'C', '0', '1', '', 'circle-check', 'workflow:completed', 'zkxw', NOW(), NOW(), 0); ``` --- ## Phase 5: Flow Designer ### Task 9: AntV X6 Flow Designer **Files:** - Create: `web-admin/src/views/system/workflow/designer/FlowDesigner.vue` **前置条件**: 安装 AntV X6 依赖 - [ ] **Step 1: 安装 AntV X6** Run: `cd web-admin && pnpm add @antv/x6 @antv/x6-plugin-selection @antv/x6-plugin-dnd @antv/x6-plugin-minimap @antv/x6-plugin-history` - [ ] **Step 2: 创建 FlowDesigner.vue** 这是一个大型组件，实现三栏布局（节点面板 / 画布 / 属性面板）。 关键实现要点： - 使用 `@antv/x6` 的 `Graph` 类初始化画布 - 注册4种自定义节点（Start/End/Approval/Condition）使用 `Graph.registerNode` - 左侧面板提供拖拽源，使用 `@antv/x6-plugin-dnd` 实现拖入画布 - 选中节点时右侧属性面板展示对应配置表单 - 保存时序列化为 `{graphJson, nodes, transitions}` 格式 - 发布调用 `fetchPublishVersion` API 组件 props： ```typescript const props = defineProps<{ visible: boolean processDefId: string processName: string }>() const emit = defineEmits<{ (e: 'update:visible', val: boolean): void (e: 'saved'): void }>() ``` 组件核心结构： ```vue ``` 节点类型定义： ```typescript const nodeTypes = [ { type: 'start', label: '开始节点', icon: '▶', color: '#10b981' }, { type: 'end', label: '结束节点', icon: '⏹', color: '#ef4444' }, { type: 'approval', label: '审批人节点', icon: '👤', color: '#3b82f6' }, { type: 'condition', label: '条件分支', icon: '◆', color: '#f59e0b' }, ] ``` 核心方法： ```typescript // 初始化画布 function initGraph() { const graph = new Graph({ container: canvasRef.value, grid: true, mousewheel: { enabled: true, modifiers: ['ctrl', 'meta'] }, connecting: { router: 'manhattan', connector: { name: 'rounded', args: { radius: 8 } }, anchor: 'center', connectionPoint: 'anchor', allowBlank: false, allowLoop: false, allowMulti: true, snap: { radius: 30 }, }, highlighting: { magnetAdsorbed: { name: 'stroke', args: { attrs: { fill: '#10b981', stroke: '#10b981' } } }, }, history: true, }) // 选中节点事件 graph.on('node:click', ({ node }) => { selectedNode.value = node }) graph.on('blank:click', () => { selectedNode.value = null }) graphRef.value = graph } // 序列化 function serializeGraph() { const graph = graphRef.value if (!graph) return null const cells = graph.getCells() const nodes = cells .filter((c) => c.isNode()) .map((n) => ({ nodeKey: n.id, nodeType: n.getData()?.nodeType || 'start', nodeName: n.getData()?.nodeName || '', configJson: JSON.stringify(n.getData()?.config || {}), })) const edges = cells .filter((c) => c.isEdge()) .map((e) => ({ fromNodeKey: (e.getSourceCellId() || ''), toNodeKey: (e.getTargetCellId() || ''), conditionExpr: e.getData()?.conditionExpr || null, })) return { graphJson: JSON.stringify(graph.toJSON()), nodes, transitions: edges, } } ``` - [ ] **Step 3: 验证设计器组件编译** Run: `cd web-admin && npx vue-tsc --noEmit 2>&1 | grep -i "FlowDesigner" || echo "No FlowDesigner errors"` Expected: 无 FlowDesigner 相关错误 --- ## Phase 6: Frontend Pages ### Task 10: Process Management Page **Files:** - Create: `web-admin/src/views/system/workflow/process/index.vue` - [ ] **Step 1: 创建流程配置管理页面** 参考 `unit/index.vue` 标准列表模板。功能：列表 + 新增/编辑弹窗 + 流程设计器弹窗。 列表字段：流程编码、流程名称、状态（启用/停用 switch）、最新版本、创建时间、操作（设计/编辑/删除）。 操作列宽度：4按钮（设计/编辑/停用启用/删除）= 400px。 关键交互： - "流程设计"按钮 → 打开 FlowDesigner 全屏弹窗 - "停用/启用"切换 → 调用 `fetchUpdateProcessDef` 修改 status - 新增/编辑弹窗使用 `.form-grid-2` + `label-position="top"` ### Task 11: Approval Dialog + Status Tag Components **Files:** - Create: `web-admin/src/views/system/workflow/components/ApprovalDialog.vue` - Create: `web-admin/src/views/system/workflow/components/ProcessStatusTag.vue` - [ ] **Step 1: 创建审批弹窗组件** ```vue ``` - [ ] **Step 2: 创建状态标签组件** ```vue ``` ### Task 12: Pending Tasks Page **Files:** - Create: `web-admin/src/views/system/workflow/task-pending/index.vue` - [ ] **Step 1: 创建待办中心页面** 参考 `order/index.vue` 复杂列表模板。 筛选：流程名称关键词、发起时间范围（el-date-picker type="daterange"）。 列表字段：流程名称、业务类型、业务单号、发起人、到达时间、操作。 操作列：审批（打开ApprovalDialog）、查看详情（展开行或弹窗显示实例详情含任务列表）。 操作列宽度：2按钮 = 260px。 ### Task 13: Completed Tasks Page **Files:** - Create: `web-admin/src/views/system/workflow/task-completed/index.vue` - [ ] **Step 1: 创建已办中心页面** 参考标准列表模板。 筛选：流程名称关键词、审批结果（通过/驳回 select）。 列表字段：流程名称、业务类型、业务单号、发起人、审批结果（ProcessStatusTag）、审批意见、审批时间。 操作列：查看详情。1按钮 = 260px。 --- ## Phase 7: Integration ### Task 14: App.vue Title Whitelist + Final Verification **Files:** - Modify: `web-admin/src/App.vue` - [ ] **Step 1: 在 App.vue panelTitleBarTextRules 中追加工作流标题** 在 `panelTitleBarTextRules` 数组中追加： ```typescript '流程配置', '待办列表', '已办列表', ``` - [ ] **Step 2: 运行后端验证** Run: `cd backend && python -c "from app.main import app; routes = [r.path for r in app.routes if hasattr(r, 'path')]; wf = [r for r in routes if 'workflow' in r]; print(f'Workflow routes: {len(wf)}'); [print(f' {r}') for r in wf]"` Expected: 13 workflow routes listed - [ ] **Step 3: 运行前端编译检查** Run: `cd web-admin && pnpm build 2>&1 | tail -5` Expected: 构建成功（或仅有预存错误） - [ ] **Step 4: Commit** ```bash git add backend/ web-admin/ docs/ git commit -m "feat: add workflow module (2.1.11) - process config, designer, task centers" ```