🏢【企业级开发】1688采购系统API对接实战（含OAuth 2.0授权完整流程｜附Python源码）
1688采购系统对接与普通商品搜索最大的区别是：订单、物流、购物车、确认收货等接口必须携带买家身份（Access Token），而这个 Token 需要通过 OAuth 2.0 授权码模式 换取。下面给你生产可用的完整链路。
一、1688 OAuth 2.0 授权整体流程
┌─────────┐ ① 拼接授权URL(带redirect_uri) ┌────────────┐ │ 你的ERP │ ────────────────────────────────▶ │ 1688登录页 │ └─────────┘ └────┬─────────┘ 用户登录并同意授权 │ ② 跳回 redirect_uri?code=xxx ▼ ┌────────────────┐ │ 你的回调Server │ └────┬───────────┘ ③ 用 code 换 access_token / refresh_token ④ 存DB，请求API时带 access_token
⚠️ 应用类型必须为「自用型应用」，回调地址须在控制台 → 应用详情 → OAuth配置 中提前登记。
二、Python完整实现：OAuth授权 + Token管理 + 采购API调用
# ali1688_oauth_client.py
import hashlib
import time
import requests
import urllib.parse
from datetime import datetime, timedelta
from typing import Dict, Optional
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ────────────────────────────────────────────────────────
# 1. OAuth 2.0 授权助手
# ────────────────────────────────────────────────────────
class Ali1688OAuth:
"""
1688 OAuth2.0 Authorization Code Flow
文档参考: https://open.1688.com/doc/auth/oauth2.html
"""
AUTH_URL = "https://auth.1688.com/oauth/authorize"
TOKEN_URL = "https://gw.open.1688.com/openapi/http/1/auth/token"
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_1688") -> str:
"""① 生成用户点击的授权URL（需在1688控制台登记 redirect_uri）"""
params = {
"client_id": self.app_key,
"redirect_uri": self.redirect_uri,
"response_type": "code",
"state": state
}
return self.AUTH_URL + "?" + urllib.parse.urlencode(params)
def exchange_token(self, code: str) -> Dict:
"""
③ 用授权码换 access_token / refresh_token
返回示例:
{
"access_token": "xxx",
"refresh_token": "yyy",
"expires_in": 3600 * 24 * 365, # 通常1年
"refresh_token_expires_in": 3600 * 24 * 365 * 3
}
"""
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 result.get("error_code"):
raise Exception(f"OAuth失败: {result}")
return result
def refresh_access_token(self, refresh_token: str) -> Dict:
"""用 refresh_token 续期（建议提前7天检测并刷新）"""
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=15)
resp.raise_for_status()
return resp.json()
# ────────────────────────────────────────────────────────
# 2. 带 OAuth Token 的 1688 采购 API Client
# ────────────────────────────────────────────────────────
class Ali1688PurchaseClient:
"""采购相关接口（需 AccessToken）：订单列表/详情/确认收货）"""
GW = "https://gw.open.1688.com/openapi/http/2/1"
def __init__(self, app_key: str, app_secret: str, access_token: str):
self.ak = app_key
self.as_ = app_secret
self.tk = access_token
# ─── 签名 ───
def _sign(self, params: Dict) -> str:
filt = sorted((k, v) for k, v in params.items()
if v is not None and str(v).strip() != '' and k != 'sign')
qs = ''.join(f"{k}{v}" for k, v in filt)
return hashlib.md5(f"{self.as_}{qs}{self.as_}".encode()).hexdigest().upper()
def _call(self, method: str, biz: Dict) -> Dict:
api_p = {
"method": method, "app_key": self.ak, "session": self.tk,
"timestamp": str(int(time.time() * 1000)),
"format": "json", "v": "2.0", "sign_method": "md5"
}
api_p["param2"] = urllib.parse.quote_plus(str(biz).replace("'", '"'))
api_p["sign"] = self._sign(api_p)
r = requests.get(self.GW, params=api_p, timeout=15)
r.raise_for_status()
d = r.json()
if "error_response" in d:
err = d["error_response"]
raise Exception(f"1688 [{err.get('code')}]: {err.get('msg')}")
return d.get(list(d.keys() - {"error_response"})[0], {})
# ─── 采购订单列表 ───
def list_purchase_orders(self, status="waitsellersend",
hours_back=72, page=1, sz=50) -> list:
from datetime import timedelta
biz = {
"orderStatus": status,
"gmtCreateStart": (datetime.now() - timedelta(hours=hours_back))
.strftime("%Y-%m-%d %H:%M:%S"),
"gmtCreateEnd": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"pageNo": page, "pageSize": sz
}
res = self._call("alibaba.trade.buyer.list", biz)
return res.get("alibaba_trade_buyer_list_response", {}
).get("tradeModelList", []) or []
# ─── 订单明细 ───
def get_order_detail(self, order_id: str) -> Dict:
biz = {"orderId": str(order_id)}
res = self._call("alibaba.trade.get", biz)
return res.get("result", {})
# ─── 确认收货（入库后调） ───
def confirm_receive(self, order_id: str, sub_ids: list = None) -> bool:
biz = {"orderId": str(order_id)}
if sub_ids:
biz["subOrderIds"] = ",".join(map(str, sub_ids))
res = self._call("alibaba.trade.confirmReceive", biz)
return res.get("alibaba_trade_confirmreceive_response", {}
).get("result", False)
# =============================================================
# 使用示例
# =============================================================
if __name__ == "__main__":
APP_KEY = "YOUR_APP_KEY"
APP_SECRET = "YOUR_APP_SECRET"
REDIRECT_URI = "https://your-erp.com/callback/ali1688"
oauth = Ali1688OAuth(APP_KEY, APP_SECRET, REDIRECT_URI)
# ====== 第一步：让用户浏览器访问此URL并登录授权 ======
auth_url = oauth.get_authorize_url()
print("🔗 请浏览器访问此URL完成授权：")
print(auth_url)
print("\n授权后 1688 会跳转到:")
print(f" {REDIRECT_URI}?code=XXXXXX&state=erp_1688")
print("-" * 50)
# ====== 第二步：在你的 Web 回调接口中取出 code 换 token ======
# 下面模拟你拿到 code（实际从 request.args['code'] 取）
demo_code = input("粘贴回调返回的 code: ").strip()
if demo_code:
token_info = oauth.exchange_token(demo_code)
access_token = token_info["access_token"]
refresh_token = token_info.get("refresh_token")
expires_at = datetime.now() + timedelta(seconds=int(token_info.get("expires_in", 31536000)))
print(f"\n✅ AccessToken 获取成功！")
print(f" access_token = {access_token[:20]}...")
print(f" 有效期至(本地估算): {expires_at.strftime('%Y-%m-%d')}")
print(f" refresh_token 请存DB，用于自动续期")
# ====== 第三步：用 token 调采购API ======
client = Ali1688PurchaseClient(APP_KEY, APP_SECRET, access_token)
orders = client.list_purchase_orders(status="waitsellersend", hours_back=48)
print(f"\n📋 待发货采购单: {len(orders)} 笔")
for o in orders[:3]:
print(f" 1688单号:{o.get('id')} 金额:{o.get('totalAmount')}")
# 示例：确认收货（谨慎！会真正确认）
# client.confirm_receive("2338123456789000")三、Web回调（Flask最小示例）
# callback_server.py (需 pip install flask)
from flask import Flask, request
from ali1688_oauth_client import Ali1688OAuth
app = Flask(__name__)
OAUTH = Ali1688OAuth("YOUR_APP_KEY", "YOUR_APP_SECRET", "https://your-erp.com/callback/ali1688")
@app.route("/callback/ali1688")
def ali1688_callback():
code = request.args.get("code")
state = request.args.get("state")
if not code:
return "授权失败：未获取到code", 400
token_info = OAUTH.exchange_token(code)
# ★ 存入数据库（user_id ↔ access_token / refresh_token / expire_time）
print("✅ Token已获取，请存入DB:", token_info.get("access_token")[:20])
return f"<h3>1688授权成功！state={state}</h3><p>Token已保存，可关闭此页。</p>"
if __name__ == "__main__":
app.run(port=8000, debug=True)四、Token 管理生产建议
要点 | 做法 |
|---|---|
存储 | DB存 access_token / refresh_token / expires_at，关联内部员工/店铺 |
提前刷新 | 定时任务每天扫，距过期 < 7天 调 refresh_access_token()更新 |
多账号 | 一个1688账号 = 一个 AccessToken，ERP支持绑定多供方账号 |
安全 | AppSecret/ refresh_token加密存储（AES），不在日志打印 |
重试 | API 调失败且返回 token 过期码 → 自动刷新并重试一次 |
五、完整采购对接链路（面试叙述版）
员工登录ERP → 点击「绑定1688账号」 → 跳转1688 OAuth授权页（get_authorize_url） → 回调拿到 code → exchange_token → 存DB 每日定时任务： 用 access_token 调 alibaba.trade.buyer.list(waitsellersend) → 创建/更新内部采购单 → 展示物流单号（alibaba.logistics.trade.ship） → 仓库入库后调 alibaba.trade.confirmReceive
需要我补 采购下单接口alibaba.trade.create完整参数封装（含SKU ID/收货地址） 或 APScheduler定时同步 + Token自动刷新脚本 吗？
