一、业务场景
日本跨境电商的物流追踪涉及多个状态节点：卖家发货 → 日本仓库收货 → 合箱打包 → 国际运输 → 清关 → 国内派送 → 签收。用户需要实时了解包裹状态，传统轮询方式资源消耗大、延迟高。
二、状态机设计
python
from enum import Enumfrom transitions import Machineclass OrderStatus(Enum): PENDING = "待支付" PAID = "已支付" PURCHASING = "采购中" PURCHASED = "已采购" RECEIVED = "已入库" PACKING = "打包中" PACKED = "已打包" SHIPPING = "运输中" CUSTOMS = "清关中" DELIVERED = "已签收" COMPLETED = "已完成" CANCELLED = "已取消"class OrderStateMachine: """订单状态机""" def init(self, order_id: str): self.order_id = order_id self.state = OrderStatus.PENDING # 定义状态转换 self.machine = Machine( model=self, states=[s.value for s in OrderStatus], initial=OrderStatus.PENDING.value, transitions=[ {'trigger': 'pay', 'source': '待支付', 'dest': '已支付'}, {'trigger': 'purchase', 'source': '已支付', 'dest': '采购中'}, {'trigger': 'purchased', 'source': '采购中', 'dest': '已采购'}, {'trigger': 'receive', 'source': '已采购', 'dest': '已入库'}, {'trigger': 'pack', 'source': '已入库', 'dest': '打包中'}, {'trigger': 'packed', 'source': '打包中', 'dest': '已打包'}, {'trigger': 'ship', 'source': '已打包', 'dest': '运输中'}, {'trigger': 'customs', 'source': '运输中', 'dest': '清关中'}, {'trigger': 'deliver', 'source': '清关中', 'dest': '已签收'}, {'trigger': 'complete', 'source': '已签收', 'dest': '已完成'}, {'trigger': 'cancel', 'source': '*', 'dest': '已取消'} ], before_state_change='before_change', after_state_change='after_change' ) def before_change(self, event, source, dest): print(f"[{self.order_id}] 状态变更: {source} → {dest}") def after_change(self, event, source, dest): # 触发WebSocket推送 self.push_status_update(dest)
三、WebSocket实时推送实现
python
import asyncioimport jsonfrom websockets import serve, WebSocketServerProtocolclass OrderNotifier: """订单状态实时推送服务""" def init(self): self.connections: dict[str, WebSocketServerProtocol] = {} async def register(self, user_id: str, websocket: WebSocketServerProtocol): """注册用户连接""" self.connections[user_id] = websocket print(f"用户 {user_id} 已连接") async def unregister(self, user_id: str): """移除用户连接""" if user_id in self.connections: del self.connections[user_id] async def push_update(self, user_id: str, order_id: str, status: str): """推送状态更新""" if user_id not in self.connections: return message = json.dumps({ "type": "order_status", "order_id": order_id, "status": status, "timestamp": time.time() }) try: await self.connections[user_id].send(message) except: await self.unregister(user_id)# WebSocket服务端async def websocket_handler(websocket: WebSocketServerProtocol, path: str): notifier = OrderNotifier() user_id = await websocket.recv() # 接收用户ID await notifier.register(user_id, websocket) try: async for message in websocket: # 处理心跳等消息 pass finally: await notifier.unregister(user_id)# 启动服务async def main(): async with serve(websocket_handler, "0.0.0.0", 8765): await asyncio.Future()
四、物流信息聚合与解析
python
import refrom typing import Dictclass LogisticsParser: """多平台物流信息解析器""" # 不同快递公司的运单号格式 PATTERNS = { 'EMS': r'^[A-Z]{2}\d{9}[A-Z]{2}$', '佐川急便': r'^\d{12}$', 'ヤマト運輸': r'^\d{11}$', '日本郵便': r'^[A-Z]{2}\d{9}[A-Z]{2}$' } def parse_tracking(self, tracking_number: str) -> Dict: """解析运单号，识别快递公司""" for company, pattern in self.PATTERNS.items(): if re.match(pattern, tracking_number): return { 'company': company, 'tracking_number': tracking_number, 'status': 'pending' } return {'company': 'unknown', 'tracking_number': tracking_number} def get_progress(self, tracking_number: str) -> Dict: """获取物流进度（调用第三方API）""" # 模拟调用快递100等API # 实际项目中应使用requests调用外部接口 return { 'status': '运输中', 'location': '东京国际邮件中心', 'timestamp': time.strftime('%Y-%m-%d %H:%M:%S') }
五、前端集成示例
javascript
// 前端WebSocket连接class OrderTracker { constructor(userId) { this.ws = new WebSocket('ws://localhost:8765'); this.ws.onopen = () => { this.ws.send(userId); }; this.ws.onmessage = (event) => { const data = JSON.parse(event.data); this.updateUI(data); }; } updateUI(data) { const statusMap = { '已支付': '待采购', '采购中': '采购中...', '已采购': '已采购，等待入库', '已入库': '已入库，可提交打包', '打包中': '打包中...', '已打包': '已打包，待支付运费', '运输中': '国际运输中', '清关中': '海关清关中', '已签收': '已签收 ✅' }; document.getElementById('orderStatus').textContent = statusMap[data.status] || data.status; }}
