"""金额、时间、响应和异常契约统一,防止各模块用浮点或不同错误格式。""" from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from enum import StrEnum from typing import Any class BusinessError(Exception): """业务错误带稳定机器码,前端不需要解析异常字符串来决定交互。""" def __init__(self, message: str, status: int = 422, code: str = "BUSINESS_ERROR", details: Any = None): super().__init__(message) self.status, self.code, self.details = status, code, details class Status(StrEnum): DRAFT = "draft" PENDING = "pending_audit" APPROVAL = "in_approval" APPROVED = "approved" REJECTED = "rejected" WITHDRAWN = "withdrawn" TRANSITIONS = { Status.DRAFT: {Status.PENDING}, Status.PENDING: {Status.APPROVAL, Status.WITHDRAWN}, Status.APPROVAL: {Status.APPROVED, Status.REJECTED, Status.DRAFT}, Status.REJECTED: {Status.PENDING}, Status.APPROVED: set(), Status.WITHDRAWN: set(), } ROLES = {"employee": 1, "supervisor": 2, "auditor": 3, "manager": 4, "admin": 5} DECISIONS = {"auto_approve", "to_supervisor", "to_finance", "to_manual"} def transition(record: Any, target: str) -> None: """每次迁移都查白名单,审批或后台事件不能绕过终态保护。""" if target not in TRANSITIONS.get(record.status, set()): raise BusinessError(f"不允许从 {record.status} 变更为 {target}", 409, "INVALID_STATE") record.status = target def utcnow() -> datetime: """SQLite 存无时区 UTC,统一去掉 tzinfo 避免比较 aware/naive 时间出错。""" return datetime.now(timezone.utc).replace(tzinfo=None) def iso(value: datetime | None) -> str | None: return value.isoformat() + "Z" if value else None def cents(value: Any, allow_zero: bool = True) -> int: """拒绝 NaN、超精度和超范围;不四舍五入隐藏用户填报错误。""" try: number = Decimal(str(value)) if not number.is_finite() or number < 0 or number > Decimal("9999999999.99"): raise ValueError() if number != number.quantize(Decimal("0.01")) or (not allow_zero and number <= 0): raise ValueError() return int(number * 100) except (InvalidOperation, ValueError, TypeError): raise BusinessError("金额必须是非负且最多两位小数,范围不得超过 9999999999.99", code="INVALID_AMOUNT") def money(value: int | None) -> str | None: return f"{Decimal(value) / 100:.2f}" if value is not None else None def envelope(data: Any, request_id: str = "", message: str = "操作成功") -> dict: return {"data": data, "message": message, "request_id": request_id} def paginate(items: list, page: int, page_size: int) -> dict: """本地版统一分页契约,前端无需针对每类资源重写分页逻辑。""" return {"items": items[(page - 1) * page_size:page * page_size], "total": len(items), "page": page, "page_size": page_size}