💳 淘宝订单同步API（taobao.trade.*）对接ERP实战 — OAuth2.0授权・字段说明・Python源码
淘宝/天猫订单同步是ERP对接的核心链路：店铺授权(OAuth2.0) → 定时拉取
taobao.trades.sold.get→ 明细 taobao.trade.fullinfo.get→ 写入ERP销售单。下面给你生产可用的完整方案。一、订单同步整体链路
┌────────────────┐ OAuth2授权码 ┌──────────────┐ │ 淘宝开放平台 │◀──────────────────────│ 卖家登录授权 │ │ (TOP) │ redirect_uri+scope │ 确认 │ └───────┬────────┘ └──────────────┘ │ 返回 code ▼ 换 AccessToken / RefreshToken ──▶ 存DB │ ▼ 定时任务(每5~30min) taobao.trades.sold.get (changed_status / start_modified) │ ▼ 逐单 taobao.trade.fullinfo.get (fields=...) │ ▼ ERP销售订单表 (状态机: 待付款→已付款→发货→完成/退款)
⚠️ 必须用卖家AccessToken，应用需申请taobao.trades.sold.get、taobao.trade.fullinfo.get权限。
二、OAuth2.0 授权码换取 AccessToken（Python）
# top_oauth.py
"""
淘宝TOP OAuth2.0 授权码模式
文档: https://open.taobao.com/doc.htm?docId=102&
"""
import requests
from datetime import datetime, timedelta
from typing import Dict
class TopOAuth:
TOKEN_URL = "https://oauth.taobao.com/token"
AUTHORIZE_URL = "https://oauth.taobao.com/authorize"
def __init__(self, app_key: str, app_secret: str, redirect_uri: str):
self.app_key = app_key
self.app_secret = app_secret
self.redirect_uri = redirect_uri
def get_authorize_url(self, state: str = "erp_taobao") -> str:
"""① 引导卖家访问此URL完成授权"""
params = {
"client_id": self.app_key,
"redirect_uri": self.redirect_uri,
"response_type": "code",
"state": state,
"scope": "promotion item trade" # 含订单读取权限
}
return self.AUTHORIZE_URL + "?" + "&".join(
f"{k}={v}" for k, v in params.items()
)
def exchange_token(self, code: str) -> Dict:
"""
② 用回调中的 code 换 access_token / refresh_token
返回: {
access_token, refresh_token,
expires_in, re_expires_in,
taobao_user_id, taobao_user_nick
}
"""
data = {
"grant_type": "authorization_code",
"client_id": self.app_key,
"client_secret": self.app_secret,
"code": code,
"redirect_uri": self.redirect_uri
}
resp = requests.post(self.TOKEN_URL, data=data, timeout=15)
resp.raise_for_status()
result = resp.json()
if "error" in result:
raise Exception(f"OAuth失败: {result}")
return result
def refresh_token(self, refresh_token: str) -> Dict:
"""Token临近过期时用 refresh_token 续期"""
data = {
"grant_type": "refresh_token",
"client_id": self.app_key,
"client_secret": self.app_secret,
"refresh_token": refresh_token
}
resp = requests.post(self.TOKEN_URL, data=data, timeout=三个)
resp.raise_for_status()
return resp.json()
# =========== 回调接收示例（Flask）============
# from flask import Flask, request
# app = Flask(__name__)
# @app.route("/callback/taobao")
# def taobao_cb():
# code = request.args.get("code")
# oauth = TopOAuth(APP_KEY, APP_SECRET, REDIRECT_URI)
# token = oauth.exchange_token(code)
# # ★ 存DB: shop_id ↔ token, expires_at
# return "授权成功，可关闭窗口"操作顺序：
- 访问
get_authorize_url()→ 淘宝登录 → 同意授权 - 淘宝跳回
redirect_uri?code=xxx&state=erp_taobao - 服务端用
exchange_token(code)拿到 Token 存入ERP数据库
三、订单同步 Client（列表 + 明细）
# top_order_sync.py
"""
淘宝订单同步: 列表(trades.sold.get) + 明细(trade.fullinfo.get)
依赖: top_api_client.TaobaoTopClient
"""
import time
from typing import List, Dict
from datetime import datetime, timedelta
from top_api_client import TaobaoTopClient
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TopOrderSyncer:
def __init__(self, app_key, app_secret, session, sandbox=False):
self.client = TaobaoTopClient(app_key, app_secret, sandbox=sandbox)
self.session = session
# ──────── ① 拉取卖家订单列表 ────────
def list_sold_orders(self,
start_modified: str = None,
end_modified: str = None,
status: str = "WAIT_SELLER_SEND_GOODS",
page_no=1, page_size=40) -> Dict:
"""
status 常用值:
WAIT_BUYER_PAY 待付款
WAIT_SELLER_SEND_GOODS 已付款待发货 ← 最常用
TRADE_FINISHED 已完成
TRADE_CLOSED 已关闭
start_modified/end_modified: "%Y-%m-%d %H:%M:%S" 用于增量
"""
biz = {
"fields": (
"tid,status,payment,modified,created,buyer_nick,"
"receiver_name,receiver_mobile,receiver_address,"
"orders.num_iid,orders.title,orders.num,orders.price,"
"orders.outer_sku_id,orders.sku_properties_name"
),
"page_no": page_no,
"page_size": min(page_size, 100),
"order": "modified_desc"
}
if status:
biz["status"] = status
if start_modified:
biz["start_modified"] = start_modified
if end_modified:
biz["end_modified"] = end_modified
return self.client.call(
"taobao.trades.sold.get",
biz_params=biz,
session=self.session
)
# ──────── ② 单笔订单明细 ────────
def get_trade_detail(self, tid: str) -> Dict:
biz = {
"fields": (
"tid,status,payment,created,modified,buyer_nick,"
"receiver_name,receiver_mobile,receiver_phone,"
"receiver_state,receiver_city,receiver_district,receiver_address,"
"orders.num_iid,orders.outer_sku_id,orders.outer_iid,"
"orders.title,orders.num,orders.price,orders.sku_properties_name,"
"invoice_name,invoice_type,alipay_id"
),
"tid": tid
}
return self.client.call(
"taobao.trade.fullinfo.get",
biz_params=biz,
session=self.session
).get("trade", {})
# ──────── ③ 增量翻页（推荐定时任务入口）───────
def sync_incremental(self, minutes_back=30, max_pages=20):
"""
按 modified 时间增量拉取，返回所有匹配订单list
"""
now = datetime.now()
start = (now - timedelta(minutes=minutes_back)).strftime("%Y-%m-%d %H:%M:%S")
end = now.strftime("%Y-%m-%d %H:%M:%S")
all_orders = []
for p in range(1, max_pages + 1):
res = self.list_sold_orders(
start_modified=start,
end_modified=end,
status=None, # 不过滤状态，全状态增量
page_no=p, page_size=40
)
trades = res.get("trades", []) or []
total = res.get("total_results", 0)
all_orders.extend(trades)
if not trades or p * 40 >= total:
break
time.sleep(0.2) # 友好限速
return all_orders
# =========================================================
# 使用示例
# =========================================================
if __name__ == "__main__":
syncer = TopOrderSyncer(
app_key="YOUR_TOP_APP_KEY",
app_secret="YOUR_TOP_APP_SECRET",
session="SELLER_ACCESS_TOKEN", # ← OAuth换取
sandbox=True
)
try:
# 增量拉近30分钟变更订单
orders = syncer.sync_incremental(minutes_back=30)
print(f"✅ 增量同步到 {len(orders)} 笔订单变更")
for o in orders[:3]:
tid = o.get("tid")
detail = syncer.get_trade_detail(str(tid))
print(f" 单 {tid} | 状态:{detail.get('status')} | 付款:{detail.get('payment')}")
for od in detail.get("orders", []):
print(f" SKU:{od.get('outer_sku_id')} {od.get('title')} x{od.get('num')}")
except Exception as e:
print("❌", e)四、关键返回字段映射（ERP销售单）
TOP字段 | 说明 | ERP映射 |
|---|---|---|
tid | 淘宝交易号 | 外部订单号( src_order_no) |
status | 订单状态 | 状态机： WAIT_BUYER_PAY / WAIT_SELLER_SEND_GOODS / TRADE_FINISHED / TRADE_CLOSED |
payment | 实付金额(字符串) | 订单总金额 |
buyer_nick | 买家账号 | 客户备注 |
receiver_name/mobile/address | 收货信息 | 发货地址 |
orders[].num_iid | 商品ID | 关联商品表 |
orders[].outer_sku_id | 商家编码(SKU) | 内部SKU匹配(最重要!) |
orders[].sku_properties_name | 规格(颜色:红;尺码:M) | 规格文本 |
orders[].price / num | 单价/数量 | 行项目 |
modified | 最后修改时间 | 增量判断基准 |
alipay_id | 支付流水 | 财务对账 |
⚠️ 务必让运营在淘宝后台维护outer_sku_id（商家编码），否则只能靠num_iid+规格文本模糊匹配，容易出错。
五、生产级建议
要点 | 做法 |
|---|---|
增量而非全量 | 用 start_modified/end_modified每5~30分钟拉变更，避免全量翻页 |
Token管理 | DB存 access_token / refresh_token / expires_at，定时任务提前7天刷新课 |
状态机处理 | 新单→待发货(WAIT_SELLER_SEND_GOODS)触发建ERP销售单；TRADE_CLOSED标记取消 |
并发幂等 | 以 tid做唯一键，modified新于本地才更新 |
发货回写 | 拣货后调 taobao.logistics.online.send回写运单号 |
限流 | 列表接口QPS≈5/s，翻页加 sleep(0.2)，遇 code=7 退避重试 |
六、一句话总结（面试/方案）
淘宝订单同步 = 卖家OAuth2授权得到AccessToken → 定时增量调taobao.trades.sold.get(modified)→ 逐单taobao.trade.fullinfo.get取明细 → 按tid幂等写入ERP销售单，关键是用outer_sku_id匹配内部SKU，状态机处理 WAIT_SELLER_SEND_GOODS→发货→完成/关闭。
需要我补
taobao.logistics.online.send发货回写完整参数 或 APScheduler定时增量同步 + Token自动刷新脚本 吗？