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

"""请求契约拒绝未知字段，避免拼错参数时服务端静默采用默认配置。"""
from decimal import Decimal
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.shared.common import cents


class Strict(BaseModel):
    model_config = ConfigDict(extra="forbid")


class Login(Strict):
    username: str = Field(min_length=1, max_length=50, description="已初始化登录名")
    password: str = Field(min_length=1, max_length=200, description="仅传输用于校验，不写入日志")


class Refresh(Strict):
    refresh_token: str = Field(min_length=20, max_length=500, description="轮换或撤销的刷新凭据")


class UserCreate(Strict):
    username: str = Field(pattern=r"^[A-Za-z][A-Za-z0-9_]{2,49}$", description="唯一登录名")
    real_name: str = Field(min_length=1, max_length=80, description="真实显示名称")
    password: str = Field(min_length=10, max_length=200, description="至少大小写、数字和符号")
    department_id: int = Field(gt=0, description="有效部门主键")
    role: Literal["employee", "supervisor", "auditor", "manager", "admin"] = Field(description="五种固定业务身份，不接受任意角色名")


class UserPatch(Strict):
    real_name: str | None = Field(default=None, min_length=1, max_length=80, description="新显示名称")
    department_id: int | None = Field(default=None, gt=0, description="新部门，历史单据部门不变")
    role: Literal["employee", "supervisor", "auditor", "manager", "admin"] | None = Field(default=None, description="管理员授权的固定身份；禁止降级自己")
    active: bool | None = Field(default=None, description="停用后立即拒绝现有访问令牌")


class DepartmentBody(Strict):
    name: str = Field(min_length=1, max_length=100, description="不重复的部门名称")
    active: bool = Field(default=True, description="停用仅影响新业务")


class ConfigPatch(Strict):
    expected_version: int = Field(gt=0, description="乐观并发版本，不覆盖他人的修改")
    enabled: bool | None = Field(default=None, description="可选能力开关，核心规则和决策不能停用")
    config: dict[str, Any] | None = Field(default=None, description="节点白名单参数，禁止密钥或代码")


class RulePatch(Strict):
    expected_version: int = Field(gt=0, description="当前规则版本")
    enabled: bool | None = Field(default=None, description="可选规则生效开关，核心检查不允许关闭")
    severity: Literal["warning", "error", "critical"] | None = Field(default=None, description="命中时的处理级别，核心严重程度有下限")
    params: dict[str, Any] | None = Field(default=None, description="白名单阈值，不接受表达式")
    scope: dict[str, Any] | None = Field(default=None, description="可选规则的部门/类型范围")


class RuleTest(Strict):
    invoice: dict[str, Any] = Field(description="待试算字段，不写正式审核")
    context: dict[str, Any] = Field(default_factory=dict, description="显式试算上下文，不冒充生产历史")


class ReimbursementCreate(Strict):
    title: str = Field(min_length=1, max_length=200, description="报销目的")
    description: str = Field(default="", max_length=5000, description="业务补充说明")
    type: Literal[1, 2, 3] = Field(default=1, description="1差旅/2招待/3办公")
    total_amount: Decimal = Field(description="含税总额，以两位小数传输")

    @field_validator("total_amount")
    @classmethod
    def money_precision(cls, value: Decimal):
        """金额契约在请求入口验证，与数据库整数分口径一致。"""
        cents(value)
        return value


class ReimbursementPatch(Strict):
    revision: int = Field(gt=0, description="必须等于当前可编辑版本")
    title: str | None = Field(default=None, min_length=1, max_length=200, description="修改业务目的，不能空字符串")
    description: str | None = Field(default=None, max_length=5000, description="修订补充说明")
    type: Literal[1, 2, 3] | None = Field(default=None, description="修改1差旅/2招待/3办公类型，影响下一轮匹配")
    total_amount: Decimal | None = Field(default=None, description="修订含税额，服务层校验非负及两位精度")


class Revision(Strict):
    revision: int = Field(gt=0, description="预期单据版本，过期返回409")


class InvoicePatch(Strict):
    revision: int = Field(gt=0, description="所属单据当前版本")
    fields: dict[str, Any] = Field(description="用户填报字段；缺失为未知，不伪造默认金额")


class Rerun(Strict):
    reason: str = Field(min_length=2, max_length=1000, description="重跑理由记录在审计日志")


class Feedback(Strict):
    approval_record_id: int = Field(gt=0, description="本人已处理审批记录")
    corrected_fields: dict[str, Any] = Field(default_factory=dict, description="纠正建议，不覆盖原证据")
    reason: str = Field(min_length=2, max_length=2000, description="判断依据")


