PayrollManagementSystem/backend/app/schemas/param.py
liuwenxin 8ad3650dd3 feat(system): 计提参数版本管理、变更留痕与计算预览接口
- 9 个端点:目录快照(asOf)/单参数历史/变更批次提交与撤回/批次分页与明细/两日期对比/Excel 导出/计算预览
- submit_change 全量前置校验(注册表/维度岗位/取值域/同批去重)→ 建批次 → 逐项留痕(旧值取提交时生效值)
- 撤回:FOR UPDATE 锁行 + 已生效禁撤,删除后解析自然回退;同日多次变更取最新留痕
- 预览按岗位模板分派公式(经理聚合口径明确拒绝),出参含所用参数与生效时点;导出走 openpyxl
- ECharts 决策记录:前端统计图使用 echarts(用户指定)

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-12 20:24:17 +08:00

236 lines
7.3 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""计提参数管理的请求/响应模型。
字段命名与 user/role 一致采用蛇形直出(前端 TS 接口同构)。
金额类入参用字符串承载(Decimal 解析),杜绝 float 精度损失。
"""
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from app.core.response import PageQuery
def _parse_amount(value: str, *, field_name: str, allow_negative: bool) -> str:
"""金额字符串校验:必须可被 Decimal 解析;默认拒绝负数(毛利润额除外)。"""
try:
amount = Decimal(value.strip())
except InvalidOperation as exc:
raise ValueError(f"{field_name}必须是有效数字:{value}") from exc
if not allow_negative and amount < 0:
raise ValueError(f"{field_name}不得为负数:{value}")
return value
class ParamDimensionValueOut(BaseModel):
"""某参数单个维度的当前生效值。"""
dimension_value: str
value_text: str
effective_date: date | None = None
class ParamCatalogItemOut(BaseModel):
"""参数目录项:定义元数据 + 各维度截至查询时点的生效值。"""
param_code: str
param_name: str
param_group: str
value_type: str
unit: str | None = None
dimension_type: str
options: list[str] = Field(default_factory=list)
default_value: str | None = None
value_min: str | None = None
value_max: str | None = None
description: str | None = None
values: list[ParamDimensionValueOut] = Field(default_factory=list)
class ParamHistoryOut(BaseModel):
"""单参数的版本历史条目(append-only 留痕)。"""
version_no: str
dimension_value: str
old_value_text: str | None = None
value_text: str
effective_date: date
operator_name: str
create_time: datetime
class ParamChangeItemIn(BaseModel):
"""变更条目:参数 + 维度 + 新值。"""
model_config = ConfigDict(str_strip_whitespace=True)
param_code: str = Field(min_length=1, max_length=64)
dimension_value: str = Field(default="", max_length=32)
value: str = Field(min_length=1, max_length=64)
class ParamChangeIn(BaseModel):
"""提交变更批次。
change_type=1 立即生效(生效日期取今日);change_type=2 指定日期生效,
生效日期必填且不得早于今日(过去没有「预约」的意义,历史回溯走重算授权)。
"""
model_config = ConfigDict(str_strip_whitespace=True)
change_type: int = Field(ge=1, le=2, description="生效方式:1立即生效 2指定日期生效")
effective_date: date | None = Field(default=None, description="指定日期生效时必填")
remark: str | None = Field(default=None, max_length=255)
items: list[ParamChangeItemIn] = Field(min_length=1, max_length=50)
@model_validator(mode="after")
def _check_effective_date(self) -> "ParamChangeIn":
if self.change_type == 2 and self.effective_date is None:
raise ValueError("指定日期生效时必须提供生效日期")
return self
class ParamChangeBatchOut(BaseModel):
"""变更批次(提交/撤回/列表共用)。"""
id: int
version_no: str
effective_date: date
change_type: int
remark: str | None = None
operator_name: str
item_count: int
create_time: datetime
class ParamBatchItemOut(BaseModel):
"""批次明细条目(旧值 → 新值)。"""
param_code: str
param_name: str
dimension_value: str
old_value_text: str | None = None
value_text: str
effective_date: date
class ParamBatchOut(BaseModel):
"""变更批次(列表项)。"""
id: int
version_no: str
effective_date: date
change_type: int
remark: str | None = None
operator_name: str
item_count: int
create_time: datetime
class ParamBatchDetailOut(BaseModel):
"""批次明细:批次头 + 逐项变更。"""
id: int
version_no: str
effective_date: date
change_type: int
remark: str | None = None
operator_name: str
item_count: int
create_time: datetime
items: list[ParamBatchItemOut] = Field(default_factory=list)
class ParamDiffOut(BaseModel):
"""两日期快照对比项(全量参数,changed 标记是否变化)。"""
param_code: str
param_name: str
param_group: str
dimension_value: str
old_value: str | None = None
new_value: str | None = None
changed: bool
class ParamPreviewIn(BaseModel):
"""计算预览入参。金额用字符串承载,Decimal 解析。"""
model_config = ConfigDict(str_strip_whitespace=True)
business_role: str | None = Field(
default=None, max_length=32, description="岗位编码,如 salesperson"
)
customer_type: str | None = Field(
default=None, max_length=32, description="客户类型:personal个人户 enterprise企业户"
)
product_category: str | None = Field(
default=None, max_length=32, description="产品类别:municipal_water/rail_transit"
)
ship_date: date = Field(description="发货日")
receipt_date: date = Field(description="回款到账日")
revenue: str = Field(min_length=1, description="销售额(不含运费)")
profit: str = Field(min_length=1, description="毛利润额(可为负)")
received_amount: str = Field(min_length=1, description="本笔回款金额")
barter: bool = Field(default=False, description="是否以物抵顶")
probation: bool = Field(default=False, description="是否试用期")
@field_validator("revenue", "received_amount")
@classmethod
def _check_non_negative(cls, value: str, info) -> str:
return _parse_amount(value, field_name=info.field_name, allow_negative=False)
@field_validator("profit")
@classmethod
def _check_profit(cls, value: str, info) -> str:
return _parse_amount(value, field_name=info.field_name, allow_negative=True)
@model_validator(mode="after")
def _check_dates(self) -> "ParamPreviewIn":
if self.receipt_date < self.ship_date:
raise ValueError("回款到账日不得早于发货日")
return self
class ParamUsedOut(BaseModel):
"""预览实际用到的参数(对账透明:取值与生效时点)。"""
param_code: str
value_text: str
effective_date: date | None = None
class ParamPreviewOut(BaseModel):
"""计算预览出参。分数串(如 97/10000)为精确值,金额为 Decimal 串。"""
payment_days: int
rate: str
rate_text: str
profit_rate: str
converted_profit: str
performance_amount: str
barter_applied: bool
performance_after_barter: str | None = None
start_payment_ok: bool
payment_weight_r: int
payment_weight: str
payable_amount: str
probation_note: str | None = None
params_used: list[ParamUsedOut] = Field(default_factory=list)
class ParamBatchQuery(PageQuery):
"""变更记录查询条件。"""
start_date: date | None = Field(default=None, description="生效日期起(含)")
end_date: date | None = Field(default=None, description="生效日期止(含)")
class ParamHistoryQuery(PageQuery):
"""参数历史查询条件。"""
dimension_value: str | None = Field(
default=None, max_length=32, description="按维度值筛选,空为全部"
)