diff --git a/backend/app/api/router.py b/backend/app/api/router.py index 9999577..392453d 100644 --- a/backend/app/api/router.py +++ b/backend/app/api/router.py @@ -10,6 +10,7 @@ from app.api import health from app.api.v1 import auth from app.api.v1.system import business_roles as system_business_roles from app.api.v1.system import logs as system_logs +from app.api.v1.system import params as system_params from app.api.v1.system import roles as system_roles from app.api.v1.system import users as system_users @@ -27,6 +28,7 @@ system_router = APIRouter(prefix="/system") system_router.include_router(system_users.router) system_router.include_router(system_roles.router) system_router.include_router(system_business_roles.router) +system_router.include_router(system_params.router) system_router.include_router(system_logs.operation_router) system_router.include_router(system_logs.login_router) v1_router.include_router(system_router) diff --git a/backend/app/api/v1/system/params.py b/backend/app/api/v1/system/params.py new file mode 100644 index 0000000..8adbd14 --- /dev/null +++ b/backend/app/api/v1/system/params.py @@ -0,0 +1,147 @@ +"""计提参数管理接口(需求 #28:参数配置化 + 版本管理 + 留痕)。 + +全部要求管理员权限(开放问题 #5 的当前默认,后续可放宽只读); +写操作(提交/撤回)经 `RequestMetaDep` 带入请求元信息写入操作日志, +变更批次本身另行留痕(变更人/时间/项/旧值/新值/生效日期)——双保险。 +导出为文件流(非统一信封),前端以 blob 接收。 +""" + +from datetime import date +from typing import Annotated + +from fastapi import APIRouter, Path, Query +from fastapi.responses import StreamingResponse + +from app.api.deps import AdminUser, DbSession, RequestMetaDep +from app.core.response import ApiResponse, PageResult, ok +from app.schemas.param import ( + ParamBatchDetailOut, + ParamBatchOut, + ParamBatchQuery, + ParamCatalogItemOut, + ParamChangeBatchOut, + ParamChangeIn, + ParamDiffOut, + ParamHistoryOut, + ParamHistoryQuery, + ParamPreviewIn, + ParamPreviewOut, +) +from app.services import param_service + +router = APIRouter(prefix="/params", tags=["系统管理 · 计提参数"]) + + +@router.get( + "", response_model=ApiResponse[list[ParamCatalogItemOut]], summary="参数目录(含当前生效值)" +) +def get_catalog( + session: DbSession, + _: AdminUser, + as_of: Annotated[date | None, Query(description="业务日期,默认今日")] = None, +) -> ApiResponse[list[ParamCatalogItemOut]]: + """全部启用参数的定义元数据与截至 asOf 的生效值(前端按分组展示)。""" + return ok(param_service.get_catalog(session, as_of=as_of)) + + +@router.post("/changes", response_model=ApiResponse[ParamChangeBatchOut], summary="提交变更批次") +def submit_change( + session: DbSession, + current_user: AdminUser, + payload: ParamChangeIn, + meta: RequestMetaDep, +) -> ApiResponse[ParamChangeBatchOut]: + """提交一批参数变更(1~50 项),返回版本号;立即生效或指定日期生效。""" + return ok( + param_service.submit_change(session, payload, operator=current_user, meta=meta), + message="变更已提交", + ) + + +@router.get( + "/changes", response_model=ApiResponse[PageResult[ParamBatchOut]], summary="变更记录分页" +) +def list_changes( + session: DbSession, + _: AdminUser, + query: Annotated[ParamBatchQuery, Query()], +) -> ApiResponse[PageResult[ParamBatchOut]]: + """按生效日期区间分页查询变更批次(创建时间倒序)。""" + return ok(param_service.list_batches(session, query)) + + +@router.get( + "/changes/{batch_id}", response_model=ApiResponse[ParamBatchDetailOut], summary="批次明细" +) +def get_batch( + session: DbSession, + _: AdminUser, + batch_id: Annotated[int, Path(ge=1)], +) -> ApiResponse[ParamBatchDetailOut]: + """批次头 + 逐项旧值 → 新值。""" + return ok(param_service.get_batch_detail(session, batch_id)) + + +@router.delete("/changes/{batch_id}", response_model=ApiResponse[None], summary="撤回未生效批次") +def withdraw_batch( + session: DbSession, + current_user: AdminUser, + batch_id: Annotated[int, Path(ge=1)], + meta: RequestMetaDep, +) -> ApiResponse[None]: + """撤回尚未生效的批次;已生效批次不可撤回。""" + param_service.withdraw_batch(session, batch_id, operator=current_user, meta=meta) + return ok(message="批次已撤回") + + +@router.get("/compare", response_model=ApiResponse[list[ParamDiffOut]], summary="两日期快照对比") +def compare( + session: DbSession, + _: AdminUser, + from_date: Annotated[date, Query(description="起始业务日期(含)")], + to_date: Annotated[date, Query(description="截止业务日期(含)")], +) -> ApiResponse[list[ParamDiffOut]]: + """两个业务日期的全量参数快照逐项对比,changed 标记是否变化。""" + return ok(param_service.compare(session, from_date=from_date, to_date=to_date)) + + +@router.get("/export", summary="导出变更记录 Excel") +def export_changes( + session: DbSession, + _: AdminUser, + start_date: Annotated[date | None, Query(description="生效日期起(含)")] = None, + end_date: Annotated[date | None, Query(description="生效日期止(含)")] = None, +) -> StreamingResponse: + """变更记录导出为 Excel 文件流(非统一信封,前端以 blob 接收)。""" + content = param_service.export_changes(session, start_date=start_date, end_date=end_date) + filename = f"param_changes_{date.today():%Y%m%d}.xlsx" + return StreamingResponse( + iter([content]), + media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) + + +@router.post("/preview", response_model=ApiResponse[ParamPreviewOut], summary="计算预览") +def preview( + session: DbSession, + _: AdminUser, + payload: ParamPreviewIn, +) -> ApiResponse[ParamPreviewOut]: + """输入业务场景(岗位/产品/发货/回款/金额),按回款到账日取参数实时试算。""" + return ok(param_service.preview(session, payload)) + + +@router.get( + "/{param_code}/history", + response_model=ApiResponse[PageResult[ParamHistoryOut]], + summary="参数版本历史", +) +def get_history( + session: DbSession, + _: AdminUser, + param_code: Annotated[str, Path(max_length=64)], + query: Annotated[ParamHistoryQuery, Query()], +) -> ApiResponse[PageResult[ParamHistoryOut]]: + """单参数的版本历史(含版本号、变更人与旧值 → 新值),最新在前。""" + return ok(param_service.get_history(session, param_code, query)) diff --git a/backend/app/core/constants.py b/backend/app/core/constants.py index 4d9f579..f21a4c0 100644 --- a/backend/app/core/constants.py +++ b/backend/app/core/constants.py @@ -51,6 +51,7 @@ PRODUCT_CATEGORY_RAIL_TRANSIT = "rail_transit" MODULE_USER = "用户管理" MODULE_ROLE = "角色管理" MODULE_BUSINESS_ROLE = "岗位角色" +MODULE_PARAM = "计提参数" ACTION_CREATE = "新增" ACTION_UPDATE = "修改" diff --git a/backend/app/repositories/param_repository.py b/backend/app/repositories/param_repository.py index 8c461d6..65f4765 100644 --- a/backend/app/repositories/param_repository.py +++ b/backend/app/repositories/param_repository.py @@ -6,20 +6,20 @@ from datetime import date -from sqlalchemy import func, select +from sqlalchemy import Select, delete, func, select from sqlalchemy.orm import Session -from app.models import ParamDefinition, ParamValue +from app.models import ParamChangeBatch, ParamDefinition, ParamValue # ==================== 读取 ==================== -def resolve_values(session: Session, biz_date: date) -> list[tuple[str, str, str]]: +def resolve_values(session: Session, biz_date: date) -> list[tuple[str, str, str, date]]: """时点化解析:每个 (param_id, dimension_value) 取生效日期 <= 业务日期的最新条目。 排序 (effective_date DESC, id DESC) 保证同日多次变更取最新一条 (不设日期唯一键,同日修正是合法留痕)。仅返回启用状态的参数定义。 - 返回 (param_code, dimension_value, value_text),供 ParamSnapshot 构造。 + 返回 (param_code, dimension_value, value_text, effective_date)。 """ row_number = ( func.row_number() @@ -34,13 +34,194 @@ def resolve_values(session: Session, biz_date: date) -> list[tuple[str, str, str ParamDefinition.param_code.label("param_code"), ParamValue.dimension_value.label("dimension_value"), ParamValue.value_text.label("value_text"), + ParamValue.effective_date.label("effective_date"), row_number, ) .join(ParamDefinition, ParamDefinition.id == ParamValue.param_id) .where(ParamValue.effective_date <= biz_date, ParamDefinition.status == 1) .subquery() ) - stmt = select(subq.c.param_code, subq.c.dimension_value, subq.c.value_text).where( - subq.c.rn == 1 + stmt = select( + subq.c.param_code, subq.c.dimension_value, subq.c.value_text, subq.c.effective_date + ).where(subq.c.rn == 1) + return [ + (row.param_code, row.dimension_value, row.value_text, row.effective_date) + for row in session.execute(stmt) + ] + + +def latest_value_before( + session: Session, param_id: int, dimension_value: str, biz_date: date +) -> ParamValue | None: + """取某参数某维度在业务日期(含)之前的最新条目 —— 提交变更时记录「旧值」。""" + stmt = ( + select(ParamValue) + .where( + ParamValue.param_id == param_id, + ParamValue.dimension_value == dimension_value, + ParamValue.effective_date <= biz_date, + ) + .order_by(ParamValue.effective_date.desc(), ParamValue.id.desc()) + .limit(1) ) - return [(row.param_code, row.dimension_value, row.value_text) for row in session.execute(stmt)] + return session.execute(stmt).scalar_one_or_none() + + +def list_definitions(session: Session, *, status: int | None = None) -> list[ParamDefinition]: + """参数定义列表(目录)。""" + stmt = select(ParamDefinition).order_by(ParamDefinition.sort_order, ParamDefinition.id) + if status is not None: + stmt = stmt.where(ParamDefinition.status == status) + return list(session.execute(stmt).scalars()) + + +def get_definition_by_code(session: Session, param_code: str) -> ParamDefinition | None: + """按编码取参数定义。""" + stmt = select(ParamDefinition).where(ParamDefinition.param_code == param_code) + return session.execute(stmt).scalar_one_or_none() + + +def list_history( + session: Session, + param_id: int, + *, + offset: int, + limit: int, + dimension_value: str | None = None, +) -> list[tuple[ParamValue, ParamChangeBatch]]: + """单参数的版本历史(含批次信息),最新在前。""" + stmt = _history_stmt(session, param_id, dimension_value=dimension_value) + return [(value, batch) for value, batch in session.execute(stmt.offset(offset).limit(limit))] + + +def count_history(session: Session, param_id: int, *, dimension_value: str | None = None) -> int: + """统计单参数的历史条目数。""" + stmt = select(func.count()).select_from( + _history_stmt(session, param_id, dimension_value=dimension_value).subquery() + ) + return int(session.execute(stmt).scalar_one()) + + +def _history_stmt(session: Session, param_id: int, *, dimension_value: str | None) -> Select: + """历史查询基语句:条目 join 批次(版本号/操作人),最新在前。""" + stmt = ( + select(ParamValue, ParamChangeBatch) + .join(ParamChangeBatch, ParamChangeBatch.id == ParamValue.batch_id) + .where(ParamValue.param_id == param_id) + .order_by(ParamValue.effective_date.desc(), ParamValue.id.desc()) + ) + if dimension_value is not None: + stmt = stmt.where(ParamValue.dimension_value == dimension_value) + return stmt + + +# ==================== 批次 ==================== + + +def create_batch(session: Session, batch: ParamChangeBatch) -> ParamChangeBatch: + """新增变更批次。只 flush 以取得自增 id。""" + session.add(batch) + session.flush() + return batch + + +def get_batch_by_id( + session: Session, batch_id: int, *, for_update: bool = False +) -> ParamChangeBatch | None: + """按主键取批次。撤回校验走 ``for_update`` 锁行(防并发双撤)。""" + if for_update: + stmt = select(ParamChangeBatch).where(ParamChangeBatch.id == batch_id).with_for_update() + return session.execute(stmt).scalar_one_or_none() + return session.get(ParamChangeBatch, batch_id) + + +def list_batches( + session: Session, + *, + offset: int, + limit: int, + start_date: date | None = None, + end_date: date | None = None, +) -> list[ParamChangeBatch]: + """分页查询变更批次(按创建时间倒序)。""" + stmt = _batch_filters(select(ParamChangeBatch), start_date=start_date, end_date=end_date) + return list( + session.execute( + stmt.order_by(ParamChangeBatch.create_time.desc(), ParamChangeBatch.id.desc()) + .offset(offset) + .limit(limit) + ).scalars() + ) + + +def count_batches( + session: Session, *, start_date: date | None = None, end_date: date | None = None +) -> int: + """统计符合条件的批次总数。""" + stmt = _batch_filters( + select(func.count(ParamChangeBatch.id)), start_date=start_date, end_date=end_date + ) + return int(session.execute(stmt).scalar_one()) + + +def _batch_filters(stmt: Select, *, start_date: date | None, end_date: date | None) -> Select: + """批次筛选(生效日期区间,含边界)。列表与计数共用。""" + if start_date is not None: + stmt = stmt.where(ParamChangeBatch.effective_date >= start_date) + if end_date is not None: + stmt = stmt.where(ParamChangeBatch.effective_date <= end_date) + return stmt + + +def list_values_by_batch( + session: Session, batch_id: int +) -> list[tuple[ParamValue, ParamDefinition]]: + """批次内全部条目(join 参数定义取编码/名称),按 id 排序。""" + stmt = ( + select(ParamValue, ParamDefinition) + .join(ParamDefinition, ParamDefinition.id == ParamValue.param_id) + .where(ParamValue.batch_id == batch_id) + .order_by(ParamValue.id) + ) + return [(value, definition) for value, definition in session.execute(stmt)] + + +def delete_values_by_batch(session: Session, batch_id: int) -> None: + """删除批次下全部条目(撤回未生效批次时使用,2.0 风格 delete 语句)。""" + session.execute(delete(ParamValue).where(ParamValue.batch_id == batch_id)) + + +def delete_batch(session: Session, batch: ParamChangeBatch) -> None: + """删除批次本体(调用方须先删条目并完成保护校验)。""" + session.delete(batch) + session.flush() + + +# ==================== 写入 ==================== + + +def create_value(session: Session, value: ParamValue) -> ParamValue: + """新增版本条目(append-only,禁改禁删已生效条目)。只 flush。""" + session.add(value) + session.flush() + return value + + +# ==================== 导出 ==================== + + +def list_export_rows( + session: Session, *, start_date: date | None = None, end_date: date | None = None +) -> list[tuple[ParamChangeBatch, ParamValue, ParamDefinition]]: + """导出行:批次 × 条目 × 定义(变更人/时间/参数/维度/旧值/新值/生效日期)。""" + stmt = ( + select(ParamChangeBatch, ParamValue, ParamDefinition) + .join(ParamValue, ParamValue.batch_id == ParamChangeBatch.id) + .join(ParamDefinition, ParamDefinition.id == ParamValue.param_id) + .order_by(ParamChangeBatch.create_time.desc(), ParamChangeBatch.id.desc(), ParamValue.id) + ) + if start_date is not None: + stmt = stmt.where(ParamChangeBatch.effective_date >= start_date) + if end_date is not None: + stmt = stmt.where(ParamChangeBatch.effective_date <= end_date) + return [(batch, value, definition) for batch, value, definition in session.execute(stmt)] diff --git a/backend/app/schemas/param.py b/backend/app/schemas/param.py new file mode 100644 index 0000000..773c52e --- /dev/null +++ b/backend/app/schemas/param.py @@ -0,0 +1,235 @@ +"""计提参数管理的请求/响应模型。 + +字段命名与 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="按维度值筛选,空为全部" + ) diff --git a/backend/app/services/formula.py b/backend/app/services/formula.py index fd2c999..56c9b43 100644 --- a/backend/app/services/formula.py +++ b/backend/app/services/formula.py @@ -13,11 +13,15 @@ from decimal import Decimal from fractions import Fraction -def _to_decimal(value: Fraction) -> Decimal: +def to_decimal(value: Fraction) -> Decimal: """Fraction → Decimal。有限小数精确;循环小数按 Decimal 上下文精度(28 位)。""" return Decimal(value.numerator) / Decimal(value.denominator) +# 模块内沿用私有名;公开别名供 service 层金额换算复用 +_to_decimal = to_decimal + + def payment_days(ship_date: date, receipt_date: date) -> int: """回款天数:发货日首日计第 1 天(对照表 #1;需规「26 天」示例作废)。""" return (receipt_date - ship_date).days + 1 diff --git a/backend/app/services/param_service.py b/backend/app/services/param_service.py index a0b6bd8..f8da0a3 100644 --- a/backend/app/services/param_service.py +++ b/backend/app/services/param_service.py @@ -1,20 +1,55 @@ -"""计提参数服务:时点化取值(解析部分)。 +"""计提参数服务:时点化取值、版本管理(变更/撤回/历史/对比)、计算预览与导出。 ParamSnapshot 是 M3 核算与计算预览共用的输入契约:按业务日期(回款到账日) 冻结当时口径,禁读「当前值」计算历史数据(算法红线)。 -变更提交/历史/对比/预览/导出在后续任务追加。 +变更批次 = 版本(一次提交一批次);条目逐项留痕(旧值/新值),append-only。 """ +import logging from collections.abc import Iterable from dataclasses import dataclass from datetime import date +from decimal import Decimal from fractions import Fraction +from io import BytesIO +from sqlalchemy import select from sqlalchemy.orm import Session -from app.core.exceptions import BadRequestError -from app.repositories.param_repository import resolve_values -from app.services.param_registry import get_spec, parse_fraction +from app.core.constants import ( + ACTION_DELETE, + ACTION_UPDATE, + MODULE_PARAM, +) +from app.core.exceptions import BadRequestError, NotFoundError +from app.core.response import PageResult +from app.models import ParamChangeBatch, ParamDefinition, ParamValue, User +from app.repositories import business_role_repository, param_repository +from app.schemas.param import ( + ParamBatchDetailOut, + ParamBatchItemOut, + ParamBatchOut, + ParamBatchQuery, + ParamCatalogItemOut, + ParamChangeBatchOut, + ParamChangeIn, + ParamDiffOut, + ParamDimensionValueOut, + ParamHistoryOut, + ParamHistoryQuery, + ParamPreviewIn, + ParamPreviewOut, + ParamUsedOut, +) +from app.services import audit_service, formula +from app.services.audit_service import AuditRecord, RequestMeta +from app.services.formula import to_decimal +from app.services.param_registry import get_spec, parse_fraction, validate_and_normalize + +logger = logging.getLogger(__name__) + + +# ==================== 快照 ==================== @dataclass(frozen=True, slots=True) @@ -26,17 +61,24 @@ class ParamSnapshot: biz_date: date _values: dict[str, dict[str, str]] + _effective: dict[tuple[str, str], date] @classmethod - def from_rows(cls, biz_date: date, rows: Iterable[tuple[str, str, str]]) -> "ParamSnapshot": - """由 (param_code, dimension_value, value_text) 行构造。 + def from_rows( + cls, biz_date: date, rows: Iterable[tuple[str, str, str, date]] + ) -> "ParamSnapshot": + """由 (param_code, dimension_value, value_text, effective_date) 行构造。 rows 须按仓储输出顺序(同一 (code, dim) 的最新生效值在前)——以先出现者为准。 """ values: dict[str, dict[str, str]] = {} - for code, dimension, value_text in rows: + effective: dict[tuple[str, str], date] = {} + for code, dimension, value_text, effective_date in rows: + if (code, dimension) in effective: + continue + effective[(code, dimension)] = effective_date values.setdefault(code, {}).setdefault(dimension, value_text) - return cls(biz_date=biz_date, _values=values) + return cls(biz_date=biz_date, _values=values, _effective=effective) def get( self, code: str, *, role_code: str | None = None, product_category: str | None = None @@ -52,6 +94,23 @@ class ParamSnapshot: raise BadRequestError(f"参数 {code} 无生效值且无默认值,请先在系统设置中维护") return spec.default_value + def effective_date_of( + self, code: str, *, role_code: str | None = None, product_category: str | None = None + ) -> date | None: + """取参数当前生效值的生效日期(回退注册表默认值时为 None)。""" + candidates = self._values.get(code) + if candidates: + for dimension in (role_code, product_category, ""): + if dimension is not None and (code, dimension) in self._effective: + return self._effective[(code, dimension)] + return None + + def iter_items(self) -> Iterable[tuple[str, str, str]]: + """遍历快照全部 (param_code, dimension_value, value_text),供对比。""" + for code, dims in self._values.items(): + for dimension, value_text in dims.items(): + yield code, dimension, value_text + def fraction( self, code: str, *, role_code: str | None = None, product_category: str | None = None ) -> Fraction: @@ -84,4 +143,488 @@ class ParamSnapshot: def resolve_snapshot(session: Session, biz_date: date) -> ParamSnapshot: """按业务日期解析全量参数快照(M3 核算与计算预览共用入口)。""" - return ParamSnapshot.from_rows(biz_date, resolve_values(session, biz_date)) + return ParamSnapshot.from_rows(biz_date, param_repository.resolve_values(session, biz_date)) + + +# ==================== 目录 ==================== + + +def get_catalog(session: Session, *, as_of: date | None = None) -> list[ParamCatalogItemOut]: + """参数目录:全部启用定义 + 各维度截至 asOf(默认今日)的生效值。""" + biz_date = as_of or date.today() + resolved = param_repository.resolve_values(session, biz_date) + by_code: dict[str, list[ParamDimensionValueOut]] = {} + for code, dimension, value_text, effective_date in resolved: + by_code.setdefault(code, []).append( + ParamDimensionValueOut( + dimension_value=dimension, value_text=value_text, effective_date=effective_date + ) + ) + items: list[ParamCatalogItemOut] = [] + for definition in param_repository.list_definitions(session, status=1): + spec = get_spec(definition.param_code) + items.append( + ParamCatalogItemOut( + param_code=definition.param_code, + param_name=definition.param_name, + param_group=definition.param_group, + value_type=definition.value_type, + unit=definition.unit, + dimension_type=definition.dimension_type, + options=list(spec.options or ()), + default_value=definition.default_value, + value_min=definition.value_min, + value_max=definition.value_max, + description=definition.description, + values=by_code.get(definition.param_code, []), + ) + ) + return items + + +# ==================== 变更与版本管理 ==================== + + +def _next_version_no(session: Session) -> str: + """生成批次版本号 P{YYYYMMDD}-{当日序号}(当日自增,两位补零)。""" + today = date.today() + prefix = f"P{today:%Y%m%d}" + seq = 0 + for (version_no,) in session.execute( + select(ParamChangeBatch.version_no).where(ParamChangeBatch.version_no.like(f"{prefix}-%")) + ): + try: + seq = max(seq, int(version_no.rsplit("-", 1)[1])) + except (IndexError, ValueError): # 手工造数容错:跳过不合规格的版本号 + continue + return f"{prefix}-{seq + 1:02d}" + + +def submit_change( + session: Session, + payload: ParamChangeIn, + *, + operator: User, + meta: RequestMeta | None = None, +) -> ParamChangeBatchOut: + """提交变更批次:全量校验 → 建批次 → 逐项写条目(旧值取提交时生效值)。 + + :raises BadRequestError: 参数未定义 / 维度不合法 / 值校验失败 / 生效日期早于今日 / + 同批次重复条目 + """ + effective = payload.effective_date if payload.change_type == 2 else date.today() + with audit_service.audit( + session, + operator=operator, + record=AuditRecord( + module=MODULE_PARAM, + action=ACTION_UPDATE, + summary=( + f"提交计提参数变更 {len(payload.items)} 项," + f"{'立即生效' if payload.change_type == 1 else f'{effective} 起生效'}" + ), + request_data=payload.model_dump(exclude_none=True), + ), + meta=meta, + ): + if payload.change_type == 2 and effective < date.today(): + raise BadRequestError("指定生效日期不得早于今日") + + # 先全量校验、再写入:任何一项不合法整批拒绝(不留半批脏数据) + normalized: list[tuple[ParamDefinition, str, str]] = [] + seen: set[tuple[str, str]] = set() + for item in payload.items: + spec = get_spec(item.param_code) # 未注册编码属代码缺陷 + definition = param_repository.get_definition_by_code(session, item.param_code) + if definition is None: + raise BadRequestError(f"参数 {item.param_code} 未在系统中定义") + dimension = item.dimension_value + if spec.dimension_type == "NONE" and dimension != "": + raise BadRequestError(f"参数 {item.param_code} 不带维度,维度值须为空") + if ( + spec.dimension_type == "BUSINESS_ROLE" + and dimension + and business_role_repository.get_by_code(session, dimension) is None + ): + raise BadRequestError( + f"参数 {item.param_code} 的维度值 {dimension!r} 不是已存在的岗位" + ) + value = validate_and_normalize(spec, item.value) + key = (item.param_code, dimension) + if key in seen: + raise BadRequestError( + f"同一批次内重复提交参数 {item.param_code}({dimension or '默认'})" + ) + seen.add(key) + normalized.append((definition, dimension, value)) + + batch = param_repository.create_batch( + session, + ParamChangeBatch( + version_no=_next_version_no(session), + effective_date=effective, + change_type=payload.change_type, + remark=payload.remark, + operator_id=operator.id, + operator_name=operator.username, + item_count=len(normalized), + ), + ) + today = date.today() + for definition, dimension, value in normalized: + old = param_repository.latest_value_before(session, definition.id, dimension, today) + param_repository.create_value( + session, + ParamValue( + param_id=definition.id, + dimension_value=dimension, + effective_date=effective, + value_text=value, + old_value_text=old.value_text if old else None, + batch_id=batch.id, + ), + ) + session.commit() + logger.info("提交参数变更 | version_no=%s | items=%s", batch.version_no, batch.item_count) + return ParamChangeBatchOut( + id=batch.id, + version_no=batch.version_no, + effective_date=batch.effective_date, + change_type=batch.change_type, + remark=batch.remark, + operator_name=batch.operator_name, + item_count=batch.item_count, + create_time=batch.create_time, + ) + + +def withdraw_batch( + session: Session, batch_id: int, *, operator: User, meta: RequestMeta | None = None +) -> None: + """撤回未生效批次。已生效禁撤;条目删除后解析自然回退到更早版本。 + + :raises NotFoundError: 批次不存在 + :raises BadRequestError: 批次已生效 + """ + batch = param_repository.get_batch_by_id(session, batch_id, for_update=True) + with audit_service.audit( + session, + operator=operator, + record=AuditRecord( + module=MODULE_PARAM, + action=ACTION_DELETE, + summary=f"撤回参数变更批次 {batch.version_no if batch else f'#{batch_id}'}", + ), + meta=meta, + ): + if batch is None: + raise NotFoundError("变更批次不存在") + if batch.effective_date <= date.today(): + raise BadRequestError("批次已生效,不可撤回") + + param_repository.delete_values_by_batch(session, batch.id) + param_repository.delete_batch(session, batch) + session.commit() + logger.info("撤回参数变更批次 | version_no=%s", batch.version_no) + + +def get_history( + session: Session, param_code: str, query: ParamHistoryQuery +) -> PageResult[ParamHistoryOut]: + """单参数版本历史(分页,最新在前)。 + + :raises NotFoundError: 参数不存在 + """ + definition = param_repository.get_definition_by_code(session, param_code) + if definition is None: + raise NotFoundError("参数不存在") + rows = param_repository.list_history( + session, + definition.id, + offset=query.offset, + limit=query.page_size, + dimension_value=query.dimension_value, + ) + total = param_repository.count_history( + session, definition.id, dimension_value=query.dimension_value + ) + return PageResult[ParamHistoryOut]( + records=[ + ParamHistoryOut( + version_no=batch.version_no, + dimension_value=value.dimension_value, + old_value_text=value.old_value_text, + value_text=value.value_text, + effective_date=value.effective_date, + operator_name=batch.operator_name, + create_time=value.create_time, + ) + for value, batch in rows + ], + total=total, + page_num=query.page_num, + page_size=query.page_size, + ) + + +def list_batches(session: Session, query: ParamBatchQuery) -> PageResult[ParamBatchOut]: + """变更批次分页(按创建时间倒序)。""" + batches = param_repository.list_batches( + session, + offset=query.offset, + limit=query.page_size, + start_date=query.start_date, + end_date=query.end_date, + ) + total = param_repository.count_batches( + session, start_date=query.start_date, end_date=query.end_date + ) + return PageResult[ParamBatchOut]( + records=[ + ParamBatchOut( + id=batch.id, + version_no=batch.version_no, + effective_date=batch.effective_date, + change_type=batch.change_type, + remark=batch.remark, + operator_name=batch.operator_name, + item_count=batch.item_count, + create_time=batch.create_time, + ) + for batch in batches + ], + total=total, + page_num=query.page_num, + page_size=query.page_size, + ) + + +def get_batch_detail(session: Session, batch_id: int) -> ParamBatchDetailOut: + """批次明细:批次头 + 逐项旧值 → 新值。 + + :raises NotFoundError: 批次不存在 + """ + batch = param_repository.get_batch_by_id(session, batch_id) + if batch is None: + raise NotFoundError("变更批次不存在") + items = [ + ParamBatchItemOut( + param_code=definition.param_code, + param_name=definition.param_name, + dimension_value=value.dimension_value, + old_value_text=value.old_value_text, + value_text=value.value_text, + effective_date=value.effective_date, + ) + for value, definition in param_repository.list_values_by_batch(session, batch_id) + ] + return ParamBatchDetailOut( + id=batch.id, + version_no=batch.version_no, + effective_date=batch.effective_date, + change_type=batch.change_type, + remark=batch.remark, + operator_name=batch.operator_name, + item_count=batch.item_count, + create_time=batch.create_time, + items=items, + ) + + +# ==================== 对比与预览 ==================== + + +def compare(session: Session, *, from_date: date, to_date: date) -> list[ParamDiffOut]: + """两日期快照逐项对比(全量参数,changed 标记变化)。 + + :raises BadRequestError: 起始日期晚于截止日期 + """ + if from_date > to_date: + raise BadRequestError("起始日期不得晚于截止日期") + old_items = { + (code, dimension): value + for code, dimension, value in resolve_snapshot(session, from_date).iter_items() + } + new_items = { + (code, dimension): value + for code, dimension, value in resolve_snapshot(session, to_date).iter_items() + } + out: list[ParamDiffOut] = [] + for code, dimension in sorted(set(old_items) | set(new_items)): + spec = get_spec(code) + old_value = old_items.get((code, dimension)) + new_value = new_items.get((code, dimension)) + out.append( + ParamDiffOut( + param_code=code, + param_name=spec.name, + param_group=spec.group, + dimension_value=dimension, + old_value=old_value, + new_value=new_value, + changed=old_value != new_value, + ) + ) + return out + + +def _preview_rate( + session: Session, + payload: ParamPreviewIn, + snapshot: ParamSnapshot, + days: int, + take, +) -> tuple[Fraction, str]: + """按岗位模板分派计提比例,返回(纯比例, 人读说明)。 + + 聚合模板(营销经理 12.5%+5%)依赖月度团队数据,不支持单笔预览。 + """ + if payload.customer_type == "enterprise": + rate = take("enterprise.rate") + return rate, "企业户固定计提比例" + + role_code = payload.business_role + template = "TPL_SALESPERSON_DECAY" + if role_code: + role = business_role_repository.get_by_code(session, role_code) + if role is None: + raise BadRequestError(f"岗位 {role_code!r} 不存在") + template = role.calc_template + + if template == "TPL_MANAGER_SHARE": + raise BadRequestError( + "营销经理绩效为团队聚合口径(本团队 12.5% + 全公司 5%),不支持单笔预览" + ) + if template == "TPL_INFO_SALES_FIXED": + rate = take("info.rate") + return rate, "信息员:本人销售额 × 计提比例" + if template == "TPL_SALESPERSON_DECAY": + start = take("salesperson.rate.start") + slope = take("salesperson.rate.slope") + floor = take("salesperson.rate.floor") + boundary = int(take("salesperson.days.boundary")) + rate = formula.salesperson_rate( + days, start=start, slope=slope, floor=floor, boundary_days=boundary + ) + return ( + rate, + f"业务员递减:{to_decimal(start)}‰ 起,{days} 天(分界 {boundary} 天,超界取 {to_decimal(floor)}‰)", + ) + raise BadRequestError(f"公式模板 {template} 暂不支持预览") + + +def preview(session: Session, payload: ParamPreviewIn) -> ParamPreviewOut: + """计算预览:入参(发货/回款/金额/岗位)→ 公式纯函数 → 出参(含所用参数)。""" + # 参数按回款到账日取值(对照表 #11) + snapshot = resolve_snapshot(session, payload.receipt_date) + revenue = Decimal(payload.revenue) + profit = Decimal(payload.profit) + received = Decimal(payload.received_amount) + used: list[ParamUsedOut] = [] + + def take(code: str) -> Fraction: + raw = snapshot.get( + code, + role_code=payload.business_role, + product_category=payload.product_category, + ) + used.append( + ParamUsedOut( + param_code=code, + value_text=raw, + effective_date=snapshot.effective_date_of( + code, + role_code=payload.business_role, + product_category=payload.product_category, + ), + ) + ) + return parse_fraction(raw) + + days = formula.payment_days(payload.ship_date, payload.receipt_date) + rate, rate_text = _preview_rate(session, payload, snapshot, days, take) + + profit_rate = formula.profit_rate(profit, revenue) + threshold = take("product.threshold") + converted = formula.converted_profit(revenue, profit_rate, threshold) + performance = formula.performance_amount(converted, rate) + + barter_applied = payload.barter + after_barter: Decimal | None = None + if barter_applied: + discount = take("barter.discount") + after_barter = formula.apply_barter(performance, discount) + + start_ratio = take("payment.start.ratio") + start_ok = formula.start_payment_ok(received, revenue, start_ratio) + weight_r = min(100, int(Fraction(received) / Fraction(revenue) * 100)) if revenue > 0 else 0 + weight = formula.payment_weight(weight_r) + base_amount = after_barter if barter_applied else performance + payable = to_decimal(Fraction(base_amount) * weight) + + probation_note: str | None = None + if payload.probation: + annual = take("probation.annual.threshold") + probation_note = f"试用期绩效照算不发放;年累计超 {to_decimal(annual)} 元部分年底一次性结算" + + return ParamPreviewOut( + payment_days=days, + rate=str(rate), + rate_text=rate_text, + profit_rate=str(profit_rate), + converted_profit=str(converted), + performance_amount=str(performance), + barter_applied=barter_applied, + performance_after_barter=str(after_barter) if after_barter is not None else None, + start_payment_ok=start_ok, + payment_weight_r=weight_r, + payment_weight=str(weight), + payable_amount=str(payable), + probation_note=probation_note, + params_used=used, + ) + + +# ==================== 导出 ==================== + + +def export_changes( + session: Session, *, start_date: date | None = None, end_date: date | None = None +) -> bytes: + """变更记录导出 Excel(openpyxl)。列:版本号/变更人/变更时间/生效日期/参数/维度/旧值/新值。""" + from openpyxl import Workbook + + rows = param_repository.list_export_rows(session, start_date=start_date, end_date=end_date) + workbook = Workbook() + sheet = workbook.active + sheet.title = "参数变更记录" + sheet.append( + [ + "版本号", + "变更人", + "变更时间", + "生效日期", + "参数编码", + "参数名称", + "维度值", + "旧值", + "新值", + ] + ) + for batch, value, definition in rows: + sheet.append( + [ + batch.version_no, + batch.operator_name, + batch.create_time.strftime("%Y-%m-%d %H:%M:%S"), + value.effective_date.strftime("%Y-%m-%d"), + definition.param_code, + definition.param_name, + value.dimension_value, + value.old_value_text, + value.value_text, + ] + ) + buffer = BytesIO() + workbook.save(buffer) + return buffer.getvalue() diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 0a7dce5..a23f11e 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -34,6 +34,7 @@ dependencies = [ # 图形验证码(纯 Python,依赖 Pillow) "captcha>=0.6", "pillow>=11.0", + "openpyxl>=3.1.5", ] [dependency-groups] diff --git a/backend/tests/unit/test_param_snapshot.py b/backend/tests/unit/test_param_snapshot.py index 8207964..551dbec 100644 --- a/backend/tests/unit/test_param_snapshot.py +++ b/backend/tests/unit/test_param_snapshot.py @@ -16,7 +16,10 @@ from app.services.param_service import ParamSnapshot def _snap(rows: list[tuple[str, str, str]]) -> ParamSnapshot: """构造快照。rows 须模拟仓储输出:同一 (code, dim) 的最新值在前。""" - return ParamSnapshot.from_rows(date(2026, 10, 1), rows) + # 仓储实际返回 4 元组(含生效日期);单测关注覆盖链,统一补同一个日期 + return ParamSnapshot.from_rows( + date(2026, 10, 1), [(code, dim, value, date(2026, 1, 1)) for code, dim, value in rows] + ) class TestCoverChain: diff --git a/backend/uv.lock b/backend/uv.lock index d8159fe..b1080f3 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -221,6 +221,15 @@ wheels = [ { url = "https://pypi.tuna.tsinghua.edu.cn/packages/f6/b6/a1faf3a27ae9405fb34b1713cc73b2d8a26b04d5c561578fa2e6ef3e5bb9/cryptography-50.0.2-cp39-abi3-win_amd64.whl", hash = "sha256:4e81d95e5bafc2d6e34e4bed780e53e4d5b9a2f928573428aa4d35fbec1eb0de", size = 3854095, upload-time = "2026-09-30T15:29:46.782Z" }, ] +[[package]] +name = "et-xmlfile" +version = "2.0.0" +source = { registry = "https://pypi.tuna.tsinghua.edu.cn/simple" } +sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/d3/38/af70d7ab1ae9d4da450eeec1fa3918940a5fafb9055e934af8d6eb0c2313/et_xmlfile-2.0.0.tar.gz", hash = "sha256:dab3f4764309081ce75662649be815c4c9081e88f0837825f90fd28317d4da54", size = 17234, upload-time = "2024-10-25T17:25:40.039Z" } +wheels = [ + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/c1/8b/5fe2cc11fee489817272089c4203e679c63b570a5aaeb18d852ae3cbba6a/et_xmlfile-2.0.0-py3-none-any.whl", hash = "sha256:7a91720bc756843502c3b7504c77b8fe44217c85c537d85037f0f536151b2caa", size = 18059, upload-time = "2024-10-25T17:25:39.051Z" }, +] + [[package]] name = "fakeredis" version = "2.39.0" @@ -378,6 +387,18 @@ wheels = [ { url = "https://pypi.tuna.tsinghua.edu.cn/packages/c5/8e/4c24208776a65878d656996945aacfbfe010d3720d1a98fc0eb8491fc03b/markupsafe-3.0.4-cp312-cp312-win_arm64.whl", hash = "sha256:a4bbd2d87dd233b9fc5812160c3d0ffbe42edc22a26ce0469f58479ede633fe9", size = 14174, upload-time = "2026-10-02T23:05:07.617Z" }, ] +[[package]] +name = "openpyxl" +version = "3.1.5" +source = { registry = "https://pypi.tuna.tsinghua.edu.cn/simple" } +dependencies = [ + { name = "et-xmlfile" }, +] +sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/3d/f9/88d94a75de065ea32619465d2f77b29a0469500e99012523b91cc4141cd1/openpyxl-3.1.5.tar.gz", hash = "sha256:cf0e3cf56142039133628b5acffe8ef0c12bc902d2aadd3e0fe5878dc08d1050", size = 186464, upload-time = "2024-06-28T14:03:44.161Z" } +wheels = [ + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/c0/da/977ded879c29cbd04de313843e76868e6e13408a94ed6b987245dc7c8506/openpyxl-3.1.5-py2.py3-none-any.whl", hash = "sha256:5282c12b107bffeef825f4617dc029afaf41d0ea60823bbb665ef3079dc79de2", size = 250910, upload-time = "2024-06-28T14:03:41.161Z" }, +] + [[package]] name = "opentelemetry-api" version = "1.45.1" @@ -407,6 +428,7 @@ dependencies = [ { name = "alembic" }, { name = "captcha" }, { name = "fastapi" }, + { name = "openpyxl" }, { name = "pillow" }, { name = "pwdlib", extra = ["bcrypt"] }, { name = "pydantic" }, @@ -433,6 +455,7 @@ requires-dist = [ { name = "alembic", specifier = ">=1.14" }, { name = "captcha", specifier = ">=0.6" }, { name = "fastapi", specifier = ">=0.115" }, + { name = "openpyxl", specifier = ">=3.1.5" }, { name = "pillow", specifier = ">=11.0" }, { name = "pwdlib", extras = ["bcrypt"], specifier = ">=0.2" }, { name = "pydantic", specifier = ">=2.9" },