864 lines
31 KiB
Python
864 lines
31 KiB
Python
"""工作流管理接口"""
|
|
from datetime import datetime
|
|
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
|
|
from app.utils.excel_export import (
|
|
EXPORT_MAX_ROWS,
|
|
ExportColumn,
|
|
ExportValidationError,
|
|
export_rows_to_xlsx_response,
|
|
filter_export_columns,
|
|
)
|
|
|
|
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 _fmt_latest_version(v: Any) -> str:
|
|
if v is None or v == "":
|
|
return "--"
|
|
return f"V{v}"
|
|
|
|
|
|
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})
|
|
|
|
|
|
PROCESS_DEF_EXPORT_COLUMNS = [
|
|
ExportColumn(title="流程编码", field="processCode"),
|
|
ExportColumn(title="流程名称", field="processName"),
|
|
ExportColumn(title="状态", field="status", dict_type="sys_status"),
|
|
ExportColumn(title="最新版本", field="latestVersion", formatter=_fmt_latest_version),
|
|
ExportColumn(title="创建时间", field="createTime"),
|
|
]
|
|
|
|
|
|
class ProcessDefExportBody(BaseModel):
|
|
keyword: Optional[str] = None
|
|
status: Optional[str] = None
|
|
ids: Optional[List[str]] = None
|
|
columns: Optional[List[str]] = None
|
|
|
|
|
|
@router.post("/defs/export")
|
|
def export_process_defs(
|
|
body: ProcessDefExportBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""导出流程定义列表（当前筛选全量，或 ids 已选）。"""
|
|
try:
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
items, _ = wf_crud.list_process_defs(
|
|
db,
|
|
page=1,
|
|
page_size=EXPORT_MAX_ROWS,
|
|
keyword=body.keyword,
|
|
status=body.status,
|
|
tenant_no=tenant_no,
|
|
user_info=user,
|
|
)
|
|
if body.ids:
|
|
idset = set(body.ids)
|
|
items = [r for r in items if r.get("id") in idset]
|
|
cols = filter_export_columns(PROCESS_DEF_EXPORT_COLUMNS, body.columns)
|
|
date_str = datetime.now().strftime("%Y%m%d")
|
|
return export_rows_to_xlsx_response(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
sheet_title="流程定义",
|
|
columns=cols,
|
|
rows=items,
|
|
filename_stem=f"流程定义_{date_str}",
|
|
)
|
|
except ExportValidationError as e:
|
|
return error_response(400, str(e))
|
|
except Exception as e:
|
|
return error_response(500, str(e))
|
|
|
|
|
|
@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,
|
|
processNameIn: Optional[str] = Query(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, process_name_in=processNameIn,
|
|
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})
|
|
|
|
|
|
class WorkflowColumnFacetBody(BaseModel):
|
|
"""工作流任务/申请列表列头可选值（去重；可附带列表上下文）"""
|
|
|
|
model_config = ConfigDict(populate_by_name=True, extra="allow")
|
|
|
|
scope: str = Field(..., description="pending | completed | mine")
|
|
keyword: Optional[str] = Field(default=None, description="对应 scope 的列表关键字")
|
|
limit: Optional[int] = Field(default=500, ge=1, le=500, description="返回条数上限")
|
|
|
|
|
|
@router.post("/column-facet-values")
|
|
def workflow_column_facet_values(
|
|
body: WorkflowColumnFacetBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""任务/申请列表列头可选值：在列表上下文（排除本列 processName）下去重。"""
|
|
from app.services.pl.facet_context import strip_simple_facet_self_params
|
|
|
|
if body.scope not in ("pending", "completed", "mine"):
|
|
return error_response(400, "不支持的 scope")
|
|
try:
|
|
user_id = user.get("user_id") or user.get("id")
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
data = body.model_dump(by_alias=True, exclude_none=True)
|
|
extra = getattr(body, "model_extra", None) or {}
|
|
for key, value in extra.items():
|
|
if key not in data and value not in (None, ""):
|
|
data[key] = value
|
|
list_keyword = (body.keyword or "").strip() or None
|
|
params = strip_simple_facet_self_params(data, "processName")
|
|
status_in = params.get("statusIn") or params.get("status_in")
|
|
status = params.get("status")
|
|
rows = wf_crud.facet_process_names(
|
|
db,
|
|
user_id,
|
|
scope=body.scope,
|
|
keyword=list_keyword or str(params.get("keyword") or "").strip() or None,
|
|
limit=body.limit if body.limit is not None else 500,
|
|
tenant_no=tenant_no,
|
|
user_info=user,
|
|
status=status,
|
|
status_in=status_in,
|
|
)
|
|
return success_response({"list": rows})
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
except Exception as e:
|
|
return error_response(500, str(e))
|
|
|
|
|
|
MY_INSTANCE_EXPORT_COLUMNS = [
|
|
ExportColumn(title="流程名称", field="processName"),
|
|
ExportColumn(title="业务类型", field="businessTypeLabel"),
|
|
ExportColumn(title="业务单号", field="businessId"),
|
|
ExportColumn(title="状态", field="status", dict_type="wf_process_status"),
|
|
ExportColumn(title="发起时间", field="createTime"),
|
|
]
|
|
|
|
|
|
class MyInstanceExportBody(BaseModel):
|
|
keyword: Optional[str] = None
|
|
status: Optional[str] = None
|
|
ids: Optional[List[str]] = None
|
|
columns: Optional[List[str]] = None
|
|
|
|
|
|
@router.post("/instances/my/export")
|
|
def export_my_instances(
|
|
body: MyInstanceExportBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""导出我的申请列表（当前筛选全量，或 ids 已选）。"""
|
|
try:
|
|
user_id = user.get("user_id") or user.get("id")
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
items, _ = wf_crud.list_my_instances(
|
|
db,
|
|
user_id,
|
|
page=1,
|
|
page_size=EXPORT_MAX_ROWS,
|
|
keyword=body.keyword,
|
|
status=body.status,
|
|
tenant_no=tenant_no,
|
|
user_info=user,
|
|
)
|
|
_enrich_task_list(db, items)
|
|
if body.ids:
|
|
idset = set(body.ids)
|
|
items = [r for r in items if r.get("id") in idset]
|
|
cols = filter_export_columns(MY_INSTANCE_EXPORT_COLUMNS, body.columns)
|
|
date_str = datetime.now().strftime("%Y%m%d")
|
|
return export_rows_to_xlsx_response(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
sheet_title="我的申请",
|
|
columns=cols,
|
|
rows=items,
|
|
filename_stem=f"我的申请_{date_str}",
|
|
)
|
|
except ExportValidationError as e:
|
|
return error_response(400, str(e))
|
|
except Exception as e:
|
|
return error_response(500, str(e))
|
|
|
|
|
|
@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,
|
|
processNameIn: Optional[str] = Query(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,
|
|
process_name_in=processNameIn,
|
|
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})
|
|
|
|
|
|
PENDING_TASK_EXPORT_COLUMNS = [
|
|
ExportColumn(title="流程名称", field="processName"),
|
|
ExportColumn(title="业务类型", field="businessTypeLabel"),
|
|
ExportColumn(title="业务单号", field="businessId"),
|
|
ExportColumn(title="发起人", field="starterUserName"),
|
|
ExportColumn(title="到达时间", field="createTime"),
|
|
]
|
|
|
|
|
|
class PendingTaskExportBody(BaseModel):
|
|
keyword: Optional[str] = None
|
|
ids: Optional[List[str]] = None
|
|
columns: Optional[List[str]] = None
|
|
|
|
|
|
@router.post("/tasks/pending/export")
|
|
def export_pending_tasks(
|
|
body: PendingTaskExportBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""导出待办任务列表（当前筛选全量，或 ids 已选）。"""
|
|
try:
|
|
user_id = user.get("user_id") or user.get("id")
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
items, _ = wf_crud.list_pending_tasks(
|
|
db,
|
|
user_id,
|
|
page=1,
|
|
page_size=EXPORT_MAX_ROWS,
|
|
keyword=body.keyword,
|
|
tenant_no=tenant_no,
|
|
user_info=user,
|
|
)
|
|
_enrich_task_list(db, items)
|
|
if body.ids:
|
|
idset = set(body.ids)
|
|
items = [r for r in items if r.get("id") in idset]
|
|
cols = filter_export_columns(PENDING_TASK_EXPORT_COLUMNS, body.columns)
|
|
date_str = datetime.now().strftime("%Y%m%d")
|
|
return export_rows_to_xlsx_response(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
sheet_title="待办任务",
|
|
columns=cols,
|
|
rows=items,
|
|
filename_stem=f"待办任务_{date_str}",
|
|
)
|
|
except ExportValidationError as e:
|
|
return error_response(400, str(e))
|
|
except Exception as e:
|
|
return error_response(500, str(e))
|
|
|
|
|
|
@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,
|
|
processNameIn: Optional[str] = Query(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,
|
|
process_name_in=processNameIn,
|
|
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})
|
|
|
|
|
|
COMPLETED_TASK_EXPORT_COLUMNS = [
|
|
ExportColumn(title="流程名称", field="processName"),
|
|
ExportColumn(title="业务类型", field="businessTypeLabel"),
|
|
ExportColumn(title="业务单号", field="businessId"),
|
|
ExportColumn(title="发起人", field="starterUserName"),
|
|
ExportColumn(title="审批结果", field="status", dict_type="wf_process_status"),
|
|
ExportColumn(title="审批意见", field="commentText"),
|
|
ExportColumn(title="审批时间", field="updateTime"),
|
|
]
|
|
|
|
|
|
class CompletedTaskExportBody(BaseModel):
|
|
keyword: Optional[str] = None
|
|
ids: Optional[List[str]] = None
|
|
columns: Optional[List[str]] = None
|
|
|
|
|
|
@router.post("/tasks/completed/export")
|
|
def export_completed_tasks(
|
|
body: CompletedTaskExportBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""导出已办任务列表（当前筛选全量，或 ids 已选）。"""
|
|
try:
|
|
user_id = user.get("user_id") or user.get("id")
|
|
tenant_no = user.get("tenant_no", "shanxia2.0")
|
|
items, _ = wf_crud.list_completed_tasks(
|
|
db,
|
|
user_id,
|
|
page=1,
|
|
page_size=EXPORT_MAX_ROWS,
|
|
keyword=body.keyword,
|
|
tenant_no=tenant_no,
|
|
user_info=user,
|
|
)
|
|
_enrich_task_list(db, items)
|
|
if body.ids:
|
|
idset = set(body.ids)
|
|
items = [r for r in items if r.get("id") in idset]
|
|
cols = filter_export_columns(COMPLETED_TASK_EXPORT_COLUMNS, body.columns)
|
|
date_str = datetime.now().strftime("%Y%m%d")
|
|
return export_rows_to_xlsx_response(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
sheet_title="已办任务",
|
|
columns=cols,
|
|
rows=items,
|
|
filename_stem=f"已办任务_{date_str}",
|
|
)
|
|
except ExportValidationError as e:
|
|
return error_response(400, str(e))
|
|
except Exception as e:
|
|
return error_response(500, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 审批 / 驳回
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@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,
|
|
})
|