class Decide(Strict):
    action: Literal["approve", "reject", "return"] = Field(description="通过、驳回或退回草稿，不接受任意状态")
    opinion: str = Field(min_length=2, max_length=2000, description="所有审批动作必须留可读意见")


class ChainNode(Strict):
    approver_id: int = Field(gt=0, description="实际有效用户，而非角色占位符")
    role: Literal["supervisor", "auditor", "manager", "admin"] = Field(description="必须与指定人员当前真实身份一致")


class ChainBody(Strict):
    name: str = Field(min_length=1, max_length=100, description="审批链显示名")
    route: Literal["to_supervisor", "to_finance", "to_manual"] = Field(description="对应确定性审核路由；系统批准不配置人工链")
    type: Literal[1, 2, 3] | None = Field(default=None, description="适用报销类型，空表示全部")
    department_id: int | None = Field(default=None, gt=0, description="匹配历史单据所属有效部门，空表示全部")
    amount_min: Decimal | None = Field(default=None, description="含税金额下限，空不限制")
    amount_max: Decimal | None = Field(default=None, description="含税金额上限，必须大于等于下限")
    priority: int = Field(default=0, ge=0, le=100, description="同具体度链的选择优先级，越大越优先")
    nodes: list[ChainNode] = Field(min_length=1, max_length=10, description="有序且非空审批链")
    active: bool = Field(default=True, description="仅影响新轮次，已有冻结链不替换")
    expected_version: int | None = Field(default=None, gt=0, description="更新时必填")


class KnowledgeBody(Strict):
    title: str = Field(min_length=1, max_length=200, description="政策原文标题，用作引用出处")
    category: str = Field(default="policy", pattern=r"^[a-zA-Z0-9_-]{1,40}$", description="policy分类参与正式政策审核")
    content: str = Field(min_length=1, max_length=200000, description="用户提供的政策原文")
    active: bool = Field(default=True, description="停用原文不参与新审核检索")
    expected_version: int | None = Field(default=None, gt=0, description="修订时必须等于当前正文版本")


class SearchBody(Strict):
    query: str = Field(min_length=1, max_length=2000, description="实际检索问题，不用作系统指令")
    mode: Literal["keyword", "semantic"] = Field(default="keyword", description="显式本地词项或真实向量语义模式")
    top_k: int = Field(default=5, ge=1, le=20, description="最多返回的真实片段数量")
    category: str | None = Field(default=None, max_length=40, description="可选分类过滤")


class Opinion(Strict):
    opinion: str = Field(min_length=2, max_length=2000, description="处理意见，不得仅置状态无依据")


class ExtractedInvoice(Strict):
    """视觉模型只输出允许字段，无法识别留空；金额不靠默认值补齐。"""
    invoice_code: str | None = Field(default=None, max_length=100, description="原票代码，未知留空")
    invoice_no: str | None = Field(default=None, max_length=100, description="原票号码，非文件去重键")
    invoice_date: str | None = Field(default=None, max_length=30, description="开票日期，规则检查格式及有效期")
    amount: Decimal | None = Field(default=None, description="不含税金额；非负且最多两位小数，缺失未知")
    tax_amount: Decimal | None = Field(default=None, description="原票税额，非负且最多两位小数")
    tax_rate: Decimal | None = Field(default=None, ge=0, le=1, description="原票声明税率，不能推测法定税率")
    buyer_name: str | None = Field(default=None, max_length=200, description="购方完整名称")
    seller_name: str | None = Field(default=None, max_length=200, description="销方完整名称")
    tax_number: str | None = Field(default=None, max_length=100, description="发票税号，仅格式检查")
    invoice_type: str | None = Field(default=None, max_length=100, description="原票类型描述")
    items: list[dict] = Field(default_factory=list, max_length=100, description="最多100条原票明细，不作为自动金额补齐依据")

    @field_validator("amount", "tax_amount")
    @classmethod
    def extraction_money(cls, value):
        if value is not None:
            cents(value)
        return value


class RiskAssessment(Strict):
    risk_level: int = Field(ge=1, le=3, description="只接受低/中/高三级")
    risk_score: Decimal = Field(ge=0, le=1, description="模型结构化归一分值，不保证是校准概率")
    risk_factors: list[str] = Field(description="模型输出的风险因素，记录依据但不能单独批准")
    reasoning: str = Field(min_length=1, max_length=5000)


class PolicyAssessment(Strict):
    status: Literal["compliant", "non_compliant", "partial", "unknown"] = Field(description="结构化政策结论；未知和部分完成不会汇总为全通过")
    reason: str = Field(min_length=1, max_length=5000)
    chunk_ids: list[int] = Field(default_factory=list, description="只能引用本次检索返回的真实块 ID")