diff --git a/backend/app/repositories/param_repository.py b/backend/app/repositories/param_repository.py new file mode 100644 index 0000000..8c461d6 --- /dev/null +++ b/backend/app/repositories/param_repository.py @@ -0,0 +1,46 @@ +"""计提参数数据访问。 + +本层只 flush 不 commit —— 事务边界由 services 层控制。 +时点化解析查询(窗口函数版)留存 db/queries/计提参数.sql。 +""" + +from datetime import date + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.models import ParamDefinition, ParamValue + +# ==================== 读取 ==================== + + +def resolve_values(session: Session, biz_date: date) -> list[tuple[str, str, str]]: + """时点化解析:每个 (param_id, dimension_value) 取生效日期 <= 业务日期的最新条目。 + + 排序 (effective_date DESC, id DESC) 保证同日多次变更取最新一条 + (不设日期唯一键,同日修正是合法留痕)。仅返回启用状态的参数定义。 + 返回 (param_code, dimension_value, value_text),供 ParamSnapshot 构造。 + """ + row_number = ( + func.row_number() + .over( + partition_by=(ParamValue.param_id, ParamValue.dimension_value), + order_by=(ParamValue.effective_date.desc(), ParamValue.id.desc()), + ) + .label("rn") + ) + subq = ( + select( + ParamDefinition.param_code.label("param_code"), + ParamValue.dimension_value.label("dimension_value"), + ParamValue.value_text.label("value_text"), + 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 + ) + return [(row.param_code, row.dimension_value, row.value_text) for row in session.execute(stmt)] diff --git a/backend/app/services/param_service.py b/backend/app/services/param_service.py new file mode 100644 index 0000000..a0b6bd8 --- /dev/null +++ b/backend/app/services/param_service.py @@ -0,0 +1,87 @@ +"""计提参数服务:时点化取值(解析部分)。 + +ParamSnapshot 是 M3 核算与计算预览共用的输入契约:按业务日期(回款到账日) +冻结当时口径,禁读「当前值」计算历史数据(算法红线)。 +变更提交/历史/对比/预览/导出在后续任务追加。 +""" + +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import date +from fractions import Fraction + +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 + + +@dataclass(frozen=True, slots=True) +class ParamSnapshot: + """某业务日期下的参数快照(冻结不可变,核算期间保持一致)。 + + 覆盖链:角色维度 > 产品类别维度 > 默认行('');均无匹配时回退注册表默认值。 + """ + + biz_date: date + _values: dict[str, dict[str, str]] + + @classmethod + def from_rows(cls, biz_date: date, rows: Iterable[tuple[str, str, str]]) -> "ParamSnapshot": + """由 (param_code, dimension_value, value_text) 行构造。 + + rows 须按仓储输出顺序(同一 (code, dim) 的最新生效值在前)——以先出现者为准。 + """ + values: dict[str, dict[str, str]] = {} + for code, dimension, value_text in rows: + values.setdefault(code, {}).setdefault(dimension, value_text) + return cls(biz_date=biz_date, _values=values) + + def get( + self, code: str, *, role_code: str | None = None, product_category: str | None = None + ) -> str: + """取参数原始值(覆盖链解析)。无值且无注册表默认值时抛 400。""" + candidates = self._values.get(code) + if candidates: + for dimension in (role_code, product_category, ""): + if dimension is not None and dimension in candidates: + return candidates[dimension] + spec = get_spec(code) # 未注册编码 → ValueError(代码缺陷) + if spec.default_value is None: + raise BadRequestError(f"参数 {code} 无生效值且无默认值,请先在系统设置中维护") + return spec.default_value + + def fraction( + self, code: str, *, role_code: str | None = None, product_category: str | None = None + ) -> Fraction: + """取参数值并精确解析为 Fraction("4/360" → 1/90,不经 float)。""" + return parse_fraction( + self.get(code, role_code=role_code, product_category=product_category) + ) + + def integer( + self, code: str, *, role_code: str | None = None, product_category: str | None = None + ) -> int: + """取参数值为整数。配置成非整数属数据缺陷(写入经校验,不应发生)。""" + value = self.fraction(code, role_code=role_code, product_category=product_category) + if value.denominator != 1: + raise ValueError(f"参数 {code} 配置为非整数:{value}") + return int(value) + + def boolean( + self, code: str, *, role_code: str | None = None, product_category: str | None = None + ) -> bool: + """取布尔参数(0/1)。""" + return self.get(code, role_code=role_code, product_category=product_category) == "1" + + def text( + self, code: str, *, role_code: str | None = None, product_category: str | None = None + ) -> str: + """取参数原文(枚举码/文本)。""" + return self.get(code, role_code=role_code, product_category=product_category) + + +def resolve_snapshot(session: Session, biz_date: date) -> ParamSnapshot: + """按业务日期解析全量参数快照(M3 核算与计算预览共用入口)。""" + return ParamSnapshot.from_rows(biz_date, resolve_values(session, biz_date)) diff --git a/backend/tests/unit/test_param_snapshot.py b/backend/tests/unit/test_param_snapshot.py new file mode 100644 index 0000000..8207964 --- /dev/null +++ b/backend/tests/unit/test_param_snapshot.py @@ -0,0 +1,92 @@ +"""参数快照(时点化取值)测试。 + +守的是:覆盖链与兜底 —— 一旦取错维度(如企业户取到市政阈值行) +或漏兜底,核算会用错口径或直接崩溃;同日多次变更必须取最新留痕。 +仓储 SQL 的排序语义(同日取 id 最大)在集成测试中覆盖。 +""" + +from dataclasses import FrozenInstanceError +from datetime import date +from fractions import Fraction + +import pytest + +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) + + +class TestCoverChain: + """覆盖链:角色维度 > 产品类别维度 > 默认行('')。""" + + def test_产品类别维度优先于默认行(self) -> None: + snap = _snap([("enterprise.rate", "", "3"), ("enterprise.rate", "municipal_water", "2")]) + assert snap.get("enterprise.rate", product_category="municipal_water") == "2" + + def test_无匹配维度时落默认行(self) -> None: + snap = _snap([("enterprise.rate", "", "3"), ("enterprise.rate", "municipal_water", "2")]) + assert snap.get("enterprise.rate", product_category="rail_transit") == "3" + + def test_角色维度优先于产品类别维度(self) -> None: + snap = _snap([("some.param", "municipal_water", "A"), ("some.param", "salesperson", "B")]) + got = snap.get("some.param", role_code="salesperson", product_category="municipal_water") + assert got == "B" + + def test_无任何维度匹配且无默认行_回退注册表默认值(self) -> None: + snap = _snap([("enterprise.rate", "municipal_water", "2")]) + # 轨交无行、无默认行 → 回退注册表 default_value(3) + assert snap.fraction("enterprise.rate", product_category="rail_transit") == Fraction(3) + + +class TestFallback: + """注册表默认值兜底(种子缺失时的安全网)。""" + + def test_未定义参数按注册表默认值兜底(self) -> None: + snap = _snap([]) + assert snap.fraction("prepay.rate") == Fraction(10) + + def test_未注册编码抛ValueError_属代码缺陷(self) -> None: + snap = _snap([]) + with pytest.raises(ValueError): + snap.fraction("not.registered.code") + + +class TestSameDayChanges: + """同日多次变更:仓储 SQL 已按 (effective_date, id) 倒序,快照取先出现者。""" + + def test_同日多次变更取最新一条(self) -> None: + snap = _snap([("enterprise.rate", "", "3"), ("enterprise.rate", "", "2")]) + assert snap.fraction("enterprise.rate") == Fraction(3) + + +class TestTypedGetters: + """类型化取值:Fraction 精确解析、整数、布尔、文本。""" + + def test_fraction精确解析分数式(self) -> None: + snap = _snap([("salesperson.rate.slope", "salesperson", "4/360")]) + assert snap.fraction("salesperson.rate.slope", role_code="salesperson") == Fraction(1, 90) + + def test_integer(self) -> None: + snap = _snap([("salesperson.days.boundary", "salesperson", "360")]) + assert snap.integer("salesperson.days.boundary", role_code="salesperson") == 360 + + def test_text返回枚举原文(self) -> None: + snap = _snap([("payment.weight.mode", "", "FULL")]) + assert snap.text("payment.weight.mode") == "FULL" + + def test_boolean(self) -> None: + snap = _snap([("some.flag", "", "1")]) + assert snap.boolean("some.flag") is True + assert snap.boolean("some.flag") is not False + + +class TestImmutability: + """快照冻结:解析结果不得被调用方改写(核算期间一致性)。""" + + def test_字段不可赋值(self) -> None: + snap = _snap([]) + with pytest.raises(FrozenInstanceError): + snap.biz_date = date(2026, 1, 1) # type: ignore[misc]