"""高级运费规则：购物车事实（CartFacts）与条件求值。 本模块只做三件事： 1. 从调用方传入的权威数据一次性构建 CartFacts（不反向去 checkout 流程里捞数据）； 2. 按冻结的 reason key 枚举求值条件，只返回 key，不返回面向客户的文案； 3. locale map 的租户默认语言回退。 定价模式/阶梯/聚合/RuleQuote 属于 Task 3，不在这里。 ── 量词语义（必须严格区分）──────────────────────────────────────────────── - cart_unit_quantity = sum(PricingCartItem.qty)，即「可售单位件数」（5 箱 = 5）。 不是行数、不是 stock_qty_override（扣库存用的基础单位，如 120 瓶）、不是公斤、不是包裹数。 - cart_line_count 单独暴露，供需要行数的规则使用。 - matched_* 是同样的口径，只对命中的行求和。 ── 地址语义 ──────────────────────────────────────────────────────────────── 沿用订单现有字段：country / province / city / district(区、NZ 的 suburb) / postcode(recv_zip_code)。 NZ 历史地址若把 suburb 写进 city 且 district 为空，则仅在运费事实中归一化为 city=province、district=原 city；不会改写订单或客户地址。 ── 集合量词 ──────────────────────────────────────────────────────────────── any = 至少一行命中；all = include 里的每个 id 都在购物车里出现；only = 购物车里没有别的行。 """ import fnmatch from dataclasses import dataclass from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from typing import Any, Mapping, Sequence from zoneinfo import ZoneInfo # 租户默认语言复用既有 i18n_multilang 插件配置，不另发明一套设置 from app.services.email import _get_tenant_default_locale as get_tenant_default_locale from .schemas import UNSUPPORTED_CONDITION_KEYS _ZERO = Decimal("0") # ── 事实 ──────────────────────────────────────────────────────────────────── @dataclass(frozen=True) class CartLineFacts: index: int product_id: int variant_id: int | None quantity: Decimal # 可售单位件数（PricingCartItem.qty） category_ids: frozenset[int] brand_id: int | None unit_weight_kg: Decimal length_cm: Decimal width_cm: Decimal height_cm: Decimal line_total: Decimal @property def weight_kg(self) -> Decimal: return self.unit_weight_kg * self.quantity @property def volume_cm3(self) -> Decimal: return self.length_cm * self.width_cm * self.height_cm * self.quantity @dataclass(frozen=True) class CartFacts: lines: tuple[CartLineFacts, ...] cart_line_count: int cart_unit_quantity: Decimal cart_weight_kg: Decimal cart_volume_cm3: Decimal max_length_cm: Decimal max_width_cm: Decimal max_height_cm: Decimal cart_subtotal: Decimal cart_total: Decimal country: str province: str city: str district: str postcode: str customer_group_ids: frozenset[int] payment_method: str | None coupon_codes: frozenset[str] base_currency: str # 规则金额一律用租户本位币 delivery_mode: str operating_timezone: str local_now: datetime # 按 operating_timezone 换算后的本地时间 def _dec(value) -> Decimal: """转 Decimal。脏值一律 ValueError——不能漏成 decimal.InvalidOperation， 那是 ArithmeticError，Task 3/5 的 `except ValueError` 接不住，整租户结账 500。""" if value is None: return _ZERO if isinstance(value, Decimal): return value if isinstance(value, bool) or not isinstance(value, (int, float, str)): raise ValueError(f"不是合法数值: {value!r}") try: parsed = Decimal(str(value)) except InvalidOperation as exc: raise ValueError(f"不是合法数值: {value!r}") from exc # NaN/Inf 必须在这里挡住：Decimal("nan") < x 抛的是 InvalidOperation（ArithmeticError）， # 同样接不住；Inf 不抛异常但会把无穷大喂进下游定价。 if not parsed.is_finite(): raise ValueError(f"不是有限数值: {value!r}") return parsed def _ids(values, key: str) -> set[int]: """ID 选择器统一按 int 比较。 Admin 表单与 Task 7 的 AI 起草经常把 ID 写成 JSON 字符串（"1"）， 原样比较会静默不命中——那是"答案错了"，比报错更糟。不可转换则显式报错。 """ out: set[int] = set() for v in values: if isinstance(v, bool) or v is None: raise ValueError(f"{key} 只接受整数 ID，收到 {v!r}") try: out.add(int(v)) except (TypeError, ValueError) as exc: raise ValueError(f"{key} 只接受整数 ID，收到 {v!r}") from exc return out def _pick(variant, product, attr): """规格上有值就用规格的，否则回落到商品（规格只填了重量、没填尺寸是常见情况）。""" if variant is not None: v = getattr(variant, attr, None) if v is not None: return v return getattr(product, attr, None) def _addr(address: Mapping[str, Any] | None, field: str) -> str: """订单地址字段名以 recv_* 为准（recv_zip_code 是邮编），同时兼容裸字段名。""" if not address: return "" for key in (f"recv_{field}", field): value = address.get(key) if value not in (None, ""): return str(value) return "" def _shipping_destination(address: Mapping[str, Any] | None) -> tuple[str, str, str, str]: """返回报价专用地址事实；兼容 NZ 旧地址把 suburb 混写在 city 的情况。""" country = _addr(address, "country").strip() province = _addr(address, "province").strip() city = _addr(address, "city").strip() district = _addr(address, "district").strip() if (country.upper() == "NZ" and province and city and city.casefold() != province.casefold() and (not district or city.casefold() == district.casefold())): return country, province, province, district or city return country, province, city, district def build_cart_facts( *, items: Sequence[Any], products: Mapping[int, Any], line_totals: Sequence[Decimal], cart_subtotal: Decimal, cart_total: Decimal, variants: Mapping[int, Any] | None = None, address: Mapping[str, Any] | None = None, customer_group_ids: Sequence[int] = (), payment_method: str | None = None, coupon_codes: Sequence[str] = (), base_currency: str = "NZD", delivery_mode: str = "standard", operating_timezone: str = "Pacific/Auckland", now: datetime | None = None, ) -> CartFacts: """一次性构建 CartFacts。items 是 PricingCartItem 序列，products/variants 由调用方查好传入。 邮编来自 address['recv_zip_code']；件数只看 item.qty，绝不看 stock_qty_override。 金额一律没有"默认 0"，line_totals / cart_subtotal / cart_total 全部必填： 正常商城下单时 PricingCartItem.unit_price_override 是 None（只有 unit_split／改价商品才有值）， 任何金额兜底成 0 都会让区间条件与百分比／阶梯定价往"运费更便宜"的方向错，必须 fail-closed。 调用方请传 PricingResult 的小计、合计与每行 line_total。 products 里缺商品同样直接报错：缺失商品会变成 0 重量 0 体积，重量／体积阶梯直接掉到最便宜档。 空购物车（items 为空）是合法输入：无条件规则命中（matched=True），所有聚合量为 0； 但任何商品范围条件都会得到 product_scope_mismatch（没有行可命中）。 "空车该不该出运费"由 Task 3/5 的结账流程决定，不在事实层拦。 variant_id 在 items 上但不在 variants 里时，**有意**回落到商品自身的重量／尺寸： 规格通常只覆盖部分字段，缺规格行按商品估算比直接报错更贴近现实。 """ if len(line_totals) != len(items): raise ValueError("line_totals 必须与 items 逐行对齐") if any(t is None for t in line_totals): raise ValueError("line_totals 不允许为 None：金额不能默认成 0") variants = variants or {} lines: list[CartLineFacts] = [] for i, item in enumerate(items): product = products.get(item.product_id) if product is None: raise ValueError(f"缺少商品事实: product_id={item.product_id}") variant = variants.get(item.variant_id) if item.variant_id else None cats = {getattr(product, "category_id", None)} cats |= {getattr(c, "id", None) for c in (getattr(product, "categories", None) or [])} lines.append(CartLineFacts( index=i, product_id=item.product_id, variant_id=item.variant_id, quantity=_dec(item.qty), # 可售单位件数，不是 stock_qty_override category_ids=frozenset(c for c in cats if c is not None), brand_id=getattr(product, "brand_id", None), unit_weight_kg=_dec(_pick(variant, product, "weight")), length_cm=_dec(_pick(variant, product, "length")), width_cm=_dec(_pick(variant, product, "width")), height_cm=_dec(_pick(variant, product, "height")), line_total=_dec(line_totals[i]), )) utc_now = now or datetime.now(timezone.utc) if utc_now.tzinfo is None: utc_now = utc_now.replace(tzinfo=timezone.utc) country, province, city, district = _shipping_destination(address) return CartFacts( lines=tuple(lines), cart_line_count=len(lines), cart_unit_quantity=sum((ln.quantity for ln in lines), _ZERO), cart_weight_kg=sum((ln.weight_kg for ln in lines), _ZERO), cart_volume_cm3=sum((ln.volume_cm3 for ln in lines), _ZERO), max_length_cm=max((ln.length_cm for ln in lines), default=_ZERO), max_width_cm=max((ln.width_cm for ln in lines), default=_ZERO), max_height_cm=max((ln.height_cm for ln in lines), default=_ZERO), cart_subtotal=_dec(cart_subtotal), cart_total=_dec(cart_total), country=country, province=province, city=city, district=district, postcode=_addr(address, "zip_code") or _addr(address, "postcode"), customer_group_ids=frozenset(customer_group_ids or ()), payment_method=payment_method, coupon_codes=frozenset(coupon_codes or ()), base_currency=base_currency, delivery_mode=delivery_mode, operating_timezone=operating_timezone, local_now=utc_now.astimezone(ZoneInfo(operating_timezone)), ) # ── 求值结果 ──────────────────────────────────────────────────────────────── @dataclass(frozen=True) class ConditionOutcome: matched: bool reason: str | None # 冻结枚举里的 key，绝不是客户可见文案 matched_lines: tuple[CartLineFacts, ...] @property def matched_unit_quantity(self) -> Decimal: return sum((ln.quantity for ln in self.matched_lines), _ZERO) @property def matched_weight_kg(self) -> Decimal: return sum((ln.weight_kg for ln in self.matched_lines), _ZERO) @property def matched_volume_cm3(self) -> Decimal: return sum((ln.volume_cm3 for ln in self.matched_lines), _ZERO) @property def matched_total(self) -> Decimal: return sum((ln.line_total for ln in self.matched_lines), _ZERO) def _fail(reason: str) -> ConditionOutcome: return ConditionOutcome(False, reason, ()) # ── 条件定义表 ────────────────────────────────────────────────────────────── _SCALAR_SOURCES = { "country": (lambda f: f.country, "zone_not_allowed"), "province": (lambda f: f.province, "zone_not_allowed"), "city": (lambda f: f.city, "city_not_allowed"), "district": (lambda f: f.district, "district_not_allowed"), "payment_methods": (lambda f: f.payment_method or "", "payment_not_allowed"), } _SET_SOURCES = { # (取值, reason, 是否按整数 ID 比较) "customer_group_ids": (lambda f: f.customer_group_ids, "customer_not_allowed", True), "coupon_codes": (lambda f: f.coupon_codes, "coupon_not_allowed", False), } _LINE_VALUES = { "product_ids": lambda ln: {ln.product_id}, "variant_ids": lambda ln: {ln.variant_id} if ln.variant_id is not None else set(), "brand_ids": lambda ln: {ln.brand_id} if ln.brand_id is not None else set(), "category_ids": lambda ln: set(ln.category_ids), } _RANGE_SOURCES = { "subtotal": lambda f: f.cart_subtotal, "total": lambda f: f.cart_total, "weight_kg": lambda f: f.cart_weight_kg, "volume_cm3": lambda f: f.cart_volume_cm3, "unit_quantity": lambda f: f.cart_unit_quantity, # 件数，见模块 docstring "line_count": lambda f: Decimal(f.cart_line_count), } def _norm(value): return value.strip().casefold() if isinstance(value, str) else value def _selector(value) -> tuple[list, list, str]: """统一成 (include, exclude, mode)。裸 list 等价于 include + any。""" if isinstance(value, dict) and ({"include", "exclude", "mode"} & set(value)): return (list(value.get("include") or []), list(value.get("exclude") or []), str(value.get("mode") or "any")) if isinstance(value, (list, tuple, set, frozenset)): return list(value), [], "any" return [value], [], "any" def _scalar_ok(actual, value) -> bool: include, exclude, _ = _selector(value) a = _norm(actual) if include and (a in ("", None) or a not in {_norm(x) for x in include}): return False if exclude and a in {_norm(x) for x in exclude}: return False return True def _set_ok(actual, value, key: str = "", as_ids: bool = False) -> bool: include, exclude, mode = _selector(value) if as_ids: have, inc, exc = _ids(actual, key), _ids(include, key), _ids(exclude, key) else: have = {_norm(x) for x in actual} inc = {_norm(x) for x in include} exc = {_norm(x) for x in exclude} if inc: if mode == "all" and not inc <= have: return False if mode != "all" and not (inc & have): return False # only 的判定必须在 inc 之外：空 include + only = "一个都不许有"，不是无条件放行 if mode == "only" and not have <= inc: return False if exc and (exc & have): return False return True def _postcode_ok(facts: CartFacts, value) -> bool: include, exclude, _ = _selector(value) target = str(_norm(facts.postcode)).replace(" ", "") def hit(patterns): return any(fnmatch.fnmatchcase(target, str(_norm(p)).replace(" ", "")) for p in patterns) if include and (not target or not hit(include)): return False if exclude and target and hit(exclude): return False return True def _product_scope(key, value, facts: CartFacts) -> tuple[list[CartLineFacts], bool]: include, exclude, mode = _selector(value) inc, exc = _ids(include, key), _ids(exclude, key) # "1" 与 1 必须命中同一行 get = _LINE_VALUES[key] lines = [] for ln in facts.lines: vals = get(ln) if exc and (vals & exc): continue if inc and not (vals & inc): continue lines.append(ln) if not lines: return [], False if mode == "only" and len(lines) != len(facts.lines): return lines, False if mode == "all" and inc: present: set = set() for ln in facts.lines: present |= get(ln) if not inc <= present: return lines, False return lines, True def _range_ok(actual: Decimal, value) -> bool: """min 闭、max 开——相邻区间才能无缝拼接且不重叠。""" if not isinstance(value, Mapping): raise ValueError("区间条件必须是 {'min': .., 'max': ..} 形状") low, high = value.get("min"), value.get("max") if low is not None and actual < _dec(low): return False if high is not None and actual >= _dec(high): return False return True def _hours_ok(local_now: datetime, value) -> bool: windows = value if isinstance(value, (list, tuple)) else [value] hour = local_now.hour for window in windows: if not isinstance(window, Mapping): raise ValueError("hours 条件必须是 {'start': .., 'end': ..}") start, end = int(window.get("start", 0)), int(window.get("end", 24)) # 越界小时曾静默变成"永远命中"——写库期已挡（schemas._check_hours_shape），这里兜底 if not (0 <= start <= 24 and 0 <= end <= 24): raise ValueError(f"hours 必须落在 0–24，收到 start={start} end={end}") if start == end: return True # 全天 if start < end: if start <= hour < end: return True elif hour >= start or hour < end: # 跨午夜窗口，如 22:00–06:00 return True return False def _intersect(lines: list[CartLineFacts], other: Sequence[CartLineFacts]) -> list[CartLineFacts]: keep = {ln.index for ln in other} return [ln for ln in lines if ln.index in keep] def _eval_node(node, facts: CartFacts) -> ConditionOutcome: if not isinstance(node, Mapping): raise ValueError("conditions 必须是 dict") matched = list(facts.lines) for key, value in node.items(): if key in UNSUPPORTED_CONDITION_KEYS: raise ValueError(f"不支持的条件 {key}：包裹数在订单落库分箱后才存在，无法用于结账报价") if key == "any": outcomes = [_eval_node(sub, facts) for sub in (value or [])] hits = [o for o in outcomes if o.matched] if not hits: return outcomes[0] if outcomes else _fail("no_shipping_quote") # 取所有命中分支的并集，不是第一个命中的分支： # 只留第一个会让 matched_* 少算，且结果取决于租户 JSON 里分支的书写顺序。 union = [ln for o in hits for ln in o.matched_lines] matched = _intersect(matched, union) elif key == "all": for sub in (value or []): outcome = _eval_node(sub, facts) if not outcome.matched: return outcome matched = _intersect(matched, outcome.matched_lines) elif key in _SCALAR_SOURCES: source, reason = _SCALAR_SOURCES[key] if not _scalar_ok(source(facts), value): return _fail(reason) elif key in _SET_SOURCES: source, reason, as_ids = _SET_SOURCES[key] if not _set_ok(source(facts), value, key, as_ids): return _fail(reason) elif key == "postcode_patterns": if not _postcode_ok(facts, value): return _fail("postcode_not_allowed") elif key in _LINE_VALUES: lines, ok = _product_scope(key, value, facts) if not ok: return _fail("product_scope_mismatch") matched = _intersect(matched, lines) elif key in _RANGE_SOURCES: if not _range_ok(_RANGE_SOURCES[key](facts), value): return _fail("no_shipping_quote") elif key == "weekdays": include, exclude, _mode = _selector(value) weekday = facts.local_now.weekday() # 周一=0，按营业时区判定 if include and weekday not in {int(x) for x in include}: return _fail("time_not_allowed") if exclude and weekday in {int(x) for x in exclude}: return _fail("time_not_allowed") elif key == "hours": if not _hours_ok(facts.local_now, value): return _fail("time_not_allowed") elif key == "delivery_mode": allowed = {_norm(x) for x in (value if isinstance(value, (list, tuple)) else [value])} actual = _norm(facts.delivery_mode) if actual not in allowed: return _fail("pickup_required" if "pickup" in allowed else "shipping_required") else: raise ValueError(f"未知条件 key: {key}") return ConditionOutcome(True, None, tuple(matched)) def evaluate_conditions(conditions, facts: CartFacts, *, enabled: bool = True) -> ConditionOutcome: """求值一条规则的 conditions。返回的 reason 一定是 schemas.REASON_KEYS 里的 key。""" if not enabled: return _fail("rule_disabled") return _eval_node(conditions or {}, facts) # ── locale 回退 ───────────────────────────────────────────────────────────── def resolve_locale_text(locale_map: Mapping[str, str] | None, locale: str | None, default_locale: str | None) -> str: """语言映射取值：精确命中 → 语言前缀命中 → 租户默认语言 → 任意一条。 default_locale 由 get_tenant_default_locale(db, tenant_id) 提供（i18n_multilang 插件配置）。 """ if not locale_map: return "" for want in (locale, default_locale): if not want: continue if want in locale_map: return locale_map[want] prefix = str(want).split("-")[0].casefold() for key, text in locale_map.items(): if str(key).split("-")[0].casefold() == prefix: return text return next(iter(locale_map.values())) # ── 公式变量 → 事实字段的唯一映射 ─────────────────────────────────────────── #: 公式里的变量名与事实字段名**不同名**（cart_quantity ↔ cart_unit_quantity）， #: 手写第二份映射写错了不会报错，只会算错价。Task 3 必须从这里取，别再写一遍。 #: 值是 (来源, 属性名)：facts=CartFacts，outcome=ConditionOutcome，runtime=由 Task 3 现场提供。 FORMULA_VARIABLE_SOURCES: dict[str, tuple[str, str]] = { "cart_total": ("facts", "cart_total"), "cart_subtotal": ("facts", "cart_subtotal"), "cart_quantity": ("facts", "cart_unit_quantity"), # 件数口径，见模块 docstring "cart_weight_kg": ("facts", "cart_weight_kg"), "cart_volume_cm3": ("facts", "cart_volume_cm3"), "matched_total": ("outcome", "matched_total"), "matched_quantity": ("outcome", "matched_unit_quantity"), "matched_weight_kg": ("outcome", "matched_weight_kg"), "shipping_fee": ("runtime", "shipping_fee"), # 链式计费时的在算运费，默认 0 } def build_formula_variables(facts: CartFacts, outcome: ConditionOutcome, shipping_fee: Decimal = _ZERO) -> dict[str, Decimal]: """按 FORMULA_VARIABLE_SOURCES 生成公式变量表，供 expression.evaluate_formula 使用。""" source_objects = {"facts": facts, "outcome": outcome} return { name: shipping_fee if origin == "runtime" else getattr(source_objects[origin], attr) for name, (origin, attr) in FORMULA_VARIABLE_SOURCES.items() } __all__ = [ "CartFacts", "CartLineFacts", "ConditionOutcome", "FORMULA_VARIABLE_SOURCES", "build_cart_facts", "build_formula_variables", "evaluate_conditions", "get_tenant_default_locale", "resolve_locale_text", ]