"""计提参数服务:时点化取值(解析部分)。 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))