code/ai_finance_agent/backend/app/shared/common.py

"""金额、时间、响应和异常契约统一，防止各模块用浮点或不同错误格式。"""
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}