shanxia2.0/backend/app/api/workflow.py

"""工作流管理接口"""
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.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, 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)


# ---------------------------------------------------------------------------
# 业务表单字段提供者注册表
# 新增业务类型时，只需添加一个 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")
async def list_process_defs(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, 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")
async 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}")
async 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}")
async 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)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))


# ---------------------------------------------------------------------------
# 版本发布 / 查看
# ---------------------------------------------------------------------------


@router.post("/defs/{def_id}/publish")
async 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")
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", "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")
async 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")
async 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")
async def list_my_instances(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, 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}")
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(404, "流程实例不存在")
    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")
async def list_pending_tasks(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, 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")
async def list_completed_tasks(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, 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")
async def approve_task(
    task_id: str,
    body: TaskActionBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        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")
async def reject_task(
    task_id: str,
    body: TaskActionBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        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}")
async def get_business_all_tasks(
    business_type: str,
    business_id: str,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    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,
    })