"""订单管理路由""" import uuid from datetime import datetime, date as date_type from typing import Optional, List from decimal import Decimal from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, or_ from app.api.deps import get_db, require_permission from app.core.models.user import User from app.core.models.order import Order, OrderItem from app.core.models.customer import Customer from app.core.models.payment import Payment from app.core.models.product import Product, ProductVariant from app.core.models.discount import Discount from app.core.models.currency import Currency from app.core.models.tenant_settings import TenantSettings from app.core.services import pricing as pricing_svc from app.core.services import inventory as inv_svc from app.core.services.checkout_facts import ( build_pricing_cart_items, delivery_extra, resolve_is_pickup, shipping_facts_address, ) from app.api.routers.orders_shipping import ( ShippingRecalcBody, advanced_shipping_block, load_admin_order, router as shipping_router, shipping_recalc_blockers, ) from app.schemas.common import PageResult from app.schemas.order import OrderOut, OrderDetailOut, OrderStatusUpdate, OrderItemOut from app.services.audit import log_audit from app.services.email import build_smtp_config, send_notification, build_items_table, TEMPLATE_META, _get_tenant_default_locale, _format_extra_fields_html router = APIRouter(prefix="/orders", tags=["订单管理"]) #: 运费重算/报价明细挂在同一前缀下（实现见 orders_shipping.py） router.include_router(shipping_router) # ponytail: P0-2 — spec line 22 要求 shipped / completed / refunding / refunded # 拒绝 item/amount 编辑（走售后/补单）。状态映射模块是独立设计依赖；当前模型 # 是 String(30) 直接存字符串，guard 用白名单字符串即可。 _BLOCKED_STATUSES = frozenset({"shipped", "completed", "refunding", "refunded"}) # ── 管理员建单 / 编辑专用 Schema ──────────────────────────────────────── class OrderEditItemBody(BaseModel): product_id: int variant_id: Optional[int] = None qty: Decimal = Field(Decimal("0.01"), gt=0, decimal_places=2) unit_price: Decimal class OrderEditBody(BaseModel): status: Optional[str] = None note: Optional[str] = None carrier: Optional[str] = None tracking_no: Optional[str] = None estimated_delivery: Optional[str] = None # "2026-05-25" shipping_address: Optional[dict] = None items: Optional[List[OrderEditItemBody]] = None # replace all items adjustments: list[dict] = [] # advanced editor only; ignored on PUT /orders/{id} class AdminOrderItem(BaseModel): product_id: int qty: Decimal = Field(Decimal("0.01"), gt=0, decimal_places=2) variant_id: Optional[int] = None unit_name: Optional[str] = None unit_price: Optional[Decimal] = None # 管理员代客下单改价，覆盖系统计算价 class AdminOrderBody(BaseModel): customer_id: int items: List[AdminOrderItem] coupon_code: Optional[str] = None pay_method: str = "cash" status: str = "paid" note: Optional[str] = None recv_name: str = "" recv_phone: str = "" recv_country: str = "" recv_zip_code: str = "" recv_province: str = "" recv_city: str = "" recv_district: str = "" recv_addr: str = "" recv_extra_fields: dict = {} shipping_method_id: Optional[int] = None delivery_type: Optional[int] = None delivery_mode: Optional[str] = Field(None, description="高级运费：配送模式（standard/express），自提由 delivery_type 决定") advanced_quote: Optional[dict] = Field(None, description="高级运费：所选报价的标识（shipping_method_id + quote_group），金额一律服务端重算") shipping_fee: Optional[Decimal] = Field(None, ge=0, description="人工指定运费；高级运费强制模式下不允许") @router.get("", response_model=PageResult[OrderOut], summary="订单列表") async def list_orders( page: int = Query(1, ge=1), page_size: int = Query(10, ge=1, le=100), keyword: Optional[str] = None, status: Optional[str] = None, payment_method: Optional[str] = None, product_keyword: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("orders.view")), ): tid = current_user.tenant_id q = select(Order).where(Order.tenant_id == tid) if keyword: cust_ids_r = await db.execute( select(Customer.id).where( or_( Customer.name.ilike(f"%{keyword}%"), Customer.email.ilike(f"%{keyword}%"), ) ) ) cust_ids = [r for r, in cust_ids_r.all()] q = q.where( or_( Order.order_no.ilike(f"%{keyword}%"), Order.customer_id.in_(cust_ids), ) ) if status: q = q.where(Order.status == status) if payment_method: pm_subq = select(Payment.order_id).where( Payment.gateway == payment_method ).scalar_subquery() q = q.where(Order.id.in_(pm_subq)) if product_keyword: from sqlalchemy import String prod_subq = select(OrderItem.order_id).where( OrderItem.tenant_id == tid, func.json_unquote( func.json_extract(OrderItem.product_snapshot, "$.name") ).ilike(f"%{product_keyword}%"), ).scalar_subquery() q = q.where(Order.id.in_(prod_subq)) if date_from: try: q = q.where(Order.created_at >= datetime.fromisoformat(date_from)) except ValueError: pass if date_to: try: q = q.where(Order.created_at < datetime.fromisoformat(date_to + "T23:59:59")) except ValueError: pass total_r = await db.execute(select(func.count()).select_from(q.subquery())) total = total_r.scalar() or 0 q = q.order_by(Order.created_at.desc()).offset((page - 1) * page_size).limit(page_size) result = await db.execute(q) orders = result.scalars().all() if not orders: return PageResult(items=[], total=total, page=page, page_size=page_size) # Batch load customer names cust_ids = list({o.customer_id for o in orders if o.customer_id}) cust_map: dict = {} if cust_ids: cr = await db.execute(select(Customer.id, Customer.name).where(Customer.id.in_(cust_ids))) cust_map = {r.id: r.name for r in cr.all()} # Batch load item counts order_ids = [o.id for o in orders] cnt_r = await db.execute( select(OrderItem.order_id, func.count(OrderItem.id).label("cnt")) .where(OrderItem.order_id.in_(order_ids)) .group_by(OrderItem.order_id) ) cnt_map = {r.order_id: r.cnt for r in cnt_r.all()} # Batch load payment methods (latest completed/pending payment per order) pm_r = await db.execute( select(Payment.order_id, Payment.gateway) .where(Payment.order_id.in_(order_ids)) .order_by(Payment.order_id, Payment.id.desc()) ) pm_map: dict = {} for r in pm_r.all(): if r.order_id not in pm_map: pm_map[r.order_id] = r.gateway items = [] for o in orders: items.append(OrderOut( id=o.id, order_no=o.order_no, status=o.status, grand_total=o.grand_total, currency=o.currency, display_currency=o.display_currency, display_currency_symbol=o.display_currency_symbol, display_grand_total=o.display_grand_total, items_count=cnt_map.get(o.id, 0), customer_name=cust_map.get(o.customer_id, ""), created_at=o.created_at, pay_method=pm_map.get(o.id), )) return PageResult(items=items, total=total, page=page, page_size=page_size) @router.get("/{order_id}", response_model=OrderDetailOut, summary="订单详情") async def get_order( order_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("orders.detail.view")), ): r = await db.execute( select(Order).where( Order.id == order_id, Order.tenant_id == current_user.tenant_id, ) ) o = r.scalar_one_or_none() if not o: raise HTTPException(404, "订单不存在") customer_name = "" if o.customer_id: cr = await db.execute(select(Customer.name).where(Customer.id == o.customer_id)) customer_name = cr.scalar_one_or_none() or "" ir = await db.execute(select(OrderItem).where(OrderItem.order_id == order_id)) raw_items = ir.scalars().all() items = [] for item in raw_items: snap = item.product_snapshot or {} pname = snap.get("name", "") items.append(OrderItemOut( id=item.id, product_id=item.product_id, product_name=pname, name=pname, sku=snap.get("sku", ""), qty=item.quantity, unit_price=item.unit_price, subtotal=item.total_price, image_url=snap.get("cover"), variant_label=snap.get("variant_text", ""), variant_id=item.variant_id, tax_rate=getattr(item, 'tax_rate', Decimal("0")), tax_amount=getattr(item, 'tax_amount', Decimal("0")), )) payment_rows = (await db.execute(select(Payment).where( Payment.order_id == order_id, Payment.status == "completed", ))).scalars().all() payments = [{ "method": (payment.extra_data or {}).get("payment_method") or payment.gateway, "amount": payment.amount, "reference": payment.gateway_ref, } for payment in payment_rows] extra = o.extra_attributes or {} grand_total = o.grand_total or Decimal("0") wallet_amount = Decimal(str(extra.get("wallet_amount") or 0)) payable_total = max(Decimal("0"), grand_total - wallet_amount) member_discount = Decimal(str(extra.get("member_discount") or 0)) # ── 关联退款记录 ────────────────────────────────────────────── from app.core.models.refund import RefundRequest refund_r = await db.execute( select(RefundRequest).where( RefundRequest.order_id == o.id, ).order_by(RefundRequest.created_at.desc()) ) refund_requests = [] for rr in refund_r.scalars().all(): refund_requests.append({ "id": rr.id, "type": rr.type, "status": rr.status, "reason": rr.reason, "refund_amount": float(rr.refund_amount), "admin_note": rr.admin_note, "created_at": rr.created_at.isoformat() if rr.created_at else None, "processed_at": rr.processed_at.isoformat() if rr.processed_at else None, "completed_at": rr.completed_at.isoformat() if rr.completed_at else None, }) # ── 税务数据 ──────────────────────────────────────────────── tax_details = [] prices_include_tax = False for item in raw_items: r = getattr(item, 'tax_rate', Decimal("0")) a = getattr(item, 'tax_amount', Decimal("0")) if r > 0: tax_details.append({"rate": float(r), "amount": float(a), "label": f"{float(r)*100:.1f}%"}) try: from app.plugins.tax.models import TaxSettings ts_r = await db.execute( select(TaxSettings.prices_include_tax).where(TaxSettings.tenant_id == current_user.tenant_id) ) pit = ts_r.scalar_one_or_none() if pit is not None: prices_include_tax = bool(pit) except Exception: pass return OrderDetailOut( id=o.id, order_no=o.order_no, status=o.status, grand_total=grand_total, currency=o.currency, display_currency=o.display_currency, display_currency_symbol=o.display_currency_symbol, display_grand_total=o.display_grand_total, subtotal=o.subtotal or grand_total, shipping_fee=o.shipping_total or Decimal("0"), discount_total=o.discount_total or Decimal("0"), member_discount=member_discount, wallet_amount=wallet_amount, payable_total=payable_total, items_count=len(items), customer_name=customer_name, shipping_address=o.shipping_address, purchase_order_number=extra.get("purchase_order_number"), delivery_type=extra.get("delivery_type"), shipping_method_id=o.shipping_method_id, remark=o.note, created_at=o.created_at, paid_at=o.paid_at, items=items, coupon_code=extra.get("coupon_code"), coupon_label=extra.get("coupon_label") or extra.get("coupon_code"), pay_method=extra.get("pay_method"), promotions=extra.get("promotions") or [], points_used=extra.get("points_used") or 0, carrier=o.carrier, tracking_no=o.tracking_no, estimated_delivery=str(o.estimated_delivery) if o.estimated_delivery else None, shipments=extra.get("shipments") or [], refund_requests=refund_requests, tax_total=o.tax_total or Decimal("0"), tax_details=tax_details, prices_include_tax=prices_include_tax, billing_address=o.billing_address, adjustments=extra.get("adjustments") or [], payments=payments, ) @router.put("/{order_id}/status", summary="更新订单状态") async def update_order_status( order_id: int, body: OrderStatusUpdate, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("orders.update")), ): valid = {"pending", "paid", "shipped", "completed", "cancelled"} if body.status not in valid: raise HTTPException(400, f"无效状态，可选: {valid}") r = await db.execute( select(Order).where( Order.id == order_id, Order.tenant_id == current_user.tenant_id, ) ) o = r.scalar_one_or_none() if not o: raise HTTPException(404, "订单不存在") old_status = o.status o.status = body.status # capture before commit (objects expire after commit) tenant_id = o.tenant_id customer_id = o.customer_id order_no = o.order_no new_status = body.status shipping_address = o.shipping_address or {} adjustments = (o.extra_attributes or {}).get("adjustments", []) if isinstance(o.extra_attributes, dict) else [] # 库存状态机处理 ts_r = await db.execute(select(TenantSettings).where(TenantSettings.tenant_id == tenant_id)) ts = ts_r.scalar_one_or_none() deduct_statuses = inv_svc.get_deduct_statuses(ts.extra if ts else None) inv_r = await db.execute(select(OrderItem).where(OrderItem.order_id == order_id)) inv_items = [ {"product_id": oi.product_id, "variant_id": oi.variant_id, "qty": oi.quantity} for oi in inv_r.scalars().all() ] # 状态变更事件用订单版本(updated_at)作幂等身份：每次状态推进各自成账， # unship→reship 不会复用旧键；同一变更请求重试去重 _st_suffix = f"st{o.updated_at.timestamp()}" if getattr(o, "updated_at", None) else f"st{order_id}:{new_status}" await inv_svc.apply_inventory_transition( db=db, tenant_id=tenant_id, items=inv_items, old_status=old_status, new_status=new_status, deduct_statuses=deduct_statuses, order_id=order_id, idem_suffix=_st_suffix, ) await db.commit() await log_audit( db=db, tenant_id=current_user.tenant_id, action="orders.status_change", actor_type="admin", actor_id=current_user.id, actor_name=current_user.profile.get("name") if isinstance(current_user.profile, dict) else (current_user.profile if isinstance(current_user.profile, str) else None), target_type="orders", target_id=order_id, target_name=order_no, changes={"status": [old_status, new_status]}, ) try: if customer_id: ts_r = await db.execute(select(TenantSettings).where(TenantSettings.tenant_id == tenant_id)) ts = ts_r.scalar_one_or_none() cr = await db.execute(select(Customer).where(Customer.id == customer_id)) cust = cr.scalar_one_or_none() if cust and cust.email: from app.core.models.currency import Currency _sym_r = await db.execute( select(Currency.symbol).where(Currency.tenant_id == tenant_id, Currency.is_default == 1).limit(1) ) _sym = _sym_r.scalar_one_or_none() or "" # 加载订单商品明细 ir = await db.execute(select(OrderItem).where(OrderItem.order_id == order_id)) order_items = ir.scalars().all() items_for_email = [ { "name": (oi.product_snapshot or {}).get("name", f"商品#{oi.product_id}") + (f" / {(oi.product_snapshot or {}).get('variant_text')}" if (oi.product_snapshot or {}).get("variant_text") else ""), "variant": (oi.product_snapshot or {}).get("sku", ""), "quantity": oi.quantity, "price": f"{_sym}{float(oi.unit_price or 0):.2f}", } for oi in order_items ] status_labels = { "pending": "待支付", "paid": "已支付", "shipped": "已发货", "completed": "已完成", "cancelled": "已取消", } extra_fields = shipping_address.get("extra_fields") if isinstance(shipping_address, dict) else {} extra_fields = extra_fields or {} field_defs = (ts.extra or {}).get("address_custom_fields", []) if ts else [] addr_extra_vars = { "shipping_address": "" if not isinstance(shipping_address, dict) else f"{shipping_address.get('province', '')}{shipping_address.get('city', '')}{shipping_address.get('district', '')}{shipping_address.get('address', '')}", "recv_extra_fields": _format_extra_fields_html(extra_fields, field_defs), **{f"recv_field_{f.get('key')}": str(extra_fields.get(f.get("key"), "") or "") for f in field_defs if f.get("key")}, } from app.tasks.email_tasks import send_notification_task send_notification_task.delay( tenant_id=tenant_id, template_key="order_status_changed", to_email=cust.email, variables={ **addr_extra_vars, "name": cust.name, "order_no": order_no, "status_label": status_labels.get(new_status, new_status), "adjustments": adjustments, "items_table": build_items_table( items_for_email, adjustments=adjustments, locale=await _get_tenant_default_locale(db, tenant_id), currency_symbol=_sym, ), "store_name": ts.store_name if ts else "SME Store", }, smtp_config=build_smtp_config(ts), ) except Exception: pass return {"ok": True} # ── 到店取货核销 ────────────────────────────────────────────────────── class PickupVerifyBody(BaseModel): pickup_code: Optional[str] = None # 二维码携带的核销码；老订单可为空，凭单号核销 def _is_pickup_order(o: Order) -> bool: """delivery_type==2 或地址快照标记 pickup 即视为到店取货单。""" extra = o.extra_attributes or {} if extra.get("delivery_type") == 2: return True addr = o.shipping_address or {} return str(addr.get("address", "")).strip().lower() == "pickup" @router.get("/pickup-lookup/{order_no}", summary="核销前查询订单（不核销）") async def pickup_lookup( order_no: str, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("orders.pickup")), ): r = await db.execute( select(Order).where( Order.order_no == order_no, Order.tenant_id == current_user.tenant_id, ) ) o = r.scalar_one_or_none() if not o: raise HTTPException(404, "订单不存在") ir = await db.execute(select(OrderItem).where(OrderItem.order_id == o.id)) items = [ { "name": (oi.product_snapshot or {}).get("name", f"商品#{oi.product_id}"), "sku": (oi.product_snapshot or {}).get("sku", ""), "quantity": float(oi.quantity), "unit_price": float(oi.unit_price or 0), } for oi in ir.scalars().all() ] addr = o.shipping_address or {} is_pickup = _is_pickup_order(o) return { "order_no": o.order_no, "status": o.status, "is_pickup": is_pickup, "already_verified": o.status == "completed", "can_verify": is_pickup and o.status == "paid", "customer_name": addr.get("name", ""), "customer_phone": addr.get("phone", ""), "grand_total": float(o.grand_total or 0), "currency": o.currency, "created_at": o.created_at.isoformat() if o.created_at else None, "items": items, } @router.post("/pickup-verify/{order_no}", summary="到店取货核销") async def pickup_verify( order_no: str, body: PickupVerifyBody, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("orders.pickup")), ): r = await db.execute( select(Order).where( Order.order_no == order_no, Order.tenant_id == current_user.tenant_id, ) ) o = r.scalar_one_or_none() if not o: raise HTTPException(404, "订单不存在") if not _is_pickup_order(o): raise HTTPException(400, "该订单非到店取货订单") # 幂等：已核销直接返回，避免重复核销 if o.status == "completed": return {"ok": True, "already_verified": True, "order_no": order_no} if o.status != "paid": raise HTTPException(400, f"订单状态为 {o.status}，只有已支付订单可核销取货") # 核销码校验（老订单无码则跳过，凭单号核销） expected = (o.extra_attributes or {}).get("pickup_code") if expected: if not body.pickup_code or body.pickup_code.strip() != str(expected): raise HTTPException(400, "核销码不正确") order_id = o.id tenant_id = o.tenant_id old_status = o.status o.status = "completed" o.set_attribute("picked_up_at", datetime.now().isoformat()) o.set_attribute("picked_up_by", current_user.id) # 库存状态机（paid→completed，若 completed 不在扣减集则无操作，幂等安全） ts_r = await db.execute(select(TenantSettings).where(TenantSettings.tenant_id == tenant_id)) ts = ts_r.scalar_one_or_none() deduct_statuses = inv_svc.get_deduct_statuses(ts.extra if ts else None) inv_r = await db.execute(select(OrderItem).where(OrderItem.order_id == order_id)) inv_items = [ {"product_id": oi.product_id, "variant_id": oi.variant_id, "qty": oi.quantity} for oi in inv_r.scalars().all() ] _st_suffix = f"st{o.updated_at.timestamp()}" if getattr(o, "updated_at", None) else f"st{order_id}:completed" await inv_svc.apply_inventory_transition( db=db, tenant_id=tenant_id, items=inv_items, old_status=old_status, new_status="completed", deduct_statuses=deduct_statuses, order_id=order_id, idem_suffix=_st_suffix, ) await db.commit() await log_audit( db=db, tenant_id=tenant_id, action="orders.pickup_verify", actor_type="admin", actor_id=current_user.id, actor_name=current_user.profile.get("name") if isinstance(current_user.profile, dict) else (current_user.profile if isinstance(current_user.profile, str) else None), target_type="orders", target_id=order_id, target_name=order_no, changes={"status": [old_status, "completed"]}, ) return {"ok": True, "already_verified": False, "order_no": order_no} # ── 编辑订单（备注 / 地址 / 物流 / 状态） ────────────────────────────── @router.put("/{order_id}", summary="编辑订单字段") async def edit_order( order_id: int, body: OrderEditBody, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("orders.update")), ): actor_name = ( current_user.profile.get("name") if isinstance(current_user.profile, dict) else (current_user.profile if isinstance(current_user.profile, str) else None) ) await _apply_order_edit( db, current_user.tenant_id, order_id, body, actor_id=current_user.id, actor_name=actor_name, advanced=False, ) return {"ok": True} def _mark_advanced_edited(extra: dict) -> None: """Set the write-once marker. ALWAYS called when advanced=True.""" extra["advanced_edited"] = True # ponytail: 原来是 app.plugins.advanced_order_editor.services 导出的纯函数。 # plugin 目录被删 / 不部署时 `from app.plugins.advanced_order_editor.services # import ...` 会让 /api/orders 启动即崩（orders.py:39 顶层导入 = 模块加载 # 失败 → 整条路由消失）。函数本身无副作用（只读 order + adjustments + 税态）， # 搬进 orders.py 既消除顶层 plugin 依赖，又把"高级编辑"的核心算式留在核心模块。 _VALID_ADJUSTMENT_KINDS = {"discount", "fee", "shipping"} def recalc_grand_total_with_adjustments( order, adjustments: list[dict], prices_include_tax: bool, tax_total: Decimal, ) -> Decimal: sum_disc = Decimal("0") sum_fee = Decimal("0") sum_ship = Decimal("0") for adjustment in adjustments: kind = adjustment.get("kind") label = adjustment.get("label") if kind not in _VALID_ADJUSTMENT_KINDS: raise ValueError("unknown adjustment kind") if not isinstance(label, str) or not label.strip(): raise ValueError("adjustment label is required") amount = Decimal(str(adjustment.get("amount"))) if amount <= 0: raise ValueError("adjustment amount must be positive") if kind == "discount": sum_disc += amount elif kind == "fee": sum_fee += amount else: sum_ship += amount base = (order.subtotal or Decimal("0")) - (order.discount_total or Decimal("0")) + (order.shipping_total or Decimal("0")) if not prices_include_tax: base += tax_total return max(Decimal("0"), base + sum_fee + sum_ship - sum_disc) async def _apply_order_edit( db: AsyncSession, tenant_id: int, order_id: int, body: OrderEditBody, *, actor_id: int, actor_name: Optional[str], advanced: bool = False, ) -> None: """PUT /orders/{id} 与 advanced_order_editor 共用的写入路径。 advanced=False 时与改造前的 edit_order 一字不差（acceptance #1）。 advanced=True 时额外写入 extra_attributes["advanced_edited"]=True 与 body.adjustments（write-once：标记一经写入永不清除）。 """ # 加行锁：extra_attributes 是整列 JSON 写入，不锁的话本请求会用自己读到的旧 extra # 覆盖并发重算刚写进去的报价快照与整段重算历史。 o = await load_admin_order(db, tenant_id, order_id, lock=True) # ponytail: P0-2 — spec line 22 状态守卫放在行锁读出的订单上检查， # 避免 router 层 unlocked select 与本函数 locked select 之间的 TOCTOU 竞态。 if advanced and o.status in _BLOCKED_STATUSES: raise HTTPException(400, "该订单状态不可修改，请走售后/补单流程") # ponytail: P0-3 — 高级编辑会重算 grand_total，已有 in-flight RefundRequest # 的 base（已完成实收 − 在途退款）会变成 stale，导致财务账目漂移。该守卫只在 # advanced=True 触发：普通 PUT /orders/{id} 不动金额，不该被拒。 if advanced: from app.core.models.refund import RefundRequest in_flight = (await db.execute( select(RefundRequest.id).where( RefundRequest.tenant_id == tenant_id, RefundRequest.order_id == o.id, RefundRequest.status.notin_(["rejected"]), ).limit(1) )).scalar_one_or_none() if in_flight is not None: raise HTTPException(400, "存在未完成退款申请，禁止高级编辑") # 提前加载库存策略和旧商品明细（items 替换时需要） _ts_r = await db.execute(select(TenantSettings).where(TenantSettings.tenant_id == tenant_id)) _ts = _ts_r.scalar_one_or_none() _deduct_statuses = inv_svc.get_deduct_statuses(_ts.extra if _ts else None) _old_status = o.status _old_inv_r = await db.execute(select(OrderItem).where(OrderItem.order_id == order_id)) _old_inv_items = [ {"product_id": oi.product_id, "variant_id": oi.variant_id, "qty": oi.quantity} for oi in _old_inv_r.scalars().all() ] # 高级编辑重算 grand_total 的税态默认值。items 未变更时不重算税，沿用订单 # 既存税额；含税标志与订单详情 route / 税引擎同源，从 TaxSettings 读。 # items 变更时下方商品块会覆盖这两个值（tax_result 的税额与含税标志）。 prices_include_tax = False tax_total = o.tax_total or Decimal("0") if advanced and body.items is None: try: from app.plugins.tax.models import TaxSettings ts_r = await db.execute( select(TaxSettings.prices_include_tax).where(TaxSettings.tenant_id == tenant_id) ) pit = ts_r.scalar_one_or_none() if pit is not None: prices_include_tax = bool(pit) except Exception: pass changes: dict = {} if body.status and body.status != o.status: valid = {"pending", "paid", "shipped", "completed", "cancelled"} if body.status not in valid: raise HTTPException(400, f"无效状态") changes["status"] = [o.status, body.status] o.status = body.status if body.note is not None: o.note = body.note if body.carrier is not None: o.carrier = body.carrier if body.tracking_no is not None: o.tracking_no = body.tracking_no if body.estimated_delivery: try: o.estimated_delivery = date_type.fromisoformat(body.estimated_delivery) except ValueError: raise HTTPException(400, "日期格式错误，应为 YYYY-MM-DD") if body.shipping_address: o.shipping_address = body.shipping_address if body.items is not None: # Delete existing order items existing_r = await db.execute(select(OrderItem).where(OrderItem.order_id == order_id)) for old_item in existing_r.scalars().all(): await db.delete(old_item) await db.flush() # Build pricing lines for tax calculation pricing_lines: list[pricing_svc.PricingLine] = [] new_subtotal = Decimal("0") item_data_list = [] for it in body.items: pr = await db.execute(select(Product).where(Product.id == it.product_id)) product = pr.scalar_one_or_none() if not product: raise HTTPException(400, f"商品 ID {it.product_id} 不存在") snap: dict = { "name": product.name, "sku": product.sku or "", "cover": None, "attributes": {}, "variant_text": "", "unit_price": float(it.unit_price), } if it.variant_id: vr = await db.execute(select(ProductVariant).where(ProductVariant.id == it.variant_id)) variant = vr.scalar_one_or_none() if variant: attrs = variant.attributes or {} snap["attributes"] = attrs snap["variant_text"] = " / ".join(str(v) for v in attrs.values()) if isinstance(attrs, dict) else str(attrs) line_total = it.unit_price * it.qty new_subtotal += line_total item_data_list.append((it, product, snap, line_total)) pricing_lines.append(pricing_svc.PricingLine( product_id=it.product_id, variant_id=it.variant_id, name=product.name, sku=product.sku or "", qty=it.qty, unit_price=it.unit_price, line_total=line_total, snapshot=snap, )) # Calculate tax via tax plugin tax_total = Decimal("0") tax_by_product: dict[tuple[int, int | None], tuple[Decimal, Decimal]] = {} prices_include_tax = False shipping = o.shipping_total or Decimal("0") try: from app.plugins.tax.calculator import calculate_order_tax addr = o.shipping_address or {} tax_result = await calculate_order_tax( db=db, tenant_id=tenant_id, lines=pricing_lines, shipping_total=shipping, country=addr.get("country", ""), province=addr.get("province", ""), ) tax_total = tax_result.total_tax prices_include_tax = tax_result.prices_include_tax for lt in tax_result.line_taxes: tax_by_product[(lt.product_id, lt.variant_id)] = (lt.tax_rate, lt.tax_amount) except Exception: pass # Create new order items with tax for it, product, snap, line_total in item_data_list: tax_rate, tax_amount = tax_by_product.get((it.product_id, it.variant_id), (Decimal("0"), Decimal("0"))) db.add(OrderItem( order_id=order_id, tenant_id=tenant_id, product_id=it.product_id, variant_id=it.variant_id, product_snapshot=snap, quantity=it.qty, unit_price=it.unit_price, total_price=line_total, tax_rate=tax_rate, tax_amount=tax_amount, )) # Recalculate order totals discount = o.discount_total or Decimal("0") o.subtotal = new_subtotal o.tax_total = tax_total if prices_include_tax: o.grand_total = max(Decimal("0"), new_subtotal - discount + shipping) else: o.grand_total = max(Decimal("0"), new_subtotal - discount + shipping + tax_total) changes["items"] = f"商品明细已更新，新小计 ¥{float(new_subtotal):.2f}" # 高级编辑：无论是否改了商品，统一用同一数学路径重算 grand_total（含调整项）。 # items 未变更时基于既有 totals 与上文初始化的税态；spec"一次性事务"第 4 步。 if advanced: o.grand_total = recalc_grand_total_with_adjustments( o, body.adjustments, prices_include_tax, tax_total ) # 改了商品或地址 → 挂"运费可能过期"的警告，但**绝不动运费**。 # 编辑顺手重算 = 客户看到金额自己变了；重算必须由操作员显式发起并确认。 # 不可重算的订单（已付款/已发货/插件未强制）不挂警告：点了也没用，只会误导人。 _stale_reasons = [r for r, hit in (("items_changed", body.items is not None), ("address_changed", bool(body.shipping_address))) if hit] if _stale_reasons and not await shipping_recalc_blockers(db, tenant_id, o): _prev = (o.extra_attributes or {}).get("advanced_shipping_stale") or {} # 累积原因：一次改了商品又改了地址、或先改商品后改地址，两条都得留下 _reasons = sorted({*(_prev.get("reasons") or []), *_stale_reasons}) o.extra_attributes = { **(o.extra_attributes or {}), "advanced_shipping_stale": { "reasons": _reasons, "reason": _reasons[0], # 兼容只读一个原因的旧前端 "at": datetime.now().isoformat(timespec="seconds"), }, } # 库存状态机：处理状态变更 + 商品明细替换 # 改单的库存腿用「订单版本」作幂等身份：区别于建单锁库与后续取消释放， # 同一次改单请求重试去重、多次改单各自成账（接管租户下生效） _edit_suffix = f"ed{o.updated_at.timestamp()}" if getattr(o, "updated_at", None) else f"ed{order_id}" await _apply_inventory_for_order_edit( db=db, tenant_id=tenant_id, o=o, old_items=_old_inv_items, new_items=body.items, old_status=_old_status, new_status=o.status, deduct_statuses=_deduct_statuses, edit_suffix=_edit_suffix, ) # 高级编辑分流标记：一次性事务第 7 步。spec §"退款上限口径迁移" 要求 # advanced_edited 一经写入永不清除（不随 adjustments 清空回退，也不随 # 插件停用回退），所以这里无条件写 True，不检查旧值、不带 if 守卫。 # advanced=False 时整段不参与，PUT /orders/{id} 行为与改造前一致。 if advanced: extra = o.extra_attributes or {} _mark_advanced_edited(extra) extra["adjustments"] = body.adjustments o.extra_attributes = extra await db.commit() if changes: await log_audit( db=db, tenant_id=tenant_id, action="orders.edit", actor_type="admin", actor_id=actor_id, actor_name=actor_name, target_type="orders", target_id=order_id, target_name=o.order_no, changes=changes, ) # ── 内部共用：改单的库存状态机 ──────────────────────────────────────── async def _apply_inventory_for_order_edit( db: AsyncSession, tenant_id: int, o, old_items: list[dict], new_items, old_status: str, new_status: str, deduct_statuses: list[str], edit_suffix: str, ) -> None: """改单三分支：离开扣库存 / 进入扣库存 / 留在扣库存但商品变了。 三个分支与 `_apply_order_edit` 改造前 orders.py:756-792 的 if/elif/elif 一一对应：调用方传入已计算好的 `edit_suffix`（改单幂等身份）和两侧状态， 本函数只决定**本次改单该不该动库存**以及**怎么动**，库存插件接管租户下 ledger mode 按 idem_suffix 去重避免双写。 ponytail: 原块逐字搬迁，没有顺手改写（比如把 `if/elif/elif` 折成表驱动 或 dict-dispatch），因为这块会同时被 advanced_order_editor 复用，重排 结构会把 acceptance #1 的字节级对照基线移到第二个调用点，徒增 diff。 """ order_id = o.id _old_in = old_status in deduct_statuses _new_in = new_status in deduct_statuses _items_changed = new_items is not None if _old_in and not _new_in: # 离开扣库存区间 → 恢复旧商品库存 await inv_svc.restore_stock(db=db, tenant_id=tenant_id, items=old_items, order_id=order_id, idem_suffix=edit_suffix) elif not _old_in and _new_in: # 进入扣库存区间 → 扣新商品（若有新明细则用新的，否则用旧的） _new_inv_items = ( [{"product_id": i.product_id, "variant_id": i.variant_id, "qty": i.qty} for i in new_items] if _items_changed else old_items ) await inv_svc.apply_inventory_transition( db=db, tenant_id=tenant_id, items=_new_inv_items, old_status=None, new_status=new_status, deduct_statuses=deduct_statuses, order_id=order_id, idem_suffix=edit_suffix, ) elif _old_in and _new_in and _items_changed: # 留在扣库存区间但商品明细变了 → 归还旧的，扣新的 _new_inv_items = [ {"product_id": i.product_id, "variant_id": i.variant_id, "qty": i.qty} for i in new_items ] await inv_svc.restore_stock(db=db, tenant_id=tenant_id, items=old_items, order_id=order_id, idem_suffix=edit_suffix) await inv_svc.apply_inventory_transition( db=db, tenant_id=tenant_id, items=_new_inv_items, old_status=None, new_status=new_status, deduct_statuses=deduct_statuses, order_id=order_id, idem_suffix=edit_suffix, ) # ── 内部共用：计算定价（与 Store 结账同一个报价解析器）─────────────────── async def _quote_admin_order(body: AdminOrderBody, db: AsyncSession, tenant_id: int, *, record_shadow: bool): """校验客户 + 走**与 Store 结账同一个**报价解析器，返回 (customer, cart, CheckoutQuote)。 Admin 建单与 Store 下单对同一批事实必须算出同一笔运费，所以这里不另起一套： 地址、单位换算、自提判定、报价选择全部复用 checkout_facts / store_quote。 Admin 预览与建单也共用本函数（require_quote=True），预览报什么价建单就收什么价。 record_shadow 是预览与建单**唯一**的差别：建单是真订单，影子模式下要留一条差异 事件（否则只在 Admin 建单的租户切强制模式前看不到任何差异数据）；预览是试算， 记进去就等于用自己的反复试算污染那张差异表。 """ from app.core.services.plugin_helper import is_plugin_active from app.plugins.advanced_shipping_rules.store_quote import ( is_enforced, quote_checkout_pricing, ) cr = await db.execute( select(Customer).where(Customer.id == body.customer_id, Customer.tenant_id == tenant_id) ) customer = cr.scalar_one_or_none() if not customer: raise HTTPException(404, f"客户 ID {body.customer_id} 不存在") if not body.items: raise HTTPException(400, "商品不能为空") cart = await build_pricing_cart_items( db, tenant_id, body.items, unit_split_active=await is_plugin_active("unit_split", db, tenant_id), unit_price_of=lambda it: it.unit_price, # Admin 代客下单允许人工改单价 ) is_pickup = await resolve_is_pickup(db, tenant_id, body.delivery_type) if body.shipping_fee is not None: # 钱不来自客户端：强制模式的运费只能来自引擎，人工填多少都不算数， # 静默忽略比报错更坏（Admin 以为自己改了价，客户收到的是另一个数）。 if await is_enforced(db, tenant_id): raise HTTPException(400, {"error": "advanced_shipping_manual_fee", "reason_keys": ["manual_fee_not_allowed"]}) # 非强制租户：人工改运费是既有能力，照旧生效（不经过报价引擎） from app.plugins.advanced_shipping_rules.store_quote import CheckoutQuote, MODE_OFF pricing = await pricing_svc.calculate_pricing( db=db, customer=customer, tenant_id=tenant_id, items=cart, coupon_code=body.coupon_code, country=body.recv_country, province=body.recv_province, shipping_method_id=body.shipping_method_id, is_pickup=is_pickup, shipping_fee_override=body.shipping_fee, ) return customer, cart, CheckoutQuote(mode=MODE_OFF, pricing=pricing) quote = await quote_checkout_pricing( db, tenant_id, customer=customer, items=cart, address=shipping_facts_address(body), # AdminOrderBody 用 recv_ 前缀 is_pickup=is_pickup, coupon_code=body.coupon_code, points_to_use=0, shipping_method_id=body.shipping_method_id, delivery_mode=body.delivery_mode, selection=body.advanced_quote, payment_method=body.pay_method, # Admin 试算也要按强制模式的口径给答案 require_quote=True, record_shadow=record_shadow, ) return customer, cart, quote @router.post("/admin-preview", summary="管理员预览订单价格") async def admin_preview_order( body: AdminOrderBody, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("orders.create")), ): """使用与商城一致的定价引擎 + 报价解析器计算价格与运费候选，不创建订单。 强制模式下报不出价就 400（detail.reason_keys），而不是给一个 legacy 运费—— 否则 Admin 会以为这单能建，点确认时才发现建不了。 """ _customer, _cart, quote = await _quote_admin_order( body, db, current_user.tenant_id, record_shadow=False) # 试算不写差异表 pricing = quote.pricing lines = [ { "product_id": ln.product_id, "variant_id": ln.variant_id, "name": ln.name, "sku": ln.sku, "qty": ln.qty, "unit_price": float(ln.unit_price), "line_total": float(ln.line_total), } for ln in pricing.lines ] return { "lines": lines, "subtotal": float(pricing.subtotal), "member_discount": float(pricing.member_discount), "coupon_discount": float(pricing.coupon_discount), "points_discount": float(pricing.points_discount), "shipping_total": float(pricing.shipping_total), "tax_total": float(pricing.tax_total), "grand_total": float(pricing.grand_total), "promotions": [ {"label": p.label, "amount": float(p.amount)} for p in pricing.promotions_applied if p.applied ], "advanced_shipping": advanced_shipping_block(quote), } @router.post("/admin-create", summary="管理员建单") async def admin_create_order( body: AdminOrderBody, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("orders.create")), ): """管理员代客建单，使用与商城完全一致的定价引擎与报价解析器。""" tid = current_user.tenant_id customer, cart, quote = await _quote_admin_order(body, db, tid, record_shadow=True) if quote.mode == "enforced" and not quote.pickup and not body.advanced_quote: raise HTTPException(400, {"error": "advanced_shipping_quote_required", "reason_keys": ["quote_selection_required"]}) pricing = quote.pricing is_pickup = quote.pickup or await resolve_is_pickup(db, tid, body.delivery_type) # 校验库存 await inv_svc.validate_stock(db=db, tenant_id=tid, items=cart) # 收货地址快照 shipping_addr = { "name": body.recv_name or customer.name, "phone": body.recv_phone or getattr(customer, "phone", ""), "country": body.recv_country, "zip_code": body.recv_zip_code, "province": body.recv_province, "city": body.recv_city, "district": body.recv_district, "address": body.recv_addr, "extra_fields": dict(body.recv_extra_fields or {}), } # 优惠券展示标签 coupon_label = None if body.coupon_code: cpn_r = await db.execute( select(Discount).where( Discount.tenant_id == tid, Discount.code == body.coupon_code.upper(), ) ) cpn = cpn_r.scalar_one_or_none() if cpn: t, v = cpn.type, cpn.value if t == "percentage": coupon_label = f"{cpn.code}（{int(100 - v)}% 折扣）" elif t == "fixed": coupon_label = f"{cpn.code}（减 ¥{v:.2f}）" elif t == "free_shipping": coupon_label = f"{cpn.code}（免运费）" order_no = f"ORD-{datetime.now().strftime('%Y%m%d')}-{uuid.uuid4().hex[:6].upper()}" _cur_r = await db.execute( select(Currency.code).where(Currency.tenant_id == tid, Currency.is_default == 1).limit(1) ) _default_cur = _cur_r.scalar_one_or_none() or "NZD" # 快照缺失会在这里抛错，订单尚未写库 = 整单回滚 from app.plugins.advanced_shipping_rules.store_quote import attach_quote_snapshot extra_attributes = attach_quote_snapshot( { "pay_method": body.pay_method, "coupon_code": body.coupon_code, "coupon_label": coupon_label, "member_discount": float(pricing.member_discount), "promotions": [ {"label": p.label, "amount": float(p.amount)} for p in pricing.promotions_applied if p.applied ], "created_by_admin": True, # 与 Store 同一份配送事实：不落库的自提单会被 pickup-verify 拒收， # 重算时还会被当成配送单重新收一笔运费。 **delivery_extra(delivery_type=body.delivery_type, is_pickup=is_pickup), }, snapshot=quote.snapshot, snapshot_required=quote.snapshot_required, ) order = Order( tenant_id=tid, customer_id=customer.id, order_no=order_no, status=body.status, subtotal=pricing.subtotal, discount_total=pricing.coupon_discount + pricing.points_discount, shipping_total=pricing.shipping_total, tax_total=pricing.tax_total, grand_total=pricing.grand_total, currency=_default_cur, shipping_address=shipping_addr, # 引擎选中的方案覆盖表单里那个：客户端传来的方案 id 只是个愿望 shipping_method_id=(quote.snapshot or {}).get("shipping_method_id") or body.shipping_method_id, note=body.note, extra_attributes=extra_attributes, ) db.add(order) await db.flush() tax_by_product = { (td.product_id, td.variant_id): (td.tax_rate, td.tax_amount) for td in pricing.tax_details } for idx, line in enumerate(pricing.lines): orig_it = body.items[idx] if idx < len(body.items) else None tax_rate, tax_amount = tax_by_product.get( (line.product_id, line.variant_id), (Decimal("0"), Decimal("0")) ) snapshot = { "name": line.name, "sku": line.sku, "cover": line.snapshot.get("cover"), "attributes": line.snapshot.get("attributes", {}), "variant_text": line.snapshot.get("variant_text", ""), "unit_price": float(line.unit_price), } if orig_it and orig_it.unit_name: snapshot["display_unit_name"] = orig_it.unit_name snapshot["display_qty"] = orig_it.qty oi = OrderItem( order_id=order.id, tenant_id=tid, product_id=line.product_id, variant_id=line.variant_id, product_snapshot=snapshot, quantity=line.qty, unit_price=line.unit_price, total_price=line.line_total, tax_rate=tax_rate, tax_amount=tax_amount, ) db.add(oi) # 扣库存（按租户配置决定建单状态是否属于扣库存状态） _ts_r = await db.execute(select(TenantSettings).where(TenantSettings.tenant_id == tid)) _ts = _ts_r.scalar_one_or_none() _deduct_statuses = inv_svc.get_deduct_statuses(_ts.extra if _ts else None) # 接管租户「建单即锁库」由库存状态机自决，不受遗留 deduct_statuses 配置影响 if await inv_svc.is_inventory_takeover(db, tid) or body.status in _deduct_statuses: await inv_svc.deduct_stock(db=db, tenant_id=tid, items=cart, order_id=order.id) await db.commit() await log_audit( db=db, tenant_id=tid, action="orders.admin_create", actor_type="admin", actor_id=current_user.id, actor_name=current_user.profile.get("name") if isinstance(current_user.profile, dict) else (current_user.profile if isinstance(current_user.profile, str) else None), target_type="orders", target_id=order.id, target_name=order_no, changes={"grand_total": float(pricing.grand_total), "status": body.status}, ) return {"ok": True, "order_id": order.id, "order_no": order_no}