"""高级运费规则：Store 结账集成（服务端权威报价 + 运费优先级）。 ── 为什么是"Store 显式请求报价"而不是"定价服务自己发现插件"──────────────── calculate_pricing 是 B2B / POS / Admin 共用的核心定价入口。只要它自己去发现并激活 本插件，那三条链路就会在租户某天开启插件的瞬间集体改变行为。所以方向是反的： Store 显式算好报价，把 fee + 报价元数据传进 calculate_pricing。插件不存在或未启用时 Store 一次都不调用本模块的报价路径，legacy 行为逐字节不变。 ── 钱永远不来自客户端 ────────────────────────────────────────────────────── 客户端只提交购物车事实与一个**选择标识**（shipping_method_id + quote_group）。 金额、规则 id、修订号、配送模式、客户资格一律服务端重算；客户端传来的 fee/rule_id 被完整忽略。选择对不上当前候选就是 stale_quote，不是"按客户端说的算"。 ── 强制模式没有兜底 ──────────────────────────────────────────────────────── 没有任何规则出价 = 结不了账（400 + reason key），绝不回落 ShippingCalculator 或 租户默认运费。那条回落正是"区域限制被绕过"的入口。 ── 影子模式只观测 ────────────────────────────────────────────────────────── 影子对比跑在 legacy 定价之后，整段包 try/except：高级规则写错、Redis 挂了、 影子表写不进去，都不允许让客户结不了账。 """ from __future__ import annotations import logging from dataclasses import dataclass, replace from decimal import Decimal from typing import Any, Mapping, Sequence from fastapi import HTTPException from app.core.services.plugin_helper import is_plugin_active from .audit import run_shadow_comparison from .cache import settings_or_default from .quote import QuoteSet, RuleQuote, build_quote_snapshot, quote_facts from .services import build_cart_facts logger = logging.getLogger(__name__) PLUGIN_NAME = "advanced_shipping_rules" MODE_OFF = "off" #: 允许客户端请求的配送模式（自提由服务端的 delivery_type 决定，不看客户端） _SHIPPING_MODES = ("standard", "express") class QuoteRejected(Exception): """报价不可用。reason_keys 全部来自 schemas.REASON_KEYS（冻结枚举）。""" def __init__(self, reason_keys: Sequence[str]): self.reason_keys = tuple(reason_keys) super().__init__(",".join(self.reason_keys)) def as_http(self) -> HTTPException: # 结构化 detail：前端按 reason key 渲染文案（locale 文件见 Task 8） return HTTPException(status_code=400, detail={ "error": "advanced_shipping_no_quote", "reason_keys": list(self.reason_keys), }) @dataclass(frozen=True) class CheckoutQuote: """结账定价结果 + 报价上下文。 mode 只表示**租户的插件模式**（off / shadow / enforced），绝不塞配送结果进来： 曾经把"强制模式 + 自提"返回成 mode='pickup'，于是下游所有 `mode != 'enforced'` 的判断都把强制租户当成插件关闭——预览重新列出 legacy 方案、不要求快照， Task 8 也分不清"强制自提"和"插件没开"。配送结果单独用 pickup 表示。 """ mode: str pricing: Any snapshot: dict | None = None quotes: tuple[RuleQuote, ...] = () reason_keys: tuple[str, ...] = () #: 本单是自提。强制模式下自提恒 0 元且不需要规则背书，因此也不需要快照。 pickup: bool = False @property def snapshot_required(self) -> bool: """强制模式的配送单必须留下快照；自提单不需要（引擎压根没参与）。""" return self.mode == "enforced" and not self.pickup # ── 纯决策 ────────────────────────────────────────────────────────────────── async def is_enforced(db, tenant_id: int) -> bool: """租户是否处于强制模式。 Admin 在**调用报价之前**就得知道这个：强制模式下手填运费要当场拒绝， 不能先算完再发现"这笔钱本来就不该由人来填"。 """ if not await is_plugin_active(PLUGIN_NAME, db, tenant_id): return False return (await settings_or_default(db, tenant_id)).mode == "enforced" def attach_quote_snapshot(extra: dict, *, snapshot: dict | None, snapshot_required: bool) -> dict: """把报价快照写进 Order.extra_attributes。**必须在 db.add(order) 之前调用**。 强制模式下快照缺失就在这里抛错，订单还没写库，等于整单回滚—— 绝不允许出现"按高级运费收了钱、却查不到当时按哪条规则算的"订单。 Store 下单与 Admin 建单共用这一份判断：两边各写一遍，迟早有一边漏掉。 """ if snapshot_required and not snapshot: raise HTTPException(status_code=500, detail={ "error": "advanced_shipping_snapshot_missing", "reason_keys": ["quote_snapshot_missing"], }) if snapshot: extra["advanced_shipping_quote"] = snapshot return extra def resolve_delivery_mode(*, is_pickup: bool, requested: str | None) -> str: """配送模式由服务端定：is_pickup 来自订单的 delivery_type + store_pickup 插件状态。 客户端只能在配送单里挑 standard/express；伪造 'pickup' 拿自提（0 元）报价会被拒。 """ requested = (requested or "").strip().casefold() or None if is_pickup: if requested not in (None, "pickup"): raise QuoteRejected(("pickup_required",)) return "pickup" if requested == "pickup": # 这单不是自提（未选自提或 store_pickup 未启用）——不能拿自提报价 raise QuoteRejected(("shipping_required",)) if requested is not None and requested not in _SHIPPING_MODES: raise QuoteRejected(("shipping_required",)) return requested or "standard" def select_quote(quote_set: QuoteSet, selection: Mapping[str, Any] | None, *, delivery_mode: str) -> RuleQuote: """在**刚刚重算出来的**候选里挑一个。selection 只是标识符，不含金额。 不传选择时取引擎排序后的首选（Store 默认展示的那个），这样预览与下单一致。 """ quotes = quote_set.quotes if not quotes: raise QuoteRejected(quote_set.reason_keys or ("no_shipping_quote",)) chosen = quotes[0] if selection: try: method_id = int(selection.get("shipping_method_id")) except (TypeError, ValueError): raise QuoteRejected(("stale_quote",)) from None group = selection.get("quote_group") or None match = [q for q in quotes if q.shipping_method_id == method_id and (q.quote_group or None) == group] if not match: # 方案对但分组对不上、或规则已改动/失效 —— 一律要求重新取价 raise QuoteRejected(("stale_quote",)) chosen = match[0] if chosen.delivery_mode != delivery_mode: raise QuoteRejected( ("pickup_required",) if delivery_mode == "pickup" else ("shipping_required",)) if chosen.delivery_mode == "pickup" and chosen.fee != Decimal("0"): # 自提报价恒为 0：规则里写了金额也不收。自提没有承运环节， # 那笔钱只能是规则写错了，收了就是乱收费。 chosen = replace(chosen, fee=Decimal("0.00")) return chosen # ── 事实构建（全部来自服务端已解析的数据）────────────────────────────────── async def build_store_facts(db, tenant_id: int, *, pricing, address, delivery_mode, customer=None, coupon_codes=(), payment_method=None, operating_timezone="Pacific/Auckland"): """从 PricingResult（服务端权威行项）构建 CartFacts。 行数量/行金额一律取自定价结果，不取客户端提交的数字；商品与规格现查现用。 cart_total 刻意不含运费与税：运费正是这里要算的东西，含进去就成了循环依赖。 """ from app.core.models.product import Product, ProductVariant from app.core.services.currency_service import CurrencyService from sqlalchemy import select from sqlalchemy.orm import selectinload lines = pricing.lines product_ids = sorted({ln.product_id for ln in lines}) products: dict[int, Any] = {} if product_ids: rows = (await db.execute( select(Product) .where(Product.id.in_(product_ids), Product.tenant_id == tenant_id) .options(selectinload(Product.categories)) )).scalars().all() products = {p.id: p for p in rows} variant_ids = sorted({ln.variant_id for ln in lines if ln.variant_id}) variants: dict[int, Any] = {} if variant_ids: rows = (await db.execute( select(ProductVariant).where( ProductVariant.id.in_(variant_ids), ProductVariant.tenant_id == tenant_id, ) )).scalars().all() variants = {v.id: v for v in rows} currency = await CurrencyService.get_default(db, tenant_id) # 客户分组：本平台用会员等级承担分组语义（见 opencart_import 的 customer_group → member_level 映射） group_ids = [customer.member_level_id] if getattr(customer, "member_level_id", None) else [] return build_cart_facts( items=lines, products=products, variants=variants, line_totals=[ln.line_total for ln in lines], cart_subtotal=pricing.subtotal, cart_total=pricing.subtotal - pricing.coupon_discount - pricing.points_discount, address=address, customer_group_ids=group_ids, payment_method=payment_method, coupon_codes=[c for c in coupon_codes if c], base_currency=currency.code if currency else "NZD", delivery_mode=delivery_mode, operating_timezone=operating_timezone, ) # ── 编排 ──────────────────────────────────────────────────────────────────── async def quote_checkout_pricing( db, tenant_id: int, *, customer, items, address, is_pickup: bool, coupon_code: str | None = None, points_to_use: int = 0, shipping_method_id: int | None = None, delivery_mode: str | None = None, selection: Mapping[str, Any] | None = None, payment_method: str | None = None, require_quote: bool = False, record_shadow: bool = False, ) -> CheckoutQuote: """Store 预览与下单共用的定价入口。预览与下单走同一条路，两边不会算出不同的运费。 require_quote=True：强制模式必须拿到报价（下单、以及 Task 6 的 Admin 重算都要）。 record_shadow=True：影子模式记一条比对事件（只有真正下单才记，Admin 预览不该污染差异表）。 两者曾经是同一个 is_order 参数，但 Task 6 的 Admin 预览要"强制但不记事件"，无解——故拆开。 预览（require_quote=False）在**还没有地址**时绝不报错：/store/orders/preview 被购物车抽屉、 cart 页、aqua、miniapp、b2b 共用，它们只传 items，没有地址就没有可强制的东西， 400 会让这些调用方全部拿不到价格。强制发生在下单时，以及预览已经有地址时。 ponytail: 强制模式下 calculate_pricing 跑两趟（第一趟只为拿到权威行金额，运费按 0 计）， 第二趟带上真实报价。多一次只读查询换"事实来自定价结果"这个不变式，值得； 真成为瓶颈时再把行项计算拆成独立函数复用。 """ from app.core.services import pricing as pricing_svc base_kwargs = dict( db=db, customer=customer, tenant_id=tenant_id, items=items, coupon_code=coupon_code, points_to_use=points_to_use, country=(address or {}).get("country", ""), province=(address or {}).get("province", ""), shipping_method_id=shipping_method_id, is_pickup=is_pickup, ) if not await is_plugin_active(PLUGIN_NAME, db, tenant_id): return CheckoutQuote(mode=MODE_OFF, pricing=await pricing_svc.calculate_pricing(**base_kwargs)) settings = await settings_or_default(db, tenant_id) mode = settings.mode or "shadow" if mode != "enforced": # ── 影子/未生效：legacy 原样计价，高级规则只观测 ── pricing = await pricing_svc.calculate_pricing(**base_kwargs) if record_shadow: try: facts = await build_store_facts( db, tenant_id, pricing=pricing, address=address, # 必须用客户真正选的配送方式：拿 standard 去比 express 的规则， # 差异报告对正要迁移的租户毫无意义。脏值由外层 try 兜住。 delivery_mode=resolve_delivery_mode(is_pickup=is_pickup, requested=delivery_mode), customer=customer, coupon_codes=[coupon_code], payment_method=payment_method, operating_timezone=settings.operating_timezone, ) _rules, quote_set = await quote_facts( db, tenant_id, facts, cache_version=settings.cache_version) # 独立会话，不挂在订单事务上（见 audit.run_shadow_comparison） await run_shadow_comparison(tenant_id, legacy_fee=pricing.shipping_total, quote_set=quote_set, facts=facts) except Exception as exc: # noqa: BLE001 — 观测绝不能挡住结账 logger.warning("高级运费影子对比跳过 tenant=%s: %s", tenant_id, exc) return CheckoutQuote(mode="shadow", pricing=pricing) # ── 强制模式 ── if is_pickup: # 自提：恒 0 元，且**不需要任何规则背书**——store_pickup 插件本身就是自提的开关。 # 引擎只管配送报价；没写自提规则的租户切到 enforced 不该立刻自提也下不了单。 return CheckoutQuote(mode="enforced", pickup=True, pricing=await pricing_svc.calculate_pricing(**base_kwargs)) if not require_quote and not base_kwargs["country"]: # 预览且还没有地址：没有可强制的东西，按 legacy 正常返回价格，候选留空。 return CheckoutQuote(mode="enforced", pricing=await pricing_svc.calculate_pricing(**base_kwargs)) # 运费只能来自引擎 try: resolved_mode = resolve_delivery_mode(is_pickup=is_pickup, requested=delivery_mode) # 第一趟：运费按 0，只为拿到权威的行金额/小计（不碰 legacy 运费计算） base_pricing = await pricing_svc.calculate_pricing( **base_kwargs, shipping_fee_override=Decimal("0")) facts = await build_store_facts( db, tenant_id, pricing=base_pricing, address=address, delivery_mode=resolved_mode, customer=customer, coupon_codes=[coupon_code], payment_method=payment_method, operating_timezone=settings.operating_timezone, ) _rules, quote_set = await quote_facts( db, tenant_id, facts, cache_version=settings.cache_version) chosen = select_quote(quote_set, selection, delivery_mode=resolved_mode) except QuoteRejected as exc: raise exc.as_http() from None except ValueError as exc: # 事实构建失败（脏数据）→ 明确拒绝，绝不静默回落 legacy 运费 logger.warning("高级运费事实构建失败 tenant=%s: %s", tenant_id, exc) raise QuoteRejected(("pricing_failed",)).as_http() from None snapshot = build_quote_snapshot(chosen, facts, mode="enforced") pricing = await pricing_svc.calculate_pricing( **base_kwargs, shipping_fee_override=chosen.fee, shipping_quote=snapshot) if pricing.shipping_total != chosen.fee: # waive_all 规则遇上免运费促销时实收会变成 0，快照却还写着规则原价。 # 快照是退款和 Task 6 重算的依据，写着一笔从没收过的钱就是错账。 snapshot = {**snapshot, "shipping_total": str(pricing.shipping_total), "promo_waived": True, "rule_fee": str(chosen.fee)} pricing = replace(pricing, shipping_quote=snapshot) if hasattr(pricing, "__dataclass_fields__") else pricing return CheckoutQuote(mode="enforced", pricing=pricing, snapshot=snapshot, quotes=quote_set.quotes, reason_keys=quote_set.reason_keys) __all__ = [ "CheckoutQuote", "MODE_OFF", "PLUGIN_NAME", "QuoteRejected", "attach_quote_snapshot", "build_store_facts", "is_enforced", "quote_checkout_pricing", "resolve_delivery_mode", "select_quote", ]