572 lines
21 KiB
Python
572 lines
21 KiB
Python
"""工作流管理接口"""
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import AliasChoices, BaseModel, ConfigDict, Field
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.data_scope import assert_row_in_data_scope
|
|
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.models.models import SysUser, WfNode, WfProcessDef, WfProcessInstance, WfProcessVersion, WfTask
|
|
from app.services.workflow_engine import WorkflowEngine
|
|
|
|
router = APIRouter(prefix="/web/admin/workflow", tags=["工作流管理"])
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 业务类型 → 中文名称映射
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_BUSINESS_TYPE_LABELS: Dict[str, str] = {
|
|
"leave": "请假申请",
|
|
}
|
|
|
|
|
|
def _business_type_label(code: Optional[str]) -> str:
|
|
if not code:
|
|
return ""
|
|
return _BUSINESS_TYPE_LABELS.get(code, code)
|
|
|
|
|
|
def _portal_user_info(user: dict) -> dict:
|
|
dept_id = (user.get("region_id") or user.get("dept_id") or "").strip()
|
|
return {
|
|
"user_id": (user.get("user_id") or user.get("id") or "").strip(),
|
|
"role": (user.get("role") or "").strip(),
|
|
"tenant_no": (user.get("tenant_no") or "").strip(),
|
|
"dept_id": dept_id,
|
|
"region_id": dept_id,
|
|
}
|
|
|
|
|
|
def _assert_wf_instance_in_scope(
|
|
db: Session,
|
|
instance: WfProcessInstance,
|
|
user: dict,
|
|
*,
|
|
skip_scope: bool = False,
|
|
) -> None:
|
|
if skip_scope:
|
|
return
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
assert_row_in_data_scope(db, instance, WfProcessInstance, _portal_user_info(user), tenant_no)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 业务表单字段提供者注册表
|
|
# 新增业务类型时，只需添加一个 provider 函数并注册即可，前端无需改动
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import logging
|
|
_form_field_logger = logging.getLogger("workflow.form_fields")
|
|
|
|
|
|
def _leave_form_fields(db: Session, business_id: str) -> List[Dict[str, str]]:
|
|
from app.crud import leave as leave_crud
|
|
item = leave_crud.get_leave(db, business_id)
|
|
if not item:
|
|
return []
|
|
leave_type_map = {"personal": "事假", "sick": "病假", "annual": "年假", "other": "其他"}
|
|
fields = [
|
|
{"label": "请假类型", "value": leave_type_map.get(item.get("leaveType", ""), item.get("leaveType", ""))},
|
|
{"label": "请假天数", "value": f"{item.get('days', 0)} 天" if item.get("days") else "--"},
|
|
{"label": "开始日期", "value": item.get("startDate", "") or "--"},
|
|
{"label": "结束日期", "value": item.get("endDate", "") or "--"},
|
|
]
|
|
if item.get("reason"):
|
|
fields.append({"label": "请假原因", "value": item["reason"]})
|
|
if item.get("remark"):
|
|
fields.append({"label": "备注", "value": item["remark"]})
|
|
return fields
|
|
|
|
|
|
_BUSINESS_FORM_FIELD_PROVIDERS: Dict[str, Any] = {
|
|
"leave": _leave_form_fields,
|
|
}
|
|
|
|
|
|
def _get_business_form_fields(db: Session, business_type: Optional[str], business_id: Optional[str]) -> List[Dict[str, str]]:
|
|
if not business_type or not business_id:
|
|
return []
|
|
provider = _BUSINESS_FORM_FIELD_PROVIDERS.get(business_type)
|
|
if not provider:
|
|
return []
|
|
try:
|
|
return provider(db, business_id)
|
|
except Exception as e:
|
|
_form_field_logger.warning("加载业务表单字段失败: type=%s id=%s err=%s", business_type, business_id, e)
|
|
return []
|
|
|
|
|
|
def _enrich_task_list(db: Session, items: List[Dict[str, Any]]) -> None:
|
|
starter_ids = {it.get("starterUserId") for it in items if it.get("starterUserId")}
|
|
user_name_map: Dict[str, str] = {}
|
|
if starter_ids:
|
|
rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(starter_ids)).all()
|
|
user_name_map = {r.id: r.nick_name for r in rows}
|
|
for it in items:
|
|
sid = it.get("starterUserId")
|
|
it["starterUserName"] = user_name_map.get(sid, sid or "")
|
|
it["businessTypeLabel"] = _business_type_label(it.get("businessType"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pydantic request bodies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ProcessDefCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
processCode: str = Field(..., alias="processCode")
|
|
processName: str = Field(..., alias="processName")
|
|
status: Optional[str] = Field("1", alias="status")
|
|
orderNum: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ProcessDefUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
processCode: Optional[str] = Field(None, alias="processCode")
|
|
processName: Optional[str] = Field(None, alias="processName")
|
|
status: Optional[str] = Field(None, alias="status")
|
|
orderNum: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class PublishVersionBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
graphJson: Optional[str] = Field(None, alias="graphJson")
|
|
nodes: List[Dict[str, Any]] = Field(default_factory=list, alias="nodes")
|
|
transitions: List[Dict[str, Any]] = Field(default_factory=list, alias="transitions")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class StartInstanceBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
processCode: str = Field(..., alias="processCode")
|
|
businessType: Optional[str] = Field(None, alias="businessType")
|
|
businessId: Optional[str] = Field(None, alias="businessId")
|
|
variables: Optional[Dict[str, Any]] = None
|
|
callbackUrl: Optional[str] = Field(None, alias="callbackUrl")
|
|
|
|
|
|
class TaskActionBody(BaseModel):
|
|
"""审批意见：前端传 commentText，兼容 comment。"""
|
|
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
comment: Optional[str] = Field(
|
|
default=None,
|
|
validation_alias=AliasChoices("comment", "commentText"),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 流程定义 CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/defs")
|
|
def list_process_defs(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=500, alias="pageSize"),
|
|
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", "shanxia2.0")
|
|
items, total = wf_crud.list_process_defs(
|
|
db, page=page, page_size=pageSize, keyword=keyword, status=status, tenant_no=tenant_no, user_info=_user
|
|
)
|
|
tp = wf_crud.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.post("/defs")
|
|
def create_process_def(
|
|
body: ProcessDefCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
obj = wf_crud.create_process_def(
|
|
db,
|
|
process_code=body.processCode,
|
|
process_name=body.processName,
|
|
tenant_no=tenant_no,
|
|
status=body.status,
|
|
order_num=body.orderNum,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(wf_crud._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/defs/{def_id}")
|
|
def update_process_def(
|
|
def_id: str,
|
|
body: ProcessDefUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
payload: Dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=True)
|
|
obj = wf_crud.update_process_def(
|
|
db,
|
|
def_id,
|
|
process_code=payload.get("processCode"),
|
|
process_name=payload.get("processName"),
|
|
status=payload.get("status"),
|
|
order_num=payload.get("orderNum"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
tenant_no=user.get("tenant_no", "shanxia2.0"),
|
|
)
|
|
return success_response(wf_crud._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/defs/{def_id}")
|
|
def delete_process_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, user_info=_user)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 版本发布 / 查看
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/defs/{def_id}/publish")
|
|
def publish_version(
|
|
def_id: str,
|
|
body: PublishVersionBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
ver = wf_crud.publish_version(
|
|
db,
|
|
process_def_id=def_id,
|
|
graph_json=body.graphJson or "",
|
|
nodes=body.nodes,
|
|
transitions=body.transitions,
|
|
tenant_no=tenant_no,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response({"versionId": ver.id, "versionNo": ver.version_no}, message="发布成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.get("/defs/{def_id}/versions")
|
|
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", "shanxia2.0")
|
|
items = wf_crud.list_versions(db, def_id, tenant_no=tenant_no, user_info=_user)
|
|
return success_response(items)
|
|
|
|
|
|
@router.get("/defs/{def_id}/designer")
|
|
def get_designer_data(
|
|
def_id: str,
|
|
_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
ver = wf_crud.get_latest_version(db, def_id)
|
|
if not ver:
|
|
return success_response({"graphJson": None, "nodes": [], "transitions": []})
|
|
nodes = wf_crud.get_nodes_by_version(db, ver.id)
|
|
transitions = wf_crud.get_transitions_by_version(db, ver.id)
|
|
return success_response({
|
|
"graphJson": ver.graph_json,
|
|
"versionId": ver.id,
|
|
"versionNo": ver.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")
|
|
def start_instance(
|
|
body: StartInstanceBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
user_id = user.get("user_id") or user.get("id")
|
|
defn = (
|
|
db.query(WfProcessDef)
|
|
.filter(
|
|
WfProcessDef.process_code == body.processCode,
|
|
WfProcessDef.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not defn:
|
|
return error_response(404, "流程定义不存在")
|
|
engine = WorkflowEngine(db)
|
|
instance = engine.start_process(
|
|
process_def_id=defn.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,
|
|
"status": instance.status,
|
|
}, message="发起成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.get("/instances/my")
|
|
def list_my_instances(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=500, alias="pageSize"),
|
|
keyword: Optional[str] = None,
|
|
status: 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", "shanxia2.0")
|
|
items, total = wf_crud.list_my_instances(
|
|
db, user_id, page=page, page_size=pageSize,
|
|
keyword=keyword, status=status, tenant_no=tenant_no, user_info=user,
|
|
)
|
|
_enrich_task_list(db, items)
|
|
tp = wf_crud.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/instances/{instance_id}")
|
|
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(404, "流程实例不存在")
|
|
_assert_wf_instance_in_scope(db, instance, user)
|
|
tasks = db.query(WfTask).filter(
|
|
WfTask.instance_id == instance_id,
|
|
WfTask.del_flag == 0,
|
|
).order_by(WfTask.create_time.asc()).all()
|
|
|
|
# 查询节点名称映射 node_key → node_name
|
|
node_name_map: Dict[str, str] = {}
|
|
if instance.process_version_id:
|
|
for n in wf_crud.get_nodes_by_version(db, instance.process_version_id):
|
|
node_name_map[n.node_key] = n.node_name or n.node_key
|
|
|
|
# 批量查询用户名
|
|
user_ids = {t.assignee_user_id for t in tasks if t.assignee_user_id}
|
|
if instance.starter_user_id:
|
|
user_ids.add(instance.starter_user_id)
|
|
user_name_map: Dict[str, str] = {}
|
|
if user_ids:
|
|
rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(user_ids)).all()
|
|
user_name_map = {r.id: r.nick_name for r in rows}
|
|
|
|
return success_response({
|
|
"id": instance.id,
|
|
"businessType": instance.business_type,
|
|
"businessTypeLabel": _business_type_label(instance.business_type),
|
|
"businessId": instance.business_id,
|
|
"businessFormFields": _get_business_form_fields(db, instance.business_type, instance.business_id),
|
|
"starterUserId": instance.starter_user_id,
|
|
"starterUserName": user_name_map.get(instance.starter_user_id, instance.starter_user_id or ""),
|
|
"status": instance.status,
|
|
"variablesJson": instance.variables_json,
|
|
"callbackUrl": instance.callback_url,
|
|
"createTime": instance.create_time.strftime("%Y-%m-%d %H:%M:%S") if instance.create_time else "",
|
|
"tasks": [
|
|
{
|
|
"id": t.id,
|
|
"nodeKey": t.node_key,
|
|
"nodeName": node_name_map.get(t.node_key, t.node_key or ""),
|
|
"assigneeUserId": t.assignee_user_id,
|
|
"assigneeUserName": user_name_map.get(t.assignee_user_id, t.assignee_user_id or ""),
|
|
"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")
|
|
def list_pending_tasks(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=500, alias="pageSize"),
|
|
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", "shanxia2.0")
|
|
items, total = wf_crud.list_pending_tasks(
|
|
db, user_id, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user
|
|
)
|
|
_enrich_task_list(db, items)
|
|
tp = wf_crud.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.get("/tasks/completed")
|
|
def list_completed_tasks(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=500, alias="pageSize"),
|
|
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", "shanxia2.0")
|
|
items, total = wf_crud.list_completed_tasks(
|
|
db, user_id, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user
|
|
)
|
|
_enrich_task_list(db, items)
|
|
tp = wf_crud.total_pages(total, pageSize)
|
|
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 审批 / 驳回
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/tasks/{task_id}/approve")
|
|
def approve_task(
|
|
task_id: str,
|
|
body: TaskActionBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
task = wf_crud.get_task(db, task_id)
|
|
if not task:
|
|
return error_response(404, "任务不存在")
|
|
instance = wf_crud.get_instance(db, task.instance_id)
|
|
if not instance:
|
|
return error_response(404, "流程实例不存在")
|
|
user_id = (user.get("user_id") or user.get("id") or "").strip()
|
|
skip_scope = task.assignee_user_id == user_id
|
|
_assert_wf_instance_in_scope(db, instance, user, skip_scope=skip_scope)
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
engine = WorkflowEngine(db)
|
|
instance = engine.approve_task(task_id, body.comment or "", tenant_no)
|
|
return success_response({"instanceId": instance.id, "status": instance.status}, message="审批成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/tasks/{task_id}/reject")
|
|
def reject_task(
|
|
task_id: str,
|
|
body: TaskActionBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
task = wf_crud.get_task(db, task_id)
|
|
if not task:
|
|
return error_response(404, "任务不存在")
|
|
instance = wf_crud.get_instance(db, task.instance_id)
|
|
if not instance:
|
|
return error_response(404, "流程实例不存在")
|
|
user_id = (user.get("user_id") or user.get("id") or "").strip()
|
|
skip_scope = task.assignee_user_id == user_id
|
|
_assert_wf_instance_in_scope(db, instance, user, skip_scope=skip_scope)
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
engine = WorkflowEngine(db)
|
|
instance = engine.reject_task(task_id, body.comment or "", tenant_no)
|
|
return success_response({"instanceId": instance.id, "status": instance.status}, message="已驳回")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 按业务查询所有审批历史（跨实例）
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/instances/by-business/{business_type}/{business_id}")
|
|
def get_business_all_tasks(
|
|
business_type: str,
|
|
business_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
instances = (
|
|
db.query(WfProcessInstance)
|
|
.filter(
|
|
WfProcessInstance.business_type == business_type,
|
|
WfProcessInstance.business_id == business_id,
|
|
WfProcessInstance.del_flag == 0,
|
|
)
|
|
.order_by(WfProcessInstance.create_time.asc())
|
|
.all()
|
|
)
|
|
if instances:
|
|
_assert_wf_instance_in_scope(db, instances[0], user)
|
|
|
|
tasks = wf_crud.get_all_tasks_for_business(db, business_type, business_id)
|
|
|
|
business_form_fields = _get_business_form_fields(db, business_type, business_id)
|
|
|
|
return success_response({
|
|
"businessType": business_type,
|
|
"businessId": business_id,
|
|
"businessFormFields": business_form_fields,
|
|
"tasks": tasks,
|
|
})
|