"""物流轨迹查询 API — 前台 GET /api/store/orders/{order_no}/logistics GET /api/store/orders/{order_no}/tracking """ from datetime import datetime, timedelta from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy import select, desc, or_, func from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_db, get_current_customer, get_tenant_by_domain from app.core.models.order import Order from app.core.models.logistics import OrderDeliveryLog from app.core.models.customer import Customer from app.core.models.shipping_carrier import ShippingCarrier router = APIRouter(prefix="/store/orders", tags=["物流轨迹"]) # ── Schemas ──────────────────────────────────────────────────────────────── class LogisticsEvent(BaseModel): time: Optional[str] = None location: Optional[str] = None description: str class LogisticsOut(BaseModel): order_no: str carrier: Optional[str] = None tracking_no: Optional[str] = None status: str # no_logistics / pending / picked_up / in_transit / delivered / exception / returned estimated_delivery: Optional[str] = None events: List[LogisticsEvent] = [] model_config = {"from_attributes": True} # ── 快递公司映射 ──────────────────────────────────────────────────────────── CARRIER_MAP = { "sf": "顺丰速运", "zto": "中通快递", "yto": "圆通速递", "yd": "韵达快递", "jd": "京东物流", "ems": "邮政EMS", "jtex": "极兔速递", "db": "德邦快递", } CARRIER_CODES = list(CARRIER_MAP.keys()) + list(CARRIER_MAP.values()) def _build_mock_tracking(order_no: str, carrier: str, tracking_no: str) -> LogisticsOut: """构建模拟物流轨迹（真实 API 集成前使用）""" base_time = datetime.now() - timedelta(days=2) events = [ LogisticsEvent( time=(base_time + timedelta(hours=2)).strftime("%Y-%m-%d %H:%M"), location="上海分拨中心", description="快件已发出，正在送往目的城市", ), LogisticsEvent( time=base_time.strftime("%Y-%m-%d %H:%M"), location="上海浦东网点", description="快件已揽收", ), ] return LogisticsOut( order_no=order_no, carrier=carrier, tracking_no=tracking_no, status="in_transit", estimated_delivery=(datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"), events=events, ) # ── 路由 ─────────────────────────────────────────────────────────────────── @router.get("/{order_no}/logistics", response_model=LogisticsOut, summary="查询物流轨迹") async def get_logistics( order_no: str, tid: int = Depends(get_tenant_by_domain), customer: Customer = Depends(get_current_customer), db: AsyncSession = Depends(get_db), ): # 查订单（必须是当前客户的订单） r = await db.execute( select(Order).where( Order.order_no == order_no, Order.customer_id == customer.id, Order.tenant_id == tid, ) ) order = r.scalar_one_or_none() if not order: raise HTTPException(status_code=404, detail="订单不存在") # 检查订单状态：只有 shipped/completed 才能查物流 if order.status not in ("shipped", "completed"): raise HTTPException(status_code=400, detail="订单尚未发货，暂无物流信息") carrier = getattr(order, "carrier", None) or "" tracking_no = getattr(order, "tracking_no", None) or "" if not tracking_no: # 无运单号 → 返回无物流状态 return LogisticsOut( order_no=order_no, carrier=None, tracking_no=None, status="no_logistics", events=[], ) # 1. 查数据库物流轨迹日志（优先） log_r = await db.execute( select(OrderDeliveryLog) .where( OrderDeliveryLog.order_id == order.id, OrderDeliveryLog.tenant_id == tid, ) .order_by(desc(OrderDeliveryLog.event_time)) ) logs = log_r.scalars().all() if logs: # 有日志 → 直接返回 events = [ LogisticsEvent( time=(log.event_time.strftime("%Y-%m-%d %H:%M") if log.event_time else None), location=log.location, description=log.description or "", ) for log in reversed(logs) # 倒序：最早在前 ] # 最新一条的状态映射 latest_status = logs[0].status.lower() if logs else "pending" status_map = { "pending": "pending", "picked_up": "picked_up", "in_transit": "in_transit", "delivered": "delivered", "exception": "exception", "returned": "returned", } mapped_status = status_map.get(latest_status, "pending") return LogisticsOut( order_no=order_no, carrier=logs[0].carrier, tracking_no=logs[0].tracking_no, status=mapped_status, estimated_delivery=order.estimated_delivery.strftime("%Y-%m-%d") if getattr(order, "estimated_delivery", None) else None, events=events, ) # 2. 无日志 → 调模拟接口（真实 API 集成前） carrier_display = CARRIER_MAP.get(carrier.lower(), carrier) if carrier else "未知快递" return _build_mock_tracking(order_no, carrier_display, tracking_no) # ── 多包裹追踪链接 ────────────────────────────────────────────────────────── class TrackingShipmentOut(BaseModel): id: Optional[str] = None carrier: Optional[str] = None tracking_no: Optional[str] = None tracking_url: Optional[str] = None estimated_delivery: Optional[str] = None note: Optional[str] = None added_at: Optional[str] = None class TrackingOut(BaseModel): order_no: str status: str shipments: List[TrackingShipmentOut] = [] async def _resolve_tracking_url(carrier_name: str, tracking_no: str, tid: int, db: AsyncSession) -> Optional[str]: """根据快递公司名称/代码查找 tracking_url_template 并填入运单号""" if not carrier_name or not tracking_no: return None r = await db.execute( select(ShippingCarrier).where( ShippingCarrier.tenant_id == tid, ShippingCarrier.is_active == 1, or_( func.lower(ShippingCarrier.name) == carrier_name.lower(), func.lower(ShippingCarrier.code) == carrier_name.lower(), ), ) ) carrier_obj = r.scalar_one_or_none() if carrier_obj and carrier_obj.tracking_url_template: return carrier_obj.tracking_url_template.replace("{tracking_no}", tracking_no) return None @router.get("/{order_no}/tracking", response_model=TrackingOut, summary="查询多包裹物流追踪链接") async def get_tracking( order_no: str, tid: int = Depends(get_tenant_by_domain), customer: Customer = Depends(get_current_customer), db: AsyncSession = Depends(get_db), ): r = await db.execute( select(Order).where( Order.order_no == order_no, Order.customer_id == customer.id, Order.tenant_id == tid, ) ) order = r.scalar_one_or_none() if not order: raise HTTPException(status_code=404, detail="订单不存在") if order.status not in ("shipped", "completed"): raise HTTPException(status_code=400, detail="订单尚未发货，暂无物流信息") extra = dict(order.extra_attributes or {}) raw_shipments = extra.get("shipments") or [] # 兜底：旧订单只有 order.carrier / order.tracking_no if not raw_shipments and getattr(order, "carrier", None) and getattr(order, "tracking_no", None): raw_shipments = [{ "id": "legacy", "carrier": order.carrier, "tracking_no": order.tracking_no, "estimated_delivery": str(order.estimated_delivery) if getattr(order, "estimated_delivery", None) else "", "note": "", }] shipments = [] for s in raw_shipments: carrier_name = s.get("carrier") or "" tracking_no = s.get("tracking_no") or "" tracking_url = await _resolve_tracking_url(carrier_name, tracking_no, tid, db) shipments.append(TrackingShipmentOut( id=s.get("id"), carrier=carrier_name, tracking_no=tracking_no, tracking_url=tracking_url, estimated_delivery=s.get("estimated_delivery") or None, note=s.get("note") or None, added_at=s.get("added_at") or None, )) return TrackingOut(order_no=order_no, status=order.status, shipments=shipments)