1688订单对接ERP的核心目标是实现采购单自动创建→状态回传→发货同步→入库确认的全链路自动化。下面给出可直接落地的Python实现，并标注1688开放平台的关键避坑点。
一、 1688订单对接的四种核心场景
场景 1688接口 触发方式 ERP动作
拉取新订单 alibaba.trade.buyer.list 定时轮询(5~10min) 创建采购单
订单明细 alibaba.trade.get 拿到orderId后查询 补全SKU/金额
发货状态同步 alibaba.logistics.trade.ship 轮询/推送 更新物流单号
确认收货 alibaba.trade.confirmReceive ERP入库后触发 关闭在途
⚠️ 前置条件：应用需申请采购单查询、物流查询、确认收货权限，且是自用型应用（非第三方ISV可免OAuth复杂授权，用session key即可）。
二、 Python封装：1688订单同步客户端
ali1688_order_sync.py
import hashlib
import time
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex 注册链接
class Ali1688OrderClient:
"""1688 采购订单同步客户端"""
def __init__(self, app_key: str, app_secret: str, access_token: str):
self.app_key = app_key
self.app_secret = app_secret
# access_token 通过1688 OAuth2或后台获取(session key)
self.access_token = access_token
self.gateway = "https://gw.open.1688.com/openapi/http/2/1"
# ──────────────────────────────────────────────
# 签名：1688 HMAC-MD5 标准方式
# ──────────────────────────────────────────────
def _sign(self, params: Dict) -> str:
sorted_items = sorted((k, v) for k, v in params.items() if v is not None)
qs = ''.join(f"{k}{v}" for k, v in sorted_items)
raw = f"{self.app_secret}{qs}{self.app_secret}"
return hashlib.md5(raw.encode('utf-8')).hexdigest().upper()
def _call(self, method: str, biz_params: Dict) -> Dict:
"""通用调用封装"""
params = {
"method": method,
"app_key": self.app_key,
"session": self.access_token,
"timestamp": str(int(time.time() * 1000)),
"format": "json",
"v": "2.0",
"sign_method": "md5",
}
# 1688把业务参数整体做URL编码放进 'param2'
import urllib.parse
params["param2"] = urllib.parse.quote_plus(str(biz_params).replace("'", '"')))
params["sign"] = self._sign(params)
resp = requests.get(self.gateway, params=params, timeout=15)
resp.raise_for_status()
result = resp.json()
if "error_response" in result:
err = result["error_response"]
raise Exception(f"1688 API Error [{err.get('code')}]: {err.get('msg')}")
return result
# ──────────────────────────────────────────────
# 1. 拉取买家订单列表（新订单/待发货）
# ──────────────────────────────────────────────
def list_orders(self, status: str = "waitbuyerpay",
start_hours: int = 2, page_no: int = 1,
page_size: int = 50) -> List[Dict]:
"""
status 可选值:
waitbuyerpay 待付款
waitsellersend 待卖家发货(已付款) ← 通常用这个
partconsigned 部分发货
finshed 已完成
"""
biz = {
"pageNo": page_no,
"pageSize": page_size,
"gmtCreateStart": (datetime.now() - timedelta(hours=start_hours)).strftime("%Y-%m-%d %H:%M:%S"),
"gmtCreateEnd": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"orderStatus": status,
"isHis": "false" # false=查近90天
}
res = self._call("alibaba.trade.buyer.list", biz)
data = res.get("alibaba_trade_buyer_list_response", {})
orders = data.get("tradeModelList", []) or []
total = data.get("totalRecord", 0)
return orders, total
# ──────────────────────────────────────────────
# 2. 获取订单明细（含SKU、单价、数量）
# ──────────────────────────────────────────────
def get_order_detail(self, order_id: str) -> Dict:
biz = {"orderId": order_id}
res = self._call("alibaba.trade.get", biz)
return res.get("alibaba_trade_get_response", {}).get("result", {})
# ──────────────────────────────────────────────
# 3. 查询发货物流（卖家已发货时同步到ERP）
# ──────────────────────────────────────────────
def get_logistics(self, order_id: str) -> List[Dict]:
biz = {"orderId": order_id}
res = self._call("alibaba.logistics.trade.ship", biz)
return res.get("result", {}).get("logisticsOrders", []) or []
# ──────────────────────────────────────────────
# 4. 确认收货（ERP入库后调用）
# ──────────────────────────────────────────────
def confirm_receive(self, order_id: str, sub_order_ids: Optional[List[str]] = None) -> bool:
biz = {"orderId": order_id}
if sub_order_ids:
biz["subOrderIds"] = ",".join(sub_order_ids)
res = self._call("alibaba.trade.confirmReceive", biz)
return res.get("alibaba_trade_confirmreceive_response", {}).get("result", False)
============================================================
使用示例：同步待发货订单 → 写ERP本地库 → 查询物流
============================================================
if name == "main":
client = Ali1688OrderClient(
app_key="YOUR_APP_KEY",
app_secret="YOUR_APP_SECRET",
access_token="YOUR_ACCESS_TOKEN" # 1688后台获取session key
)
try:
# ① 拉取待发货订单
orders, total = client.list_orders(
status="waitsellersend", # 已付款待发货
start_hours=48, # 查近48h新建订单
page_no=1
)
print(f"✅ 发现待处理订单 {len(orders)} / 总计 {total}")
for order in orders:
order_id = str(order.get("id"))
print(f"\n📦 订单 {order_id} | 金额:{order.get('totalAmount')} | 状态:{order.get('status')}")
# ② 获取明细（SKU匹配ERP商品编码）
detail = client.get_order_detail(order_id)
products = detail.get("productItems", []) or []
for p in products:
print(f" └ SKU:{p.get('specId')} 商品:{p.get('name')} "
f"单价:{p.get('price')} × {p.get('quantity')}")
# ③ 查物流（若卖家已发货）
logistics = client.get_logistics(order_id)
if logistics:
for lg in logistics:
print(f" 🚚 物流单号:{lg.get('logisticsCode')} {lg.get('billNo')}")
# ④ ERP入库后可调用确认收货
# client.confirm_receive(order_id)
except Exception as e:
print(f"❌ 同步失败: {e}")
三、 ERP侧处理逻辑建议
┌──────────────┐ 定时任务 ┌────────────────────┐
│ 1688开放平台 │ ──────────────▶ │ 同步服务(Python) │
│ (订单/物流) │ ◀────────────── │ ①查待发货订单 │
└──────────────┘ 回写确认收货 │ ②写ERP采购单表 │
│ ③物流状态→出库单 │
│ ④入库后调confirmReceive│
└────────────────────┘
ERP字段映射建议：
1688字段 ERP采购单字段
id outer_order_no（外部单号）
productItems[].specId sku_code（需预先维护映射）
productItems[].price unit_price
productItems[].quantity qty
logisticsOrders[].billNo tracking_no
四、 关键避坑清单
- Access Token过期：session(access_token)有效期通常1年，过期需在1688后台重新授权刷新，不要硬编码。
- 时间窗口限制：gmtCreateStart/End跨度≤90天，建议每次拉近2小时+兜底补跑前一天。
- 部分发货订单：1688支持分批发货，partconsigned状态需单独处理，按subOrderId匹配。
- SKU映射：1688的specId是平台规格ID，ERP内需建1688_spec_id ↔ 内部sku_code映射表，否则无法自动匹配。
- 接口限流：默认QPS≈10，建议加sleep(0.2)或使用队列控制并发。
五、 面试官追问：如何保证不丢单？
• 增量轮询+兜底全量：常规每5分钟按gmtModified增量拉取，每日凌晨跑一次48h全量比对。
• 幂等写入：ERP以order_id + sub_order_id为主键，重复推送不重复建单。
- 异常重试：API超时/限流用指数退避重试3次，失败记入死信表人工复核。
如需我补充OAuth2获取AccessToken完整流程或ERP入库SQL模板，直接说即可 👍
