shanxia2.0/backend/app/services/bx/workflow.py

from __future__ import annotations

from typing import List, Optional, Sequence, Tuple

from sqlalchemy.orm import Session

from app.core.bx_auth import BxUserContext
from app.core.tenant_context import get_tenant_no
from app.core.utils import generate_uuid
from app.models.bx_models import BxWorkflowDef, BxWorkflowNode
from app.services.bx.base import BxCrudService


_DEFAULT_WORKFLOWS = {
    "advance": {
        "wf_name": "备用金审批流程",
        "remark": "备用金申请默认审批流程",
    },
    "reimburse": {
        "wf_name": "报销单审批流程",
        "remark": "报销申请默认审批流程",
    },
    "vehicle_recharge": {
        "wf_name": "ETC/油卡充值审批流程",
        "remark": "充值申请默认审批流程",
    },
}


class WorkflowService:
    @staticmethod
    def _tenant_no(db: Session, user: BxUserContext) -> str:
        return get_tenant_no(db, user.to_tenant_user_dict())

    @staticmethod
    def get_by_id(
        db: Session,
        workflow_id: str,
        tenant_no: str,
        *,
        user: Optional[BxUserContext] = None,
        skip_scope: bool = False,
    ) -> Optional[BxWorkflowDef]:
        return BxCrudService.get_by_id(
            BxWorkflowDef, db, workflow_id, tenant_no, user=user, skip_scope=skip_scope
        )

    @staticmethod
    def get_paged(
        db: Session,
        tenant_no: str,
        page: int,
        page_size: int,
        *,
        user: Optional[BxUserContext] = None,
    ) -> Tuple[Sequence[BxWorkflowDef], int]:
        # 流程定义为租户配置主数据，列表不做 create_by 数据范围过滤
        return BxCrudService.get_paged(
            BxWorkflowDef, db, tenant_no, page, page_size, user=user, skip_scope=True
        )

    @staticmethod
    def create(
        db: Session,
        obj_in: dict,
        *,
        user: BxUserContext,
    ) -> BxWorkflowDef:
        tenant_no = WorkflowService._tenant_no(db, user)
        return BxCrudService.create(
            BxWorkflowDef,
            db,
            obj_in,
            tenant_no=tenant_no,
            user_id=user.id,
            dept_id=user.dept_id,
        )

    @staticmethod
    def update(
        db: Session,
        workflow_id: str,
        obj_in: dict,
        *,
        user: BxUserContext,
    ) -> Optional[BxWorkflowDef]:
        tenant_no = WorkflowService._tenant_no(db, user)
        return BxCrudService.update(
            BxWorkflowDef,
            db,
            workflow_id,
            tenant_no,
            obj_in,
            user_id=user.id,
            user=user,
            skip_scope=True,
        )

    @staticmethod
    def soft_delete(db: Session, workflow_id: str, *, user: BxUserContext) -> bool:
        tenant_no = WorkflowService._tenant_no(db, user)
        deleted = BxCrudService.soft_delete(
            BxWorkflowDef, db, workflow_id, tenant_no, user_id=user.id, user=user, skip_scope=True
        )
        if deleted:
            db.query(BxWorkflowNode).filter(
                BxWorkflowNode.wf_id == workflow_id,
                BxWorkflowNode.tenant_no == tenant_no,
            ).update({"del_flag": 1}, synchronize_session=False)
            db.flush()
        return deleted

    @staticmethod
    def get_by_biz_type(db: Session, biz_type: str, tenant_no: str) -> Optional[BxWorkflowDef]:
        return (
            db.query(BxWorkflowDef)
            .filter(
                BxWorkflowDef.biz_type == biz_type,
                BxWorkflowDef.tenant_no == tenant_no,
                BxWorkflowDef.del_flag == 0,
            )
            .first()
        )

    @staticmethod
    def ensure_by_biz_type(
        db: Session,
        biz_type: str,
        *,
        user: BxUserContext,
    ) -> Optional[BxWorkflowDef]:
        """按业务类型获取工作流；已知类型不存在时自动创建空定义（无节点）。"""
        tenant_no = WorkflowService._tenant_no(db, user)
        workflow = WorkflowService.get_by_biz_type(db, biz_type, tenant_no)
        if workflow:
            return workflow

        meta = _DEFAULT_WORKFLOWS.get(biz_type)
        if not meta:
            return None

        workflow = WorkflowService.create(
            db,
            {
                "biz_type": biz_type,
                "wf_name": meta["wf_name"],
                "remark": meta["remark"],
                "status": 1,
            },
            user=user,
        )
        db.commit()
        db.refresh(workflow)
        return workflow

    @staticmethod
    def save_nodes(
        db: Session,
        wf_id: str,
        nodes: List[dict],
        *,
        user: BxUserContext,
    ) -> List[BxWorkflowNode]:
        tenant_no = WorkflowService._tenant_no(db, user)
        workflow = WorkflowService.get_by_id(db, wf_id, tenant_no, user=user, skip_scope=True)
        if not workflow:
            raise ValueError("工作流不存在")

        disburse_count = 0
        for node_data in nodes:
            action = (
                node_data.get("action_type")
                or node_data.get("actionType")
                or "approve"
            )
            if action == "disburse":
                disburse_count += 1
        if disburse_count > 1:
            raise ValueError("发放登记节点至多一个")

        db.query(BxWorkflowNode).filter(
            BxWorkflowNode.wf_id == wf_id,
            BxWorkflowNode.tenant_no == tenant_no,
        ).delete(synchronize_session=False)

        created: List[BxWorkflowNode] = []
        for node_data in nodes:
            data = dict(node_data)
            data.pop("actionType", None)
            data["id"] = generate_uuid()
            data["wf_id"] = wf_id
            data["tenant_no"] = tenant_no
            data["create_by"] = user.id
            data["create_dept_id"] = user.dept_id or ""
            if data.get("approver_id") is not None:
                data["approver_id"] = str(data["approver_id"]).strip() or None
            if not data.get("action_type"):
                data["action_type"] = "approve"
            node = BxWorkflowNode(**data)
            db.add(node)
            created.append(node)
        db.flush()
        db.expire(workflow, ["nodes"])
        return created