"""高级运费规则：报价引擎（定价模式 → 阶梯 → 聚合 → 隐藏边 → 排序 → 快照）。 放在独立模块而不是塞进 services.py：services.py 已经承担「事实构建 + 条件 DSL + locale 回退 + 公式变量表」四件事约 520 行，再加引擎会到 900+。一个文件一个职责更好维护。 ── 区间词汇（务必与 Task 2 对齐）────────────────────────────────────────────── **条件**区间写作 {min, max}，**阶梯**写作 {start, end}，两者语义完全一致： 起点闭、终点开（start <= v < end）。两端都闭会让边界值同时落进相邻两档， 累进计算直接把那一档收两遍；两端都开则边界值无人认领，直接掉成"没有报价"。 ── inherit：显式委派，不是"没匹配上的兜底"──────────────────────────────────── mode="inherit" 表示这条规则的价格由核心 ShippingCalculator.calculate_for_order(method_id=...) 用同一套权威事实算出，可审计。两个必须记住的后果： 1. calculate_for_order 在方案不存在或 is_active=0 时抛 ValueError （app/core/services/shipping_calculator.py:148-149）。本引擎会接住它并给出 reason 'rule_disabled' —— 事后被停用的方案只能降级成"被拒绝的候选"，绝不能变成 500。 2. calculate_for_order 内部已经调用过 _apply_surcharges，**inherit 拿到的 fee 已经含核心 ShippingSurcharge 行**。绝对不要再把同一笔附加费建成 sum 聚合的另一个成员，那是重复收费。 Task 4 写 Admin 表单/文案时请把这条写进提示。 ── 金额与税 ──────────────────────────────────────────────────────────────── 规则金额一律是租户本位币，RuleQuote.currency 带的就是币种代码。选中的 fee 原样作为 shipping_total 交给既有税务计算器，和 legacy 运费走同一条路——本插件不引入第二套税率概念。 """ from __future__ import annotations from dataclasses import dataclass, field from decimal import ROUND_CEILING, ROUND_HALF_UP, Decimal, DecimalException from typing import Any, Awaitable, Callable, Mapping, Sequence from .expression import FormulaError, evaluate_formula from .services import ( CartFacts, ConditionOutcome, _dec, build_formula_variables, evaluate_conditions, ) _ZERO = Decimal("0") _CENT = Decimal("0.01") #: 引擎支持的聚合策略。cheapest/cumulative 是历史写法，等价于 lowest/sum。 AGGREGATE_STRATEGIES: tuple[str, ...] = ( "independent", "sum", "cumulative", "lowest", "cheapest", "highest", "average", "all_required", ) _STRATEGY_ALIASES = {"cheapest": "lowest", "cumulative": "sum"} def canonical_strategy(name: str) -> str: """把历史别名折叠成引擎内部用的策略名（cheapest→lowest，cumulative→sum）。 写库层校验"同组不许混策略"时必须用同一套折叠，否则 sum 和 cumulative 会被当成两种。 """ return _STRATEGY_ALIASES.get(name, name) class PricingRejected(Exception): """条件命中了但价格算不出来。reason 一定是 schemas.REASON_KEYS 里的 key。""" def __init__(self, reason: str): super().__init__(reason) self.reason = reason # ── 规则集 ────────────────────────────────────────────────────────────────── @dataclass(frozen=True) class RuleSpec: """一条规则在引擎里的只读形态。可 JSON 往返，所以能整套丢进 Redis。""" id: int shipping_method_id: int delivery_mode: str quote_group: str | None aggregate_strategy: str presentation: Mapping[str, Any] conditions: Mapping[str, Any] pricing: Mapping[str, Any] priority: int revision_no: int enabled: bool = True @classmethod def from_row(cls, row) -> "RuleSpec": return cls( id=row.id, shipping_method_id=row.shipping_method_id, delivery_mode=row.delivery_mode, quote_group=row.quote_group, aggregate_strategy=row.aggregate_strategy, presentation=row.presentation or {}, conditions=row.conditions or {}, pricing=row.pricing or {}, priority=row.priority, revision_no=row.current_revision_no, enabled=bool(row.enabled), ) def to_json(self) -> dict: return { "id": self.id, "shipping_method_id": self.shipping_method_id, "delivery_mode": self.delivery_mode, "quote_group": self.quote_group, "aggregate_strategy": self.aggregate_strategy, "presentation": dict(self.presentation), "conditions": dict(self.conditions), "pricing": dict(self.pricing), "priority": self.priority, "revision_no": self.revision_no, "enabled": self.enabled, } @classmethod def from_json(cls, data: Mapping[str, Any]) -> "RuleSpec": return cls(**{k: data[k] for k in ( "id", "shipping_method_id", "delivery_mode", "quote_group", "aggregate_strategy", "presentation", "conditions", "pricing", "priority", "revision_no", "enabled", )}) @dataclass(frozen=True) class RuleSet: """租户的「已校验、已启用」规则集 + 隐藏边。缓存的就是它，不含任何客户信息。""" rules: tuple[RuleSpec, ...] = () hides: tuple[tuple[int, int], ...] = () def to_json(self) -> dict: return {"rules": [r.to_json() for r in self.rules], "hides": [[a, b] for a, b in self.hides]} @classmethod def from_json(cls, data: Mapping[str, Any]) -> "RuleSet": return cls( rules=tuple(RuleSpec.from_json(r) for r in data["rules"]), hides=tuple((int(a), int(b)) for a, b in data.get("hides") or ()), ) # ── 报价 ──────────────────────────────────────────────────────────────────── @dataclass(frozen=True) class RuleQuote: shipping_method_id: int quote_group: str | None rule_ids: tuple[int, ...] #: 与 rule_ids 逐位对齐的 current_revision_no，用于审计"当时用的是哪一版规则" revision_ids: tuple[int, ...] delivery_mode: str fee: Decimal currency: str title_map: Mapping[str, str] = field(default_factory=dict) description_map: Mapping[str, str] = field(default_factory=dict) reason_keys: tuple[str, ...] = () promo_free_shipping: bool = False priority: int = 100 @dataclass(frozen=True) class QuoteSet: quotes: tuple[RuleQuote, ...] = () rejected: tuple[RuleQuote, ...] = () @property def reason_keys(self) -> tuple[str, ...]: """去重后的拒绝原因。一条原因都没有又没出价时，明确回 no_shipping_quote—— 强制模式下没有兜底：不回落 ShippingCalculator，也不给租户默认运费。""" seen: list[str] = [] for candidate in self.rejected: for key in candidate.reason_keys: if key not in seen: seen.append(key) if not self.quotes and not seen: return ("no_shipping_quote",) return tuple(seen) @property def selected(self) -> RuleQuote | None: return self.quotes[0] if self.quotes else None # ── 定价 ──────────────────────────────────────────────────────────────────── def _money(value, label: str = "金额") -> Decimal: try: return _dec(value) except ValueError as exc: raise PricingRejected("pricing_failed") from exc def _quantize(value: Decimal) -> Decimal: try: return value.quantize(_CENT, rounding=ROUND_HALF_UP) except DecimalException as exc: raise PricingRejected("pricing_failed") from exc #: 阶梯模式 → 被分档的度量值。matched_* 只看命中的行。 _TIER_MEASURES: dict[str, Callable[[CartFacts, ConditionOutcome, Mapping], Decimal]] = { "tiered_quantity": lambda f, o, p: f.cart_unit_quantity, "tiered_weight": lambda f, o, p: f.cart_weight_kg, "tiered_volume": lambda f, o, p: f.cart_volume_cm3, "tiered_total": lambda f, o, p: f.cart_total, "tiered_subtotal": lambda f, o, p: f.cart_subtotal, "tiered_matched_quantity": lambda f, o, p: o.matched_unit_quantity, "tiered_matched_weight": lambda f, o, p: o.matched_weight_kg, "tiered_matched_total": lambda f, o, p: o.matched_total, # 体积重与实重取大者，除数与核心 ShippingMethod.dimensional_divisor 同义 "tiered_dimensional_weight": lambda f, o, p: max( f.cart_weight_kg, f.cart_volume_cm3 / _divisor(p.get("dimensional_divisor")), ), } def _divisor(value) -> Decimal: """体积重除数：必须是正数。 `or 5000` 只能挡住 int 0，挡不住字符串 "0"——那是真值，会一路走到 Decimal 除法抛 DivisionByZero（ArithmeticError，不是 PricingRejected）， 在 Task 5 层就是一个 500。写库边界不校验这个字段，所以只能在这里 fail-closed。 负除数同理：会让体积重变负、被 max() 静默吃掉，等于体积白算。 """ if value in (None, "", 0): return Decimal(5000) divisor = _money(value, "体积重除数") if divisor <= _ZERO: raise PricingRejected("pricing_failed") return divisor #: 百分比定价的基数。刻意只开这三个：按重量/件数算百分比没有业务含义。 _PERCENT_BASES: dict[str, Callable[[CartFacts, ConditionOutcome], Decimal]] = { "cart_subtotal": lambda f, o: f.cart_subtotal, "cart_total": lambda f, o: f.cart_total, "matched_total": lambda f, o: o.matched_total, } def _tier_contribution(tier: Mapping[str, Any], units: Decimal) -> Decimal: """一档的贡献：proportional=按单位计价，block=按整块向上取整，否则固定金额。""" amount = _money(tier.get("amount", 0)) if tier.get("proportional"): return amount * units block = tier.get("block") if block not in (None, "", 0): size = _money(block) if size <= _ZERO: raise PricingRejected("pricing_failed") blocks = (units / size).to_integral_value(rounding=ROUND_CEILING) return amount * max(blocks, _ZERO) return amount def _tier_fee(pricing: Mapping[str, Any], value: Decimal) -> Decimal: """按 start 闭 / end 开 定档。single 只收命中档，cumulative 逐档累加。""" tiers = pricing.get("tiers") if not isinstance(tiers, Sequence) or isinstance(tiers, (str, bytes)) or not tiers: raise PricingRejected("pricing_failed") cumulative = str(pricing.get("calculation") or "single") == "cumulative" parts: list[Decimal] = [] for tier in tiers: if not isinstance(tier, Mapping): raise PricingRejected("pricing_failed") start = _money(tier.get("start", 0)) raw_end = tier.get("end") end = None if raw_end in (None, "") else _money(raw_end) if end is not None and end <= start: raise PricingRejected("pricing_failed") if value < start: continue if not cumulative: if end is not None and value >= end: # end 开区间：正好等于上界属于下一档 continue return _tier_contribution(tier, value - start) upper = value if end is None else min(value, end) parts.append(_tier_contribution(tier, upper - start)) if parts: return sum(parts, _ZERO) # 值落在所有档之外（例如超过最后一个有上界的档）——不许静默变成免运费 raise PricingRejected("no_shipping_quote") def _waives_on_promo(pricing: Mapping[str, Any]) -> bool: """规则对"促销免运费"的态度：waive_all=允许促销把本报价归零，其余（默认）一律保留。 默认 preserve 是有意的 fail-closed：偏远地区附加费、易碎品操作费这类高级运费 不该被一张"全场包邮"券静默抹掉——那是直接的收入漏损。 """ return str(pricing.get("promo_free_shipping") or "preserve") == "waive_all" def price_rule( pricing: Mapping[str, Any], facts: CartFacts, outcome: ConditionOutcome, *, inherit_fee: Decimal | None = None, ) -> tuple[Decimal, bool]: """算一条规则的运费。返回 (fee, promo_free_shipping)。 inherit_fee 是核心 ShippingCalculator 已算好的运费（**已含 ShippingSurcharge**）， 由 evaluate_rule_set 解析后传入；mode="formula" 时它就是公式里的 shipping_fee 变量。 失败一律抛 PricingRejected（带 REASON_KEYS 里的 key），不抛 500。 """ if not isinstance(pricing, Mapping): raise PricingRejected("pricing_failed") if pricing.get("free_shipping"): return _ZERO.quantize(_CENT), True mode = str(pricing.get("mode") or "flat") if mode == "inherit": if inherit_fee is None: # 方案被停用/未提供解析器：降级成被拒候选，绝不 500 raise PricingRejected("rule_disabled") raw = inherit_fee elif mode == "flat": raw = _money(pricing.get("amount", 0)) elif mode == "percentage": base_key = str(pricing.get("base") or "cart_subtotal") if base_key not in _PERCENT_BASES: raise PricingRejected("pricing_failed") raw = _PERCENT_BASES[base_key](facts, outcome) * _money(pricing.get("percentage", 0)) / 100 elif mode == "formula": variables = build_formula_variables(facts, outcome, inherit_fee or _ZERO) try: raw = evaluate_formula(str(pricing.get("formula") or ""), variables) except FormulaError as exc: # 运行期除零/脏取值 → 'pricing_failed'，与"没规则出价"区分开 raise PricingRejected("pricing_failed") from exc elif mode in _TIER_MEASURES: raw = _tier_fee(pricing, _TIER_MEASURES[mode](facts, outcome, pricing)) else: raise PricingRejected("pricing_failed") modifier = pricing.get("modifier") or {} if not isinstance(modifier, Mapping): raise PricingRejected("pricing_failed") if modifier.get("multiply") is not None: raw = raw * _money(modifier["multiply"]) if modifier.get("add") is not None: raw = raw + _money(modifier["add"]) if pricing.get("floor") is not None: raw = max(raw, _money(pricing["floor"])) if pricing.get("cap") is not None: raw = min(raw, _money(pricing["cap"])) # expression.evaluate_formula 有意不给结果兜底（"5 - cart_total" 可以是负数）， # 归零是本层的责任：负运费没有业务含义，也会把订单总额算成倒贴。 return _quantize(max(raw, _ZERO)), _waives_on_promo(pricing) # ── 聚合与求值 ────────────────────────────────────────────────────────────── InheritResolver = Callable[[int], Awaitable[Decimal]] def _needs_inherit(pricing: Mapping[str, Any]) -> bool: mode = str(pricing.get("mode") or "flat") if mode == "inherit": return True return mode == "formula" and "shipping_fee" in str(pricing.get("formula") or "") def _strategy(rule: RuleSpec) -> str: return _STRATEGY_ALIASES.get(rule.aggregate_strategy, rule.aggregate_strategy) def _group_key(rule: RuleSpec): """聚合 = (shipping_method_id, delivery_mode, quote_group)。 independent 或没有分组的规则各自成组（把规则 id 掺进 key）。""" if _strategy(rule) == "independent" or not rule.quote_group: return (rule.shipping_method_id, rule.delivery_mode, None, rule.id) return (rule.shipping_method_id, rule.delivery_mode, rule.quote_group, None) def _reject(rule: RuleSpec, facts: CartFacts, reason: str) -> RuleQuote: return RuleQuote( shipping_method_id=rule.shipping_method_id, quote_group=rule.quote_group, rule_ids=(rule.id,), revision_ids=(rule.revision_no,), delivery_mode=rule.delivery_mode, fee=_ZERO, currency=facts.base_currency, title_map=dict(rule.presentation.get("name") or {}), description_map=dict(rule.presentation.get("description") or {}), reason_keys=(reason,), priority=rule.priority, ) async def evaluate_rule_set( rule_set: RuleSet, facts: CartFacts, *, inherit_resolver: InheritResolver | None = None, ) -> QuoteSet: """对整套规则求值，返回可选报价与被拒候选。 inherit_resolver(method_id) -> Decimal 由调用方注入（见 make_inherit_resolver）， 抛 ValueError 表示方案已停用 → 该候选降级为 reason 'rule_disabled'。 """ rejected: list[RuleQuote] = [] groups: dict[Any, list[tuple[RuleSpec, Decimal | None, bool]]] = {} group_members: dict[Any, list[RuleSpec]] = {} for rule in sorted(rule_set.rules, key=lambda r: (r.priority, r.id)): key = _group_key(rule) group_members.setdefault(key, []).append(rule) if not rule.enabled: rejected.append(_reject(rule, facts, "rule_disabled")) group_members[key].pop() # 停用规则不参与 all_required 的"全员命中" continue if rule.delivery_mode != facts.delivery_mode: rejected.append(_reject( rule, facts, "pickup_required" if rule.delivery_mode == "pickup" else "shipping_required")) continue try: outcome = evaluate_conditions(rule.conditions, facts) except ValueError: rejected.append(_reject(rule, facts, "pricing_failed")) continue if not outcome.matched: rejected.append(_reject(rule, facts, outcome.reason or "no_shipping_quote")) continue inherit_fee: Decimal | None = None if _needs_inherit(rule.pricing) and inherit_resolver is not None: try: inherit_fee = _money(await inherit_resolver(rule.shipping_method_id)) except ValueError: # calculate_for_order 对不存在/已停用的方案抛 ValueError（shipping_calculator.py:148） rejected.append(_reject(rule, facts, "rule_disabled")) continue try: fee, promo = price_rule(rule.pricing, facts, outcome, inherit_fee=inherit_fee) except PricingRejected as exc: rejected.append(_reject(rule, facts, exc.reason)) continue groups.setdefault(key, []).append((rule, fee, promo)) quotes: list[RuleQuote] = [] for key, matched in groups.items(): strategy = _strategy(matched[0][0]) if strategy == "all_required" and len(matched) != len(group_members.get(key, [])): continue # 有成员没命中 → 整组不出价，原因已在 rejected 里 fees = [fee for _r, fee, _p in matched] if strategy in ("sum", "all_required"): total = sum(fees, _ZERO) elif strategy == "lowest": total = min(fees) elif strategy == "highest": total = max(fees) elif strategy == "average": total = sum(fees, _ZERO) / len(fees) else: # independent：一组只有一条 total = fees[0] lead = matched[0][0] quotes.append(RuleQuote( shipping_method_id=lead.shipping_method_id, quote_group=lead.quote_group, rule_ids=tuple(r.id for r, _f, _p in matched), revision_ids=tuple(r.revision_no for r, _f, _p in matched), delivery_mode=lead.delivery_mode, fee=_quantize(total), currency=facts.base_currency, title_map=dict(lead.presentation.get("name") or {}), description_map=dict(lead.presentation.get("description") or {}), promo_free_shipping=any(p for _r, _f, p in matched), priority=lead.priority, )) # 隐藏边：命中的规则隐藏它声明的目标。hidden 集合一次性算完再过滤， # 互相隐藏的两条规则会一起消失——结果与规则书写顺序无关。 matched_ids = {rid for q in quotes for rid in q.rule_ids} hidden = {b for a, b in rule_set.hides if a in matched_ids} if hidden: quotes = [q for q in quotes if not hidden & set(q.rule_ids)] quotes.sort(key=lambda q: (q.priority, q.fee, q.shipping_method_id, q.quote_group or "", q.rule_ids)) return QuoteSet(quotes=tuple(quotes), rejected=tuple(rejected)) async def quote_facts( db, tenant_id: int, facts: CartFacts, *, cache_version: int, ) -> tuple[RuleSet, QuoteSet]: """事实 → 报价的**唯一**组合路径：取规则集 → 求值 → inherit 委派给核心运费计算器。 结账（Task 5）与 Admin 模拟（Task 4）都必须走这里。抄成两份三行代码， 迟早一边改了另一边没改，模拟和真实结账就会给出不同的运费。 返回 (rule_set, quote_set)：模拟要用规则集做"一条都没命中"的诊断， 结账不需要，直接 `_, quote_set = await quote_facts(...)` 即可—— 这样规则集只加载一次，不必为了诊断再查一遍。 导入放在函数体内：cache.py 反向 import 本模块（RuleSet/RuleSpec）， 模块级 import 会循环；ShippingCalculator 同理避免测试收集期副作用。 """ from app.core.services.shipping_calculator import ShippingCalculator from .cache import get_rule_set rule_set = await get_rule_set(db, tenant_id, cache_version=cache_version) quote_set = await evaluate_rule_set( rule_set, facts, inherit_resolver=make_inherit_resolver(ShippingCalculator(db, tenant_id), facts), ) return rule_set, quote_set def make_inherit_resolver(calculator, facts: CartFacts) -> InheritResolver: """把 ShippingCalculator 包成 inherit 解析器：同一套权威事实，显式委派。 calculator 是 app.core.services.shipping_calculator.ShippingCalculator 实例。 它返回的 fee **已含核心 ShippingSurcharge**，不要在规则里再加一遍。 """ async def resolve(method_id: int) -> Decimal: return await calculator.calculate_for_order( method_id=method_id, country=facts.country, province=facts.province, subtotal=facts.cart_subtotal, total_weight_kg=facts.cart_weight_kg, total_items=int(facts.cart_unit_quantity), length_cm=facts.max_length_cm, width_cm=facts.max_width_cm, height_cm=facts.max_height_cm, ) return resolve # ── 快照 ──────────────────────────────────────────────────────────────────── def build_quote_snapshot(quote: RuleQuote, facts: CartFacts, *, mode: str) -> dict: """报价的不可变快照：全部是原始类型，不引用任何活对象，落库后不再改。 shipping_total 就是原样交给既有税务计算器的那个金额——本插件不做第二套税。 """ return { "mode": mode, "shipping_method_id": quote.shipping_method_id, "quote_group": quote.quote_group, "delivery_mode": quote.delivery_mode, "rule_ids": list(quote.rule_ids), "revision_ids": list(quote.revision_ids), "shipping_total": str(quote.fee), "currency": quote.currency, "promo_free_shipping": bool(quote.promo_free_shipping), "reason_keys": list(quote.reason_keys), "title_map": dict(quote.title_map), "facts": { "destination": { "country": facts.country, "province": facts.province, "city": facts.city, "district": facts.district, "postcode": f"{facts.postcode[:2]}***" if facts.postcode else "", }, "cart": { "line_count": facts.cart_line_count, "unit_quantity": str(facts.cart_unit_quantity), "weight_kg": str(facts.cart_weight_kg), "subtotal": str(facts.cart_subtotal), "total": str(facts.cart_total), }, "payment_method": facts.payment_method, "customer_group_ids": sorted(facts.customer_group_ids), "coupon_applied": bool(facts.coupon_codes), }, "operating_timezone": facts.operating_timezone, "quoted_at": facts.local_now.isoformat(), } __all__ = [ "AGGREGATE_STRATEGIES", "PricingRejected", "QuoteSet", "RuleQuote", "RuleSet", "RuleSpec", "build_quote_snapshot", "canonical_strategy", "evaluate_rule_set", "make_inherit_resolver", "price_rule", "quote_facts", ]