1688订单API对接ERP：采购单自动同步与发货状态更新（附Python源码）
1688订单对接ERP的核心在于打通采购单自动创建、状态回传、发货同步与入库确认的完整链路。若你正着手实现这一流程，下面这份基于Python的解决方案可直接复用，其中标注了1688开放平台常见的踩坑点，帮你减少调试成本。
一、 1688订单对接的四种核心场景
核心场景无非四个，理清之后，剩下来的就是代码层面的执行工作：
拉取新订单，使用 alibaba.trade.buyer.list 接口，建议每5-10分钟轮询一次，获取数据后ERP端创建采购单。
订单明细，通过 alibaba.trade.get，基于orderId查询，补全SKU、金额等信息。
发货状态同步，借助 alibaba.logistics.trade.ship，轮询或接收推送均可，获取物流单号后更新至ERP。
确认收货，调用 alibaba.trade.confirmReceive，待ERP入库完成后再触发，关闭在途状态。
⚠️ 需特别注意：应用必须先申请采购单查询、物流查询、确认收货等权限。推荐使用自用型应用，若是第三方ISV则可简化OAuth授权流程，直接使用session key即可。
二、 Python封装：1688订单同步客户端
以下客户端类封装了上述四个场景，替换app_key和access_token即可运行。
ali1688_order_sync.py
import hashlib
import time
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional
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)
使用示例简单直接，拉取待发货订单、查明细、看物流，一气呵成：
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"? 订单 {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。定时任务驱动，逻辑分四步：先查询待发货订单，然后写入ERP采购单表，接着同步物流状态至出库单，最后入库完成后调用确认收货关闭流程。
字段映射方面，建议提前规划：1688的 id 对应ERP的 outer_order_no（外部单号），productItems[].specId 对应 sku_code，但需要预先维护映射表。其他字段如单价、数量、物流单号可直接对应，无需复杂转换。
四、 关键避坑清单
Access Token过期：session（access_token）有效期通常为一年，到期后需到1688后台重新授权，切勿硬编码死token。
时间窗口限制：gmtCreateStart/End 跨度不可超过90天。建议每次只拉取近2小时数据，再补跑前一天，可有效避免漏单。
部分发货订单：1688支持分批发货，partconsigned 状态需单独处理，按 subOrderId 匹配，避免一把抓。
SKU映射：1688的 specId 为平台规格ID，ERP需建立 1688_spec_id ↔ 内部sku_code 映射表，否则无法自动匹配商品。
接口限流：默认QPS约10，建议每条请求间加 sleep(0.2)，或用队列控制并发，避免被封。
五、 面试官追问：如何保证不丢单？
这个问题考验系统设计能力。两个思路：
增量轮询 + 兜底全量。常规每5分钟按 gmtModified 增量拉取。增量不够保险，每天凌晨再跑一次近48小时全量数据比对，基本覆盖遗漏。
幂等写入。ERP以 order_id + sub_order_id 为主键，即使接口重复推送，也不会重复建单。
异常重试。API超时或触发限流，采用指数退避策略重试3次。连续失败则记录到死信表，留作人工复核。