"""工单状态机与权限判定(纯函数,不碰 ORM 以外的世界)。 状态机: open ──accept──▶ accepted ──start──▶ in_progress ──submit──▶ pending_verify ▲ │ └────────reject────────────┘ │approve ▼ done `canceled` 可以从任何未终结状态进入;`done` 与 `canceled` 是终态。 两条刻意的规则: 1. **驳回不是打回 open,而是回 in_progress**。办理人已经在做了,让他重新"认领"一次纯属折腾; 驳回次数单独计数(`reject_count`),用来提示"这事返工过"。 2. **谁也不能替别人验收**,除了管理员/所有者。提需求的人验收是这套流程的核心信任点, 如果办理人能自己点"通过",工单就退化成"打个勾"。 """ from __future__ import annotations import uuid from ..errors import AppError, ErrorCode TICKET_TRANSITIONS: dict[str, tuple[str, ...]] = { "open": ("accepted", "canceled"), "accepted": ("in_progress", "canceled"), "in_progress": ("pending_verify", "canceled"), "pending_verify": ("done", "in_progress", "canceled"), "done": (), "canceled": (), } # 动作 → (允许的前置状态, 目标状态);cancel 特殊处理 ACTION_RULES: dict[str, tuple[tuple[str, ...], str]] = { "accept": (("open",), "accepted"), "start": (("accepted",), "in_progress"), "submit": (("in_progress",), "pending_verify"), "approve": (("pending_verify",), "done"), "reject": (("pending_verify",), "in_progress"), } CANCELABLE = ("open", "accepted", "in_progress", "pending_verify") NOW_FIELD: dict[str, str] = { "accepted": "accepted_at", "pending_verify": "submitted_at", "done": "verified_at", } def ensure_transition(current: str, target: str) -> None: allowed = TICKET_TRANSITIONS.get(current, ()) if target not in allowed: raise AppError( ErrorCode.TICKET_STATUS_INVALID, f"工单当前是「{current}」,不能直接变成「{target}」", details={"from": current, "to": target, "allowed": list(allowed)}, ) def target_status(action: str, current: str) -> str: """动作 → 目标状态,顺带校验前置状态。""" if action == "cancel": if current not in CANCELABLE: raise AppError( ErrorCode.TICKET_STATUS_INVALID, f"「{current}」的工单不能取消", details={"from": current, "allowed": list(CANCELABLE)}, ) return "canceled" if action not in ACTION_RULES: raise AppError(ErrorCode.BAD_REQUEST, f"未知动作:{action}") allowed_from, target = ACTION_RULES[action] if current not in allowed_from: raise AppError( ErrorCode.TICKET_STATUS_INVALID, f"工单当前是「{current}」,不能执行「{action}」", # allowed 回答"现在还能去哪儿",allowed_from 回答"这个动作要什么前置状态": # 客户端两种都要用得上,别让它自己拼。 details={ "from": current, "action": action, "allowed": list(TICKET_TRANSITIONS.get(current, ())), "allowed_from": list(allowed_from), }, ) return target def ensure_can_act( *, action: str, actor_id: uuid.UUID, role: str, assignee_id: uuid.UUID | None, requester_id: uuid.UUID | None, ) -> None: """谁能做这个动作。 管理员/所有者可以代任何人操作(人休假、离职时的兜底), 普通成员只能动自己的位置上的事。 """ is_admin = role in ("owner", "admin") is_assignee = assignee_id is not None and actor_id == assignee_id is_requester = requester_id is not None and actor_id == requester_id if action in ("accept", "start", "submit"): allowed = is_assignee or is_admin hint = "只有被指派的人才能操作这条工单" elif action in ("approve", "reject"): allowed = is_requester or is_admin hint = "只有提出这条工单的人(或管理员)才能验收" elif action == "cancel": allowed = is_requester or is_assignee or is_admin hint = "只有提出人、办理人或管理员能取消" else: allowed = is_admin or is_requester hint = "权限不足" if not allowed: raise AppError(ErrorCode.TICKET_FORBIDDEN, hint, details={"action": action}) def returned_task_status(ticket_status: str) -> str | None: """工单状态变化后,原任务该回到什么状态;None 表示不动。 闭环回流只发生在验收通过那一刻——这是"事情真的办完了"的唯一信号。 取消不回填:取消意味着不办了,原任务继续躺在 blocked 里等人处理才对。 """ if ticket_status == "done": return "todo" return None