feat(workflow): add Excel export for process and task lists
Support filtered and selected rows for process defs, pending/completed tasks, and my applications. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
ee56dad875
commit
e72bcc67bc
6 files changed
+369
-4
No files matched your search
@@ -1,4 +1,5 @@
|
||||
"""工作流管理接口"""
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
@@ -12,6 +13,13 @@ 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=["工作流管理"])
|
||||
|
||||
@@ -30,6 +38,12 @@ def _business_type_label(code: Optional[str]) -> str:
|
||||
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 {
|
||||
@@ -189,6 +203,59 @@ def list_process_defs(
|
||||
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,
|
||||
@@ -379,6 +446,62 @@ def list_my_instances(
|
||||
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
||||
|
||||
|
||||
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,
|
||||
@@ -461,6 +584,60 @@ def list_pending_tasks(
|
||||
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),
|
||||
@@ -479,6 +656,62 @@ def list_completed_tasks(
|
||||
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))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 审批 / 驳回
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in new issue
Block a user