"""请求契约拒绝未知字段，避免拼错参数时服务端静默采用默认配置。""" 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")