"""多级审批流引擎 — 支持串行、并行、条件审批。 对标 Yearning/Bytebase 的审批能力: - 启动审批流程时根据工单绑定的 workflow 创建审批实例 - 支持通过、拒绝、转交、跳过等操作 - 自动分配审批人(基于负载均衡) - 超时自动处理 """ import json import random from datetime import datetime, timezone, timedelta from typing import List, Optional, Dict, Any from dataclasses import dataclass from sqlalchemy import select, func, and_ from sqlalchemy.ext.asyncio import AsyncSession from app.models.workflow import WorkflowDefinition, WorkflowStep, TicketApproval from app.models.ticket import Ticket from app.models.user import User @dataclass class ApprovalProgress: """审批进度数据。""" ticket_id: int current_step: int total_steps: int status: str # pending / reviewing / approved / rejected steps: List[Dict[str, Any]] class WorkflowEngine: """多级审批流引擎。""" def __init__(self, db: AsyncSession): self.db = db async def start_workflow(self, ticket_id: int, workflow_id: Optional[int] = None) -> List[TicketApproval]: """为工单启动审批流程。 如果没有指定 workflow_id,使用默认流程。 为每个步骤创建 TicketApproval 记录。 """ if workflow_id is None: workflow = await self._get_default_workflow() else: result = await self.db.execute( select(WorkflowDefinition).where(WorkflowDefinition.id == workflow_id) ) workflow = result.scalar_one_or_none() if not workflow: workflow = await self._get_default_workflow() if not workflow: # 没有配置审批流,直接标记为 approved return [] approvals = [] for step in workflow.steps: approval = TicketApproval( ticket_id=ticket_id, step_number=step.step_number, step_name=step.name, approver_role=step.approver_role, action="pending", ) # 如果指定了具体审批人 if step.approver_ids: ids = json.loads(step.approver_ids) if ids: approval.approver_id = ids[0] self.db.add(approval) approvals.append(approval) await self.db.flush() return approvals async def approve(self, approval_id: int, user_id: int, comment: str = "") -> Dict[str, Any]: """审批通过。 Returns: dict with keys: success, new_status, message, next_step """ approval = await self._get_approval(approval_id) if not approval: return {"success": False, "message": "审批记录不存在"} if approval.action != "pending": return {"success": False, "message": f"当前状态为 {approval.action},无法审批"} approval.action = "approved" approval.approver_id = user_id approval.comment = comment or "审批通过" approval.updated_at = datetime.now(timezone.utc) # 检查是否所有步骤都已完成 ticket_approvals = await self._get_ticket_approvals(approval.ticket_id) # 判断当前步骤类型 current_step_num = approval.step_number step_approvals = [a for a in ticket_approvals if a.step_number == current_step_num] if approval.step_type == "parallel": # 并行审批:需要所有审批人都通过 if all(a.action == "approved" for a in step_approvals): all_done = await self._check_all_steps_done(ticket_approvals, current_step_num) if all_done: return await self._complete_approval(approval.ticket_id, "approved") return {"success": True, "new_status": "reviewing", "message": "当前步骤并行审批已完成,等待后续步骤", "next_step": current_step_num + 1} elif approval.step_type == "any_one": # 任一通过即可 all_done = await self._check_all_steps_done(ticket_approvals, current_step_num) if all_done: return await self._complete_approval(approval.ticket_id, "approved") return {"success": True, "new_status": "reviewing", "message": "当前步骤已有人审批通过,等待后续步骤", "next_step": current_step_num + 1} else: # 串行审批 all_done = await self._check_all_steps_done(ticket_approvals, current_step_num) if all_done: return await self._complete_approval(approval.ticket_id, "approved") return {"success": True, "new_status": "reviewing", "message": "审批已通过,等待后续步骤", "next_step": current_step_num + 1} async def reject(self, approval_id: int, user_id: int, comment: str = "") -> Dict[str, Any]: """审批拒绝。任一拒绝则整个工单拒绝。""" approval = await self._get_approval(approval_id) if not approval: return {"success": False, "message": "审批记录不存在"} if approval.action != "pending": return {"success": False, "message": f"当前状态为 {approval.action},无法操作"} approval.action = "rejected" approval.approver_id = user_id approval.comment = comment or "审批拒绝" approval.updated_at = datetime.now(timezone.utc) # 拒绝整个工单 result = await self._complete_approval(approval.ticket_id, "rejected") result["message"] = f"审批已拒绝: {comment}" return result async def transfer(self, approval_id: int, from_user: int, to_user: int, comment: str = "") -> Dict[str, Any]: """转交审批给其他人。""" approval = await self._get_approval(approval_id) if not approval: return {"success": False, "message": "审批记录不存在"} # 创建新的审批记录给目标用户 new_approval = TicketApproval( ticket_id=approval.ticket_id, step_number=approval.step_number, step_name=approval.step_name, approver_id=to_user, approver_role=approval.approver_role, action="pending", ) self.db.add(new_approval) # 标记原审批为转交 approval.action = "transferred" approval.comment = f"转交给用户 {to_user}: {comment}" approval.updated_at = datetime.now(timezone.utc) await self.db.flush() return {"success": True, "message": "审批已转交", "new_approval_id": new_approval.id} async def skip_step(self, approval_id: int, user_id: int, reason: str = "") -> Dict[str, Any]: """管理员跳过当前步骤。""" approval = await self._get_approval(approval_id) if not approval: return {"success": False, "message": "审批记录不存在"} approval.action = "skipped" approval.approver_id = user_id approval.comment = f"跳过: {reason}" approval.updated_at = datetime.now(timezone.utc) # 检查是否全部完成 ticket_approvals = await self._get_ticket_approvals(approval.ticket_id) current_step_num = approval.step_number all_done = await self._check_all_steps_done(ticket_approvals, current_step_num) if all_done: return await self._complete_approval(approval.ticket_id, "approved") return {"success": True, "message": "步骤已跳过", "new_status": "reviewing"} async def get_approval_progress(self, ticket_id: int) -> ApprovalProgress: """获取工单的审批进度。""" approvals = await self._get_ticket_approvals(ticket_id) ticket_result = await self.db.execute(select(Ticket).where(Ticket.id == ticket_id)) ticket = ticket_result.scalar_one_or_none() steps_data = [] for approval in approvals: approver_name = "未分配" if approval.approver_id: user_result = await self.db.execute( select(User).where(User.id == approval.approver_id) ) user = user_result.scalar_one_or_none() approver_name = user.display_name if user else "未知" steps_data.append({ "step_number": approval.step_number, "step_name": approval.step_name, "approver_name": approver_name, "approver_role": approval.approver_role, "action": approval.action, "comment": approval.comment, "created_at": approval.created_at.isoformat() if approval.created_at else None, "updated_at": approval.updated_at.isoformat() if approval.updated_at else None, }) current_step = 1 total_steps = len(approvals) status = ticket.status if ticket else "unknown" # 找到当前待审批的步骤 for a in approvals: if a.action == "pending": current_step = a.step_number break return ApprovalProgress( ticket_id=ticket_id, current_step=current_step, total_steps=total_steps, status=status, steps=steps_data, ) async def auto_assign(self, step: WorkflowStep) -> Optional[int]: """基于负载自动分配审批人。 优先选择当前待审批任务最少的 DBA。 """ if step.approver_ids: ids = json.loads(step.approver_ids) if ids: return ids[0] if step.approver_role: # 查找该角色下负载最小的用户 result = await self.db.execute( select(User.id, User.display_name) .where(User.role == step.approver_role, User.is_active == True) ) candidates = result.all() if not candidates: return None # 统计每个候选人的待审批数量 workloads = [] for user_id, _ in candidates: count_result = await self.db.execute( select(func.count(TicketApproval.id)) .where(TicketApproval.approver_id == user_id, TicketApproval.action == "pending") ) count = count_result.scalar() or 0 workloads.append((user_id, count)) # 选择负载最小的 workloads.sort(key=lambda x: x[1]) return workloads[0][0] return None async def check_timeout_and_process(self): """检查超时的审批并自动处理。""" timeout_threshold = datetime.now(timezone.utc) - timedelta(hours=24) result = await self.db.execute( select(TicketApproval).where( and_(TicketApproval.action == "pending", TicketApproval.created_at < timeout_threshold) ) ) timed_out = result.scalars().all() for approval in timed_out: # 获取步骤定义,检查是否配置为超时自动通过 workflow_result = await self.db.execute( select(WorkflowStep).where(WorkflowStep.id == approval.step_number) ) step_def = workflow_result.scalar_one_or_none() if step_def and step_def.auto_approve_on_timeout: approval.action = "approved" approval.comment = "超时自动通过" approval.updated_at = datetime.now(timezone.utc) # ── 内部辅助方法 ── async def _get_default_workflow(self) -> Optional[WorkflowDefinition]: """获取默认审批流程。""" result = await self.db.execute( select(WorkflowDefinition) .where(WorkflowDefinition.is_default == True, WorkflowDefinition.is_active == True) ) return result.scalar_one_or_none() async def _get_approval(self, approval_id: int) -> Optional[TicketApproval]: result = await self.db.execute( select(TicketApproval).where(TicketApproval.id == approval_id) ) return result.scalar_one_or_none() async def _get_ticket_approvals(self, ticket_id: int) -> List[TicketApproval]: result = await self.db.execute( select(TicketApproval) .where(TicketApproval.ticket_id == ticket_id) .order_by(TicketApproval.step_number) ) return result.scalars().all() async def _check_all_steps_done(self, approvals: List[TicketApproval], current_step: int) -> bool: """检查当前步骤及之前的所有步骤是否都已完成。""" for a in approvals: if a.step_number <= current_step and a.action in ("pending",): return False return True async def _complete_approval(self, ticket_id: int, final_status: str) -> Dict[str, Any]: """完成审批,更新工单状态。""" ticket_result = await self.db.execute( select(Ticket).where(Ticket.id == ticket_id) ) ticket = ticket_result.scalar_one_or_none() if ticket: ticket.status = final_status if final_status == "approved": ticket.review_time = datetime.now(timezone.utc) await self.db.flush() return { "success": True, "new_status": final_status, "message": f"审批流程已完成,工单状态: {final_status}", "next_step": None, } @property def step_type(self) -> str: """获取当前步骤类型。""" return "serial" # 简化实现