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 BxAdvanceApply, BxReimburseApply, 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) workflow = WorkflowService.lock_definition(db, workflow_id, tenant_no) if workflow is None: return None values = {key: value for key, value in obj_in.items() if key in ("biz_type", "wf_name", "status", "remark")} changed = {key: value for key, value in values.items() if getattr(workflow, key) != value} if not changed: return workflow # 名称和备注不改变节点结构。停用或业务类型变化仍整单拒绝,避免名称先被写入。 if any(key in ("biz_type", "status") for key in changed): WorkflowService._assert_not_pending(db, workflow) for key, value in changed.items(): setattr(workflow, key, value) workflow.update_by = user.id db.flush() return workflow @staticmethod def lock_definition(db: Session, wf_id: str, tenant_no: str) -> Optional[BxWorkflowDef]: """提交与配置保存共享定义锁,防止检查后又启动旧流程。""" return db.query(BxWorkflowDef).filter( BxWorkflowDef.id == wf_id, BxWorkflowDef.tenant_no == tenant_no, BxWorkflowDef.del_flag == 0, ).populate_existing().with_for_update().first() @staticmethod def _assert_not_pending(db: Session, workflow: BxWorkflowDef) -> None: models = [(BxAdvanceApply, "pending"), (BxReimburseApply, "pending")] if workflow.biz_type == "vehicle_recharge": from app.models.pl_models import PlVehicleRechargeApply models.append((PlVehicleRechargeApply, "in_approval")) for model, pending_status in models: pending = db.query(model.id).filter( model.wf_id == workflow.id, model.tenant_no == workflow.tenant_no, model.del_flag == 0, model.status == pending_status, ).with_for_update().first() if pending: raise ValueError("该流程存在审批中的单据,请待审批结束后再修改流程") @staticmethod def soft_delete(db: Session, workflow_id: str, *, user: BxUserContext) -> bool: tenant_no = WorkflowService._tenant_no(db, user) workflow = WorkflowService.lock_definition(db, workflow_id, tenant_no) if workflow is None: return False WorkflowService._assert_not_pending(db, workflow) 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.lock_definition(db, wf_id, tenant_no) if not workflow: raise ValueError("工作流不存在") existing = db.query(BxWorkflowNode).filter( BxWorkflowNode.wf_id == wf_id, BxWorkflowNode.tenant_no == tenant_no, BxWorkflowNode.del_flag == 0, ).populate_existing().with_for_update().all() by_id = {row.id: row for row in existing} by_name = {} for row in existing: by_name.setdefault(row.node_name, []).append(row) prepared = [] kept, names, orders = set(), set(), set() disburse_count = 0 for raw in nodes: node_id = str(raw.get("id") or "").strip() name = str(raw.get("node_name") or "").strip() if not name or name in names: raise ValueError("审批节点名称不能为空或重复") names.add(name) if node_id: row = by_id.get(node_id) if row is None or node_id in kept: raise ValueError("审批节点ID重复或不属于当前流程") else: matches = by_name.get(name, []) if len(matches) > 1: raise ValueError("审批节点名称不唯一,请刷新后使用节点ID保存") row = matches[0] if matches else None if row: if row.id in kept: raise ValueError("审批节点ID重复") kept.add(row.id) order = int(raw["node_order"]) if order in orders: raise ValueError("审批节点顺序不能重复") orders.add(order) values = {"node_name": name, "node_order": order} for key in ("approver_id", "approver_name", "action_type"): value = raw.get(key, getattr(row, key, None)) if key == "action_type": value = value or raw.get("actionType") or "approve" if value not in ("approve", "disburse"): raise ValueError("审批节点操作类型无效") disburse_count += value == "disburse" else: value = str(value).strip() if value is not None else None value = value or None values[key] = value prepared.append((row, values)) if disburse_count > 1: raise ValueError("发放登记节点至多一个") def differs(row, values): return row is None or any( (getattr(row, key) or None) != (value or None) for key, value in values.items() ) changed = len(kept) != len(existing) or any(differs(row, values) for row, values in prepared) if not changed: return [row for row, _ in prepared] WorkflowService._assert_not_pending(db, workflow) for row in existing: if row.id not in kept: row.del_flag = 1 row.update_by = user.id saved = [] for row, values in prepared: if row is None: row = BxWorkflowNode( id=generate_uuid(), wf_id=wf_id, tenant_no=tenant_no, create_by=user.id, create_dept_id=user.dept_id or "", **values, ) db.add(row) elif differs(row, values): for key, value in values.items(): setattr(row, key, value) row.update_by = user.id saved.append(row) db.flush() db.expire(workflow, ["nodes"]) return saved