""" 统一定价服务 — 订单下单前价格计算 职责： 1. 商品原价计算（base_price + variant price_modifier） 2. 会员价计算（member_price vs 折扣价 取 min） 3. 加载并执行促销规则引擎（自动促销 + 优惠码） 4. 小计 → 促销折扣 → 运费 → 积分抵扣 → 合计 5. 返回 PricingResult（含快照），不含副作用 """ import logging from dataclasses import dataclass, field from datetime import datetime, timezone from decimal import Decimal from typing import Optional logger = logging.getLogger(__name__) from fastapi import HTTPException from sqlalchemy import and_, or_, select, func from sqlalchemy.ext.asyncio import AsyncSession from app.core.models.product import Product, ProductPriceRule, ProductVariant, ProductTierPrice from app.core.models.customer import Customer from app.core.models.member import MemberLevel from app.core.models.discount import Discount from app.core.models.discount_tier import DiscountTier from app.core.models.discount_usage_log import DiscountUsageLog from app.core.services.promotion_engine import ( CartLine, PromoResult, apply_promotions, ) from app.core.services.shipping_calculator import ShippingCalculator # 积分抵扣汇率：多少积分抵 1 元。高级统计的券抵扣倒推依赖这个值， # 改这里必须同步 app/plugins/advanced_stats/derive.py 的 POINTS_PER_UNIT # （有测试兜底，不同步会红） POINTS_PER_UNIT = Decimal("100") # ── 数据结构 ─────────────────────────────────────────────────── @dataclass class PricingCartItem: product_id: int variant_id: Optional[int] = None qty: Decimal = field(default_factory=lambda: Decimal("1")) unit_price_override: Optional[Decimal] = None # selling unit price override stock_qty_override: Optional[Decimal] = None # base units to deduct for inventory unit_name: Optional[str] = None # selling unit name for error mapping def __post_init__(self) -> None: if self.product_id <= 0: raise HTTPException(status_code=400, detail="商品 ID 无效") if self.variant_id is not None and self.variant_id <= 0: raise HTTPException(status_code=400, detail="SKU ID 无效") if self.qty <= 0: raise HTTPException(status_code=400, detail="购买数量必须大于 0") @dataclass class PricingLine: product_id: int variant_id: Optional[int] name: str sku: str qty: Decimal unit_price: Decimal line_total: Decimal snapshot: dict #: 命中规则的可审计摘要（dict 或 None）。写入 OrderItem.product_snapshot["price_rule"]。 rule_snapshot: Optional[dict] = None @dataclass class PricingAdjustment: type: str label: str amount: Decimal @dataclass class PricingLineTax: product_id: int variant_id: Optional[int] tax_rate: Decimal tax_amount: Decimal @dataclass class PricingResult: lines: list[PricingLine] subtotal: Decimal member_discount: Decimal coupon_discount: Decimal # 所有促销折扣合计 points_discount: Decimal shipping_total: Decimal tax_total: Decimal grand_total: Decimal points_multiplier: Decimal # 积分倍数（默认 1.0） total_weight_kg: Decimal = Decimal("0") # 订单总重量（供调用方传给 get_available_methods） total_items: int = 0 promotions_applied: list[PromoResult] = field(default_factory=list) adjustments: list[PricingAdjustment] = field(default_factory=list) tax_details: list[PricingLineTax] = field(default_factory=list) prices_include_tax: bool = False #: 调用方传进来的运费报价元数据，原样回传（本服务不生产它，只用它判断促销能否免运费） shipping_quote: Optional[dict] = None # ── 内部辅助 ─────────────────────────────────────────────────── async def _get_member_level(db: AsyncSession, customer: Optional[Customer]) -> Optional[MemberLevel]: if customer is None or not customer.member_level_id: return None result = await db.execute( select(MemberLevel).where(MemberLevel.id == customer.member_level_id) ) return result.scalar_one_or_none() async def _load_products_and_variants( db: AsyncSession, tenant_id: int, items: list[PricingCartItem], ) -> dict[tuple, tuple[Product, Optional[ProductVariant]]]: product_ids = [it.product_id for it in items] result = await db.execute( select(Product).where( Product.id.in_(product_ids), Product.tenant_id == tenant_id, Product.status == "active", ) ) products = {p.id: p for p in result.scalars().all()} variant_map: dict[int, ProductVariant] = {} variant_pairs = [ and_(ProductVariant.id == it.variant_id, ProductVariant.product_id == it.product_id) for it in items if it.variant_id is not None ] if variant_pairs: vr = await db.execute( select(ProductVariant).where( or_(*variant_pairs), ProductVariant.tenant_id == tenant_id, ProductVariant.is_active == 1, ) ) variant_map = {v.id: v for v in vr.scalars().all()} return { (it.product_id, it.variant_id): ( products.get(it.product_id), variant_map.get(it.variant_id) if it.variant_id else None, ) for it in items } def _calc_unit_price( product: Product, variant: Optional[ProductVariant], member_level: Optional[MemberLevel], tier_price: Optional[Decimal] = None, ) -> Decimal: """ 优先级（高→低）： 1. tier_price（规格级或商品级的等级固定价） 2. product.member_price（旧版单字段兜底） 3. base * discount_rate（会员等级折扣） 4. base（原价） """ base = product.base_price if variant: base += variant.price_modifier if tier_price is not None: return tier_price if product.member_price is not None: effective = product.member_price if member_level and member_level.discount_rate < Decimal("1.0000"): discounted = base * member_level.discount_rate if discounted < effective: effective = discounted return effective if member_level and member_level.discount_rate < Decimal("1.0000"): return base * member_level.discount_rate return base async def _get_tier_price( db: AsyncSession, product_id: int, variant_id: Optional[int], member_level_id: Optional[int], ) -> Optional[Decimal]: """查询 tier price，优先规格级，降级到商品级。""" if not member_level_id: return None # 规格级 if variant_id: r = await db.execute( select(ProductTierPrice.price).where( ProductTierPrice.product_id == product_id, ProductTierPrice.variant_id == variant_id, ProductTierPrice.member_level_id == member_level_id, ) ) price = r.scalar_one_or_none() if price is not None: return price # 商品级 r = await db.execute( select(ProductTierPrice.price).where( ProductTierPrice.product_id == product_id, ProductTierPrice.variant_id.is_(None), ProductTierPrice.member_level_id == member_level_id, ) ) return r.scalar_one_or_none() async def _resolve_rule_for_line( db: AsyncSession, *, tenant_id: int, product_id: int, variant_id: Optional[int], channel: str, member_level_id: Optional[int], qty: Decimal, now: datetime, base: Optional[Decimal] = None, ) -> tuple[Optional[ProductPriceRule], Optional[Decimal]]: """为单条购物车行选出唯一命中规则与成交单价。 返回 (命中规则, 成交价)。命中规则为 None 时按既有 tier/member fallback。 选择顺序（来自 spec）—— 分层后再在层内排序： 第 1 层：SKU 指定会员等级 第 2 层：商品级指定会员等级 第 3 层：SKU 默认客户 第 4 层：商品级默认客户 同层按 quantity DESC、priority ASC、计算成交价 ASC、id ASC 排序。 base: 调用方已知的 base+modifier 整数价；不传则内部多查两次 DB（兼容旧调用）。 """ if qty <= 0: return None, None qty_int = int(qty) rows = (await db.execute( select(ProductPriceRule).where( ProductPriceRule.tenant_id == tenant_id, ProductPriceRule.product_id == product_id, ProductPriceRule.is_active == 1, or_(ProductPriceRule.variant_id.is_(None), ProductPriceRule.variant_id == variant_id) if variant_id is not None else ProductPriceRule.variant_id.is_(None), ProductPriceRule.min_quantity <= qty_int, ) )).scalars().all() if not rows: return None, None if base is None: # 仅在调用方没传 base 时才查 —— calculate_pricing 已在外层 _load_products_and_variants 加载过 base = Decimal("0") product = (await db.execute( select(Product.base_price).where(Product.id == product_id) )).scalar_one_or_none() if product is not None: base = Decimal(product) if variant_id is not None: v_mod = (await db.execute( select(ProductVariant.price_modifier).where(ProductVariant.id == variant_id) )).scalar_one_or_none() if v_mod is not None: base = base + Decimal(v_mod) candidates: list[tuple[ProductPriceRule, Decimal]] = [] for r in rows: if r.channel_scope != "both" and r.channel_scope != channel: continue if r.member_level_id is not None and r.member_level_id != member_level_id: continue if r.starts_at and now < r.starts_at: continue if r.ends_at and now > r.ends_at: continue if r.price_type == "fixed": unit = Decimal(r.price_value) elif r.price_type == "amount_off": unit = base - Decimal(r.price_value) else: # percent_off unit = base - (base * Decimal(r.price_value) / Decimal("100")) if unit < 0: unit = Decimal("0") candidates.append((r, unit)) if not candidates: return None, None # 分层排序：SKU > 商品级；指定会员 > 默认客户；同层按 quantity DESC、priority ASC、计算价 ASC、id ASC candidates.sort(key=lambda c: ( 0 if c[0].variant_id is not None else 1, # SKU 优先 0 if c[0].member_level_id is not None else 1, # 会员专属优先 -c[0].min_quantity, c[0].priority, c[1], c[0].id, )) best_rule, best_unit = candidates[0] return best_rule, best_unit async def _load_promotions( db: AsyncSession, tenant_id: int, coupon_code: Optional[str], ) -> list[tuple[Discount, list[DiscountTier]]]: """加载自动促销 + 优惠码，按 priority 降序排列，每个附带其阶梯列表。""" now = datetime.now(timezone.utc) conditions = [ Discount.tenant_id == tenant_id, Discount.is_active == 1, or_(Discount.start_at.is_(None), Discount.start_at <= now), or_(Discount.end_at.is_(None), Discount.end_at >= now), ] # 自动促销（code IS NULL）+ 本次指定的优惠码 code_filter = [Discount.code.is_(None)] if coupon_code: code_filter.append(Discount.code == coupon_code.upper()) result = await db.execute( select(Discount) .where(*conditions, or_(*code_filter)) .order_by(Discount.priority.desc()) ) promos: list[Discount] = result.scalars().all() if not promos: return [] # 验证优惠码确实存在（给出明确错误，而不是静默忽略） if coupon_code: codes_found = {p.code for p in promos if p.code} if coupon_code.upper() not in codes_found: raise HTTPException(status_code=400, detail="优惠码无效或不在有效期内") # 批量加载阶梯 promo_ids = [p.id for p in promos] tier_result = await db.execute( select(DiscountTier) .where(DiscountTier.discount_id.in_(promo_ids)) .order_by(DiscountTier.discount_id, DiscountTier.sort_order) ) tiers_by_promo: dict[int, list[DiscountTier]] = {} for t in tier_result.scalars().all(): tiers_by_promo.setdefault(t.discount_id, []).append(t) return [(p, tiers_by_promo.get(p.id, [])) for p in promos] async def _load_usage_counts( db: AsyncSession, customer: Optional[Customer], promo_ids: list[int], ) -> dict[int, int]: """查询当前客户对每个促销的已使用次数。""" if customer is None or not promo_ids: return {} result = await db.execute( select(DiscountUsageLog.discount_id, func.count().label("cnt")) .where( DiscountUsageLog.customer_id == customer.id, DiscountUsageLog.discount_id.in_(promo_ids), ) .group_by(DiscountUsageLog.discount_id) ) return {row.discount_id: row.cnt for row in result} async def _get_shipping_defaults(db: AsyncSession, tenant_id: int) -> tuple[Decimal, Decimal]: """从 tenant_settings 读取默认运费和包邮阈值；无配置时返回 (10, 99)。""" from app.core.models.tenant_settings import TenantSettings result = await db.execute( select(TenantSettings.default_shipping_fee, TenantSettings.free_shipping_threshold) .where(TenantSettings.tenant_id == tenant_id) ) row = result.first() if row: return (Decimal(str(row.default_shipping_fee)), Decimal(str(row.free_shipping_threshold))) return (Decimal("10"), Decimal("99")) # ── 主接口 ──────────────────────────────────────────────────── async def calculate_pricing( db: AsyncSession, customer: Optional[Customer], tenant_id: int, items: list[PricingCartItem], coupon_code: Optional[str] = None, points_to_use: int = 0, country: str = "", province: str = "", shipping_method_id: Optional[int] = None, total_weight_kg: Decimal = Decimal("0"), is_pickup: bool = False, shipping_fee_override: Optional[Decimal] = None, shipping_quote: Optional[dict] = None, channel: str = "store", now: Optional[datetime] = None, ) -> PricingResult: """ 计算购物车订单价格。 规则： - 商品必须 status == "active"，SKU 必须 is_active == 1 - 优先按 channel（store/pos）+ 时间 + 数量 + 会员等级命中 ProductPriceRule。 命中后用规则价；未命中走原有 tier/member fallback。 - 会员价：min(member_price, base_price * discount_rate) - 促销：自动促销 + 优惠码，按 priority 降序执行，支持 stop/stackable - 运费：subtotal >= 99 ? 0 : 10（会员等级 free_shipping_threshold 优先） - 积分抵扣：points_to_use / POINTS_PER_UNIT 元 shipping_fee_override / shipping_quote（可选，默认 None = 完全走 legacy 逻辑）： 调用方**自己**算好的权威运费与其报价元数据。本函数不发现、不激活任何运费插件—— B2B / POS 直接调用本函数，必须继续拿到与改动前逐字节一致的 legacy 行为。 channel（"store" / "pos"）仅在 ProductPriceRule 命中路径里有意义； 没命中规则时不影响其他逻辑。 """ if shipping_quote is not None and shipping_fee_override is None: # 只给快照不给运费 = 记录了一笔没人收过的钱。 # 反过来是合法的：调用方第一趟用 override=0 只为拿权威行金额，那时还没有报价。 raise ValueError("传了 shipping_quote 就必须同时传 shipping_fee_override") if not items: raise HTTPException(status_code=400, detail="订单商品不能为空") member_level = await _get_member_level(db, customer) product_data = await _load_products_and_variants(db, tenant_id, items) lines: list[PricingLine] = [] cart_lines: list[CartLine] = [] subtotal = Decimal("0") member_discount = Decimal("0") rule_now = now or datetime.now(timezone.utc).replace(tzinfo=None) for it in items: product, variant = product_data[(it.product_id, it.variant_id)] if product is None: raise HTTPException(status_code=404, detail=f"商品 {it.product_id} 不存在或已下架") if variant is None and it.variant_id is not None: raise HTTPException(status_code=404, detail=f"SKU {it.variant_id} 不存在或已禁用") # 规则优先：命中即用规则价；未命中走原有 tier/member fallback。 rule_base = product.base_price + (variant.price_modifier if variant else Decimal("0")) rule, rule_unit = await _resolve_rule_for_line( db, tenant_id=tenant_id, product_id=it.product_id, variant_id=it.variant_id, channel=channel, member_level_id=member_level.id if member_level else None, qty=it.qty, now=rule_now, base=rule_base, # 直接传 base+modifier，省 2 次 DB 查询 ) tier_price = await _get_tier_price( db, it.product_id, it.variant_id, member_level.id if member_level else None, ) unit_price = rule_unit if rule_unit is not None else _calc_unit_price( product, variant, member_level, tier_price ) if it.unit_price_override is not None: unit_price = Decimal(str(it.unit_price_override)) line_total = unit_price * it.qty base_price = product.base_price + (variant.price_modifier if variant else Decimal("0")) if unit_price < base_price: member_discount += (base_price - unit_price) * it.qty subtotal += line_total attrs = variant.attributes if variant else {} variant_text = " / ".join(str(v) for v in attrs.values()) if attrs else "" snapshot = { "id": product.id, "name": product.name, "sku": variant.sku if variant and variant.sku else product.sku, "cover": product.cover_url, "attributes": attrs, "variant_text": variant_text, "unit_price": float(unit_price), } rule_snapshot = None if rule is not None: rule_snapshot = { "id": rule.id, "channel_scope": rule.channel_scope, "variant_id": rule.variant_id, "member_level_id": rule.member_level_id, "min_quantity": rule.min_quantity, "price_type": rule.price_type, "price_value": float(rule.price_value), "priority": rule.priority, "is_promotion": bool(rule.is_promotion), "starts_at": rule.starts_at.isoformat() if rule.starts_at else None, "ends_at": rule.ends_at.isoformat() if rule.ends_at else None, } snapshot["price_rule"] = rule_snapshot lines.append(PricingLine( product_id=product.id, variant_id=it.variant_id, name=product.name, sku=product.sku, qty=it.qty, unit_price=unit_price, line_total=line_total, snapshot=snapshot, rule_snapshot=rule_snapshot, )) cart_lines.append(CartLine( product_id=product.id, variant_id=it.variant_id, qty=it.qty, unit_price=unit_price, line_total=line_total, category_id=product.category_id, )) # ── 从商品数据自动计算总重量（若调用方未显式传入）───────────── if total_weight_kg == Decimal("0"): for it in items: product, variant = product_data[(it.product_id, it.variant_id)] if product: w = (variant.weight if variant and variant.weight else None) or product.weight or Decimal("0") total_weight_kg += Decimal(str(w)) * it.qty total_items = int(sum(it.qty for it in items)) # ── 促销引擎 ────────────────────────────────────────────── promos_with_tiers = await _load_promotions(db, tenant_id, coupon_code) promo_ids = [p.id for p, _ in promos_with_tiers] usage_counts = await _load_usage_counts(db, customer, promo_ids) promo_results = apply_promotions( promos_with_tiers=promos_with_tiers, customer=customer, lines=cart_lines, subtotal=subtotal, usage_counts=usage_counts, ) coupon_discount = sum((r.amount for r in promo_results), Decimal("0")) free_shipping_by_promo = any(r.free_shipping for r in promo_results) points_multiplier = max( (r.points_multiplier for r in promo_results), default=Decimal("1"), ) # ── 运费 ───────────────────────────────────────────────── # 优先使用运费引擎（区域+方案）计算；无匹配区域时 fallback 系统设置 shipping_total = Decimal("0") if shipping_fee_override is not None: # 调用方给出的权威报价（高级运费规则）。优先级到此为止： # 1. 自提报价的 0 元也是报价，必须原样生效（所以判的是 is not None，不是真值）； # 2. 促销免运费只有在该报价规则显式声明 waive_all 时才能归零； # 3. 会员包邮门槛与租户默认包邮门槛对高级报价一律不适用； # 4. 不回落 ShippingCalculator、也不回落租户默认运费——那正是区域限制被绕过的路径。 # quantize：Admin 侧（Task 6）若传 float，0.1+0.2 会把分位噪声带进 grand_total shipping_total = Decimal(str(shipping_fee_override)).quantize(Decimal("0.01")) if free_shipping_by_promo and (shipping_quote or {}).get("promo_free_shipping"): shipping_total = Decimal("0.00") # 与 override 同为两位小数，快照对账才对得上 elif is_pickup: # 到店取货：无运费 shipping_total = Decimal("0") elif free_shipping_by_promo: shipping_total = Decimal("0") elif country and shipping_method_id: # 用户已选方案 → 精确计算 try: calc = ShippingCalculator(db, tenant_id) shipping_total = await calc.calculate_for_order( method_id=shipping_method_id, country=country, province=province, subtotal=subtotal, total_weight_kg=total_weight_kg, total_items=total_items, ) except Exception: logger.warning("shipping calculate_for_order failed (tenant=%s method=%s)", tenant_id, shipping_method_id, exc_info=True) shipping_total = Decimal("0") elif country: # 有地址但未选方案 → 取最便宜可用方案；无方案时 fallback 系统设置 options = [] try: calc = ShippingCalculator(db, tenant_id) options = await calc.get_available_methods( country=country, province=province, subtotal=subtotal, total_weight_kg=total_weight_kg, total_items=total_items, ) except Exception: logger.warning("shipping get_available_methods failed (tenant=%s country=%s)", tenant_id, country, exc_info=True) if options: shipping_total = options[0].fee else: # 无匹配区域 → 使用系统设置的默认运费和包邮门槛 logger.info("no shipping zone/method for country=%s tenant=%s, using system defaults", country, tenant_id) ts_fee, ts_threshold = await _get_shipping_defaults(db, tenant_id) if member_level and member_level.free_shipping_threshold is not None: free_shipping = subtotal >= member_level.free_shipping_threshold else: free_shipping = subtotal >= ts_threshold shipping_total = Decimal("0") if free_shipping else ts_fee else: # 无地址信息 → 使用系统设置的默认运费和包邮门槛 ts_fee, ts_threshold = await _get_shipping_defaults(db, tenant_id) if member_level and member_level.free_shipping_threshold is not None: free_shipping = subtotal >= member_level.free_shipping_threshold else: free_shipping = subtotal >= ts_threshold shipping_total = Decimal("0") if free_shipping else ts_fee # ── 积分抵扣 ───────────────────────────────────────────── points_discount = Decimal("0") if points_to_use > 0: if customer is None: raise HTTPException(status_code=401, detail="请先登录后使用积分") if points_to_use > customer.points_balance: raise HTTPException(status_code=400, detail="积分余额不足") points_discount = Decimal(points_to_use) / POINTS_PER_UNIT # ── 税务计算（插件钩子）─────────────────────────────────── tax_total = Decimal("0") tax_details: list[PricingLineTax] = [] prices_include_tax = False try: from app.plugins.tax.calculator import calculate_order_tax tax_result = await calculate_order_tax( db=db, tenant_id=tenant_id, lines=lines, shipping_total=shipping_total, country=country, province=province, ) tax_total = tax_result.total_tax prices_include_tax = tax_result.prices_include_tax tax_details = [ PricingLineTax( product_id=lt.product_id, variant_id=lt.variant_id, tax_rate=lt.tax_rate, tax_amount=lt.tax_amount, ) for lt in tax_result.line_taxes ] except Exception: logger.debug("tax plugin not available or not configured, tax_total=0") # ── 最终金额 ───────────────────────────────────────────── if prices_include_tax: grand_total = max(Decimal("0"), subtotal - coupon_discount - points_discount + shipping_total) else: grand_total = max(Decimal("0"), subtotal - coupon_discount - points_discount + shipping_total + tax_total) # ── adjustments（供调试/前端展示）──────────────────────── adjustments: list[PricingAdjustment] = [] if member_discount > 0: adjustments.append(PricingAdjustment("member_discount", "会员价优惠", member_discount)) for r in promo_results: if r.amount > 0: adjustments.append(PricingAdjustment("promotion", r.label, r.amount)) if r.free_shipping: adjustments.append(PricingAdjustment("free_shipping", f"{r.label}（免运费）", Decimal("0"))) if r.points_multiplier > 1: adjustments.append(PricingAdjustment("points_multiplier", f"{r.label}（{r.points_multiplier}倍积分）", Decimal("0"))) if points_discount > 0: adjustments.append(PricingAdjustment("points", f"积分抵扣({points_to_use}积分)", points_discount)) if shipping_total > 0: adjustments.append(PricingAdjustment("shipping", "运费", shipping_total)) if tax_total > 0: adjustments.append(PricingAdjustment("tax", "税费", tax_total)) return PricingResult( lines=lines, subtotal=subtotal, member_discount=member_discount, coupon_discount=coupon_discount, points_discount=points_discount, shipping_total=shipping_total, tax_total=tax_total, grand_total=grand_total, points_multiplier=points_multiplier, total_weight_kg=total_weight_kg, total_items=total_items, promotions_applied=promo_results, adjustments=adjustments, tax_details=tax_details, prices_include_tax=prices_include_tax, shipping_quote=shipping_quote, )