"""高级运费规则 Pydantic 契约：租户配额、聚合不变式、时区校验。 写库前必须经过本模块校验；报价求值/公式引擎见后续任务，这里只定义边界与形状。 """ import json import math from collections import Counter from typing import Any, Literal from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from pydantic import BaseModel, Field, field_validator, model_validator from .expression import MAX_FORMULA_CHARS, FormulaError, validate_formula # ── 租户硬性配额（写库前必须校验）────────────────────────────────────────── MAX_RULES_PER_TENANT = 200 MAX_TIERS_PER_RULE = 50 MAX_JSON_BYTES = 32 * 1024 MAX_SELECTOR_IDS = 200 MAX_POSTCODE_PATTERNS = 64 MAX_HIDE_EDGES_PER_RULE = 20 MAX_CONDITION_DEPTH = 32 # fail-closed 递归上限，防止病态嵌套触发 RecursionError（见 32KiB 上限无法拦截的深嵌套小 payload） # 计数类配额：key 在 conditions/pricing 任意深度、任意出现次数，都汇总进同一个 Counter 后统一比较一次。 # （之前按"每次命中"单独比较，导致同一 key 分片/多处出现/跨 conditions+pricing 能绕过。） _LIMITS = { "category_ids": MAX_SELECTOR_IDS, "product_ids": MAX_SELECTOR_IDS, "brand_ids": MAX_SELECTOR_IDS, "variant_ids": MAX_SELECTOR_IDS, "customer_group_ids": MAX_SELECTOR_IDS, "postcode_patterns": MAX_POSTCODE_PATTERNS, "tiers": MAX_TIERS_PER_RULE, } #: 本次发布不支持的条件 key。auto_dispatch 只在订单落库之后才分箱， #: 结账时根本不存在包裹数，任何 package_count 规则都无法诚实定价——显式拒绝，不静默忽略。 UNSUPPORTED_CONDITION_KEYS = frozenset({"package_count"}) #: 冻结的拒绝原因枚举。引擎只返回这些 key，客户可见文案由 #: Admin (frontend/admin/src/locales/{en,zh}.json) 与 #: Store (frontend/store/i18n/locales/{en,zh}.json) 的 shippingReason 段渲染。 #: 新增 key 必须同时补齐这四个 locale 文件。 REASON_KEYS: tuple[str, ...] = ( "no_shipping_quote", # 没有任何规则出价（含数值区间不落在范围内） "zone_not_allowed", # 国家/省份不在允许范围 "city_not_allowed", # 城市不在允许范围 "district_not_allowed", # 区/郊区（NZ suburb）不在允许范围 "postcode_not_allowed", # 邮编不匹配 "product_scope_mismatch", # 商品/分类/品牌/规格范围不匹配 "customer_not_allowed", # 客户分组不在允许范围 "time_not_allowed", # 星期/时段不在允许范围（按 operating_timezone 判定） "payment_not_allowed", # 支付方式不在允许范围 "coupon_not_allowed", # 优惠券不在允许范围 "pickup_required", # 该规则只对自提生效 "shipping_required", # 该规则只对配送生效 "rule_disabled", # 规则未启用 "stale_quote", # 报价已过期，需要重新计算 "pricing_failed", # 条件命中了，但定价算不出来（公式运行期除零等）——与"没规则出价"不同 ) #: 数值区间条件的 key。值形状固定为 {"min": 数字, "max": 数字}（min 闭、max 开）。 #: services._RANGE_SOURCES 必须与本集合一致（有测试盯着）。 RANGE_CONDITION_KEYS = frozenset({ "subtotal", "total", "weight_kg", "volume_cm3", "unit_quantity", "line_count", }) def _check_number(value, label: str) -> None: """写库期就把非数字挡掉：留到结账时 Decimal() 才炸，抛的是 ArithmeticError， 调用方 except ValueError 接不住，整个租户的结账都会 500。""" if isinstance(value, bool) or not isinstance(value, (int, float, str)): raise ValueError(f"{label} 必须是数字，收到 {value!r}") if isinstance(value, str): try: parsed = float(value) except ValueError as exc: raise ValueError(f"{label} 必须是数字，收到 {value!r}") from exc else: parsed = float(value) # float("nan")/float("inf") 都是合法的 float，必须单独挡： # "nan" 存得进去，求值时 Decimal("nan") 比较会抛 InvalidOperation（ArithmeticError）。 if not math.isfinite(parsed): raise ValueError(f"{label} 必须是有限数字，收到 {value!r}") def _check_range_shape(key: str, value) -> None: if not isinstance(value, dict): raise ValueError(f"{key} 条件必须是 {{'min': .., 'max': ..}} 形状") for bound in ("min", "max"): if value.get(bound) is not None: _check_number(value[bound], f"{key}.{bound}") def _check_hours_shape(value) -> None: windows = value if isinstance(value, (list, tuple)) else [value] for window in windows: if not isinstance(window, dict): raise ValueError("hours 条件必须是 {'start': .., 'end': ..} 形状") for bound in ("start", "end"): if bound not in window: continue hour = window[bound] if isinstance(hour, bool) or not isinstance(hour, int) or not 0 <= hour <= 24: raise ValueError(f"hours.{bound} 必须是 0–24 的整数，收到 {hour!r}") LocaleMap = dict[str, str] def _json_byte_len(*parts: Any) -> int: try: return sum( len(json.dumps(p, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) for p in parts ) except TypeError as exc: raise ValueError(f"payload 包含无法序列化的值: {exc}") from exc def _check_depth(depth: int) -> None: if depth > MAX_CONDITION_DEPTH: raise ValueError(f"conditions/pricing 嵌套深度超过上限 {MAX_CONDITION_DEPTH}") def _count_list(items, depth: int) -> int: """list/tuple 内部计数：标量或 dict 各算 1 个条目；嵌套的 list/tuple（分片）展开继续求和。""" _check_depth(depth) total = 0 for item in items: if isinstance(item, (list, tuple)): total += _count_list(item, depth + 1) else: total += 1 return total def _count_size(value, depth: int) -> int: """命中配额 key 后，统计其 value 贡献的总条目数： - list/tuple：委托给 _count_list（内部分片继续展开求和） - dict：include/exclude 或"用 key 模拟集合"的包装——逐个 value 计数， value 是 list/tuple 才展开求和，否则算 1 个条目（覆盖纯 dict 计数场景） - 标量：不计数""" _check_depth(depth) if isinstance(value, (list, tuple)): return _count_list(value, depth + 1) if isinstance(value, dict): return sum( _count_list(v, depth + 1) if isinstance(v, (list, tuple)) else 1 for v in value.values() ) return 0 def _accumulate(node: Any, counts: Counter, depth: int = 0) -> None: """单趟遍历 conditions/pricing：命中 _LIMITS 里的 key 就把其 value 的条目数累加进共享的 counts （不在命中处立即比较——比较统一放到 _check_bounds 末尾一次性做，避免"每次命中重开配额"）。 formula 是字符串长度检查，不是计数，单独处理。""" _check_depth(depth) if isinstance(node, dict): for key, value in node.items(): if key in UNSUPPORTED_CONDITION_KEYS: raise ValueError( f"不支持的条件 {key}：包裹数在订单落库分箱后才存在，无法用于结账报价" ) if key in RANGE_CONDITION_KEYS: _check_range_shape(key, value) elif key == "hours": _check_hours_shape(value) if key in _LIMITS: counts[key] += _count_size(value, depth + 1) elif key == "formula" and isinstance(value, str): # 长度先判：超长源码不该再进 ast.parse if len(value) > MAX_FORMULA_CHARS: raise ValueError(f"formula 超过上限 {MAX_FORMULA_CHARS} 字符") try: validate_formula(value) except FormulaError as exc: raise ValueError(f"formula 非法: {exc}") from exc _accumulate(value, counts, depth + 1) elif isinstance(node, (list, tuple)): for item in node: _accumulate(item, counts, depth + 1) class RulePresentation(BaseModel): """用户可见文案必须是语言映射，禁止裸字符串。如 {"en-NZ": "Standard", "zh-CN": "标准"}。 实际展示时按租户默认语言回退，回退逻辑见后续任务。""" name: LocaleMap = Field(..., min_length=1) description: LocaleMap | None = None class AdvancedShippingRuleIn(BaseModel): shipping_method_id: int quote_group: str | None = Field(None, max_length=60) # 必须与 quote.AGGREGATE_STRATEGIES 一致：写库层收不下的策略，引擎永远跑不到（有测试盯着）。 # cheapest/cumulative 是 Task 2 的历史写法，等价于 lowest/sum，保留以免旧数据写不回去。 aggregate_strategy: Literal[ "independent", "sum", "cumulative", "lowest", "cheapest", "highest", "average", "all_required", ] = "independent" delivery_mode: Literal["standard", "express", "pickup"] = "standard" presentation: RulePresentation conditions: dict = Field(default_factory=dict) pricing: dict = Field(default_factory=dict) priority: int = Field(100, ge=0, le=100000) enabled: bool = False hide_rule_ids: list[int] = Field(default_factory=list, max_length=MAX_HIDE_EDGES_PER_RULE) @model_validator(mode="after") def _check_bounds(self): # 先查体积上限：体积检查不递归，天然对病态深嵌套安全； # 递归的 _accumulate 放后面，且自带深度上限兜底（见 MAX_CONDITION_DEPTH）。 size = _json_byte_len( self.conditions, self.pricing, self.presentation.model_dump(exclude_none=True) ) if size > MAX_JSON_BYTES: raise ValueError(f"conditions/pricing/presentation 合计超过 {MAX_JSON_BYTES} 字节") # conditions 与 pricing 共用同一个 counts：配额是"每规则"的， # 不能让同一个 key 在不同位置/不同字段各自重开一份额度。 counts: Counter = Counter() _accumulate(self.conditions, counts) _accumulate(self.pricing, counts) for key, limit in _LIMITS.items(): if counts[key] > limit: raise ValueError(f"{key} 超过每规则上限 {limit}") return self class AdvancedShippingRuleAggregateIn(BaseModel): """一个报价聚合 = (shipping_method_id, delivery_mode, quote_group)。 成员规则必须与聚合共享同一 shipping_method_id / delivery_mode / quote_group。""" shipping_method_id: int delivery_mode: str quote_group: str members: list[AdvancedShippingRuleIn] = Field(..., min_length=1) @model_validator(mode="after") def _check_aggregate_invariant(self): for member in self.members: if member.shipping_method_id != self.shipping_method_id: raise ValueError("聚合成员的 shipping_method_id 必须与聚合一致") if member.delivery_mode != self.delivery_mode: raise ValueError("聚合成员的 delivery_mode 必须与聚合一致") if member.quote_group != self.quote_group: raise ValueError("聚合成员的 quote_group 必须与聚合一致") return self class AdvancedShippingSettingsIn(BaseModel): mode: str = Field("shadow", pattern="^(shadow|enforced)$") operating_timezone: str = Field("Pacific/Auckland", max_length=64) cache_version: int = Field(1, ge=1) @field_validator("operating_timezone") @classmethod def _validate_iana_timezone(cls, v: str) -> str: try: ZoneInfo(v) except ZoneInfoNotFoundError as exc: raise ValueError(f"不是合法的 IANA 时区: {v!r}") from exc return v def check_tenant_rule_count(existing_count: int) -> None: """新建规则前的租户配额检查（含已禁用规则）。写库前调用。""" if existing_count >= MAX_RULES_PER_TENANT: raise ValueError(f"租户规则数已达上限 {MAX_RULES_PER_TENANT}")