# 3 天接通淘宝 + 1688 + 京东，商品、订单、物流全打通（附 Python 源码）
> 三天不是靠加班堆出来的。是靠「先申请权限、砍掉非核心、把基础设施一次做对」。
## 一、先把话说清楚：3 天能做什么
很多人看到「3 天接通」第一反应是吹牛。所以先把边界划清。
**3 天能做到的：**
- 三个平台的签名、鉴权、限流、重试全部跑通
- 商品（列表 / 详情 / 库存）、订单（增量拉取 / 详情）、物流（运单 / 轨迹）三大资源能拿到数据
- 一个统一的客户端，业务层不需要知道平台差异
- 冒烟测试 + 错误码映射 + 日志，能支撑后续迭代
**3 天做不到的：**
- 完整的商品发布（类目映射、属性对齐要单独做，那是另一个工程）
- 高并发场景（需要压测、连接池调优、缓存设计）
- 全部接口（三个平台加起来几百个接口，只接核心的）
**关键前提：权限必须提前申请。** 这是 3 天能不能成立的最大变量，下一节单独说。
## 二、Day 0：最不可控的是权限，不是代码
这是接过的都懂、没接过的都会踩的坑。
代码写三天能写完，但**应用审核可能要三周**。
| 平台 | 需要什么 | 现实周期 |
| ---- | ---------------------- | ------------------- |
| 淘宝 | 企业开发者认证 + 应用审核 + 接口权限包 | 3~10 个工作日 |
| 1688 | 企业营业执照 + 诚信通账号 | 5~15 个工作日 |
| 京东 | 商家签约 + 应用授权 | 3~10 个工作日，部分类目要额外资质 |
所以真实项目里，**第 0 天该做的是提交申请，不是写代码。**
我的做法是：申请提交后，立刻用沙箱环境或测试账号把代码写完。淘宝有沙箱，1688 和京东部分接口支持测试 appkey，够你把 90% 的逻辑验完。
如果等到权限下来才动手，那就是三周不是三天。
**接入前必须准备好这一份东西：**
```
□ 三个平台的 app_key / app_secret
□ 一个能正常访问的服务器（固定 IP，有些平台要加白名单）
□ 已授权的一个测试店铺（每平台至少一个）
□ 该店铺下 3~5 个真实商品 ID（用于调试）
□ 近 7 天至少 10 笔订单（用于调订单接口）
□ 一个已发货的运单号（用于调物流接口）
```
最后三项最容易被忽略。很多人代码写完了，发现测试店铺是个空店，没商品没订单，接口返回空数组，根本没法验证解析逻辑对不对。
## 三、三天怎么排
### Day 1：签名 + 通道打通（不碰业务）
这一天的目标只有一个：**三个平台各自能成功调用一个最简单的接口。**
不要一上来写业务。第一天的所有时间都花在「让请求能通」上，因为签名这块的坑最深，而且报错信息最不友好。
```python
# signers.py
import hashlib
import hmac
def sign_taobao(params: dict, secret: str) -> str:
"""TOP：MD5(secret + 排序 k+v + secret)，大写"""
keys = sorted(params)
raw = secret + "".join(f"{k}{params[k]}" for k in keys) + secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def sign_alibaba(params: dict, secret: str) -> str:
"""1688：HMAC-SHA1，注意是 HMAC 不是 MD5"""
keys = sorted(params)
raw = "".join(f"{k}{params[k]}" for k in keys)
return hmac.new(secret.encode("utf-8"), raw.encode("utf-8"),
hashlib.sha1).hexdigest().upper()
def sign_jd(params: dict, secret: str) -> str:
"""宙斯：和淘宝一样是 MD5 包夹，但参数名和拼接细节有差异"""
keys = sorted(params)
raw = secret + "".join(f"{k}{params[k]}" for k in keys) + secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
```
三个签名里，1688 最容易错。淘宝和京东都是 MD5 包夹，长得几乎一样，很多人以为 1688 也是，结果一直报签名错误。
**签名调试的三个技巧：**
1. 先拿官方 SDK 算一次，把自己的结果和它对比。别对着文档猜。
2. 注意编码。参数里有中文时，UTF-8 编码后再签名，不是先 URL 编码。
3. 注意大小写。三个平台都要求大写 HEX，`hexdigest()` 出来是小写，必须 `.upper()`。
Day 1 的验收标准：跑一个 `get_time` 或者最轻量的查询接口，三个平台都返回成功。
### Day 2：数据模型 + 三大资源
这一天做两件事：定模型，接资源。
```python
# models.py
from dataclasses import dataclass, field
from decimal import Decimal
@dataclass
class Product:
platform: str
item_id: str
title: str
price: Decimal
stock: int
images: list[str] = field(default_factory=list)
@dataclass
class Order:
platform: str
order_id: str
status: str
pay_amount: Decimal
created_at: str
buyer: str
items: list[dict] = field(default_factory=list)
@dataclass
class Shipment:
platform: str
order_id: str
waybill: str
company: str
status: str = ""
traces: list[dict] = field(default_factory=list)
```
模型只定这三个，不要贪多。字段以「业务真正会用到的」为准，多余的一律不加——加一个字段，三个平台都要写一遍映射，成本是乘三的。
然后按资源依次接：
**商品**：列表（分页）→ 详情 → 库存。顺序不能反，列表能拿到 ID 才能调详情。
**订单**：增量列表 → 详情。增量参数一定要用「修改时间」不是「创建时间」，原因后面踩坑那节说。
**物流**：运单查询 → 轨迹查询。轨迹要么订阅推送，要么低频轮询，别高频拉。
### Day 3：基础设施 + 冒烟测试
第三天不写业务，写「让业务能长期跑」的东西：
- 统一错误码映射
- 重试策略
- 限流
- 日志
- 冒烟测试
这部分是接入期最容易被跳过、但后期最费时间的。跳过它们，你会在上线第一周花掉两倍时间排查「为什么这个接口偶尔失败」。
## 四、核心代码：一个客户端打通三个平台
这套设计的思路：**业务层只调 `client.call(resource, action, params)`，平台差异全在客户端里消化。**
```python
# client.py
import asyncio
import hashlib
import hmac
import time
import random
from dataclasses import dataclass
import httpx
from signers import sign_taobao, sign_alibaba, sign_jd
# ---------- 统一错误类型 ----------
class ErrorKind:
AUTH = "auth" # 签名/授权问题，重试无用
RATE_LIMIT = "rate_limit" # 限流，退避可恢复
SERVER = "server" # 平台 5xx，重试可恢复
PARAM = "param" # 参数错，必须先修代码
BUSINESS = "business" # 业务拒绝（如库存不足），看具体语义
@dataclass
class PlatformError(Exception):
kind: str
code: str
message: str
platform: str
def __str__(self):
return f"[{self.platform}] {self.kind}({self.code}): {self.message}"
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 错误码映射：接入期最值钱的表，把各平台的模糊报错统一成五类
ERROR_MAP = {
"taobao": {
"isv.invalid-signature": ErrorKind.AUTH,
"isv.missing-signature": ErrorKind.AUTH,
"isv.user-not-exist": ErrorKind.AUTH,
"isp.top-remote-connection-timeout": ErrorKind.SERVER,
"isp.top.throttled-by-user-defined-limit": ErrorKind.RATE_LIMIT,
"isv.parameters-mismatch": ErrorKind.PARAM,
},
"alibaba": {
"401": ErrorKind.AUTH,
"403": ErrorKind.AUTH,
"429": ErrorKind.RATE_LIMIT,
"500": ErrorKind.SERVER,
"502": ErrorKind.SERVER,
},
"jd": {
"1001": ErrorKind.AUTH,
"1002": ErrorKind.AUTH,
"1099": ErrorKind.RATE_LIMIT,
"1005": ErrorKind.SERVER,
},
}
ENDPOINTS = {
"taobao": "https://eco.taobao.com/router/rest",
"alibaba": "https://gw.open.1688.com/openapi/param2/1",
"jd": "https://api.jd.com/routerjson",
}
class PlatformClient:
"""
三平台统一客户端。业务层不需要知道任何平台差异。
"""
def __init__(self, platform: str, app_key: str, secret: str,
qps: int = 5, timeout: float = 15.0):
self.platform = platform
self.app_key = app_key
self.secret = secret
self.qps = qps
self.sem = asyncio.Semaphore(qps)
self.http = httpx.AsyncClient(timeout=timeout)
self._last_call = 0.0
async def _throttle(self):
"""简单匀速限流：保证调用间隔 >= 1/qps"""
interval = 1.0 / self.qps
wait = self._last_call + interval - time.time()
if wait > 0:
await asyncio.sleep(wait)
self._last_call = time.time()
def _sign(self, params: dict) -> str:
fn = {"taobao": sign_taobao,
"alibaba": sign_alibaba,
"jd": sign_jd}[self.platform]
return fn(params, self.secret)
def _build(self, method: str, params: dict) -> tuple[str, dict]:
if self.platform == "taobao":
body = {
"method": method, "app_key": self.app_key,
"timestamp": str(int(time.time())),
"format": "json", "v": "2.0", "sign_method": "md5",
**params,
}
body["sign"] = self._sign(body)
return ENDPOINTS["taobao"], body
if self.platform == "alibaba":
body = {"access_token": self.app_key, **params}
body["_aop_signature"] = self._sign(body)
api = method.replace(".", "/")
return f"{ENDPOINTS['alibaba']}/{api}/{self.app_key}", body
# jd
body = {
"method": method, "app_key": self.app_key,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"format": "json", "v": "2.0",
**params,
}
body["sign"] = self._sign(body)
return ENDPOINTS["jd"], body
async def call(self, method: str, params: dict | None = None,
retries: int = 3) -> dict:
"""
统一调用入口。可重试错误自动退避，不可重试的立刻抛。
"""
params = params or {}
url, body = self._build(method, params)
last_err = None
for attempt in range(retries):
try:
async with self.sem:
await self._throttle()
t0 = time.time()
r = await self.http.post(url, data=body)
cost = (time.time() - t0) * 1000
data = r.json()
self._log(method, cost, r.status_code)
err = self._extract_error(data)
if err is None:
return data
if err.kind in (ErrorKind.AUTH, ErrorKind.PARAM, ErrorKind.BUSINESS):
raise err # 重试无意义
last_err = err
except PlatformError:
raise
except Exception as e:
last_err = PlatformError(
ErrorKind.SERVER, "NETWORK", str(e), self.platform)
# 指数退避 + 抖动，避免同时重试打爆平台
backoff = (2 ** attempt) + random.uniform(0, 0.5)
await asyncio.sleep(backoff)
raise last_err or PlatformError(
ErrorKind.SERVER, "UNKNOWN", "重试耗尽", self.platform)
def _extract_error(self, data: dict) -> PlatformError | None:
"""把三种不同的错误结构统一成 PlatformError"""
if self.platform == "taobao":
e = data.get("error_response")
if e:
code = e.get("sub_code") or e.get("code", "")
return PlatformError(
ERROR_MAP["taobao"].get(code, ErrorKind.BUSINESS),
code, e.get("sub_msg") or e.get("msg", ""), "taobao")
if self.platform == "alibaba":
if "error_code" in data:
code = str(data["error_code"])
return PlatformError(
ERROR_MAP["alibaba"].get(code, ErrorKind.BUSINESS),
code, data.get("error_message", ""), "alibaba")
if self.platform == "jd":
if "error_response" in data:
e = data["error_response"]
code = str(e.get("code", ""))
return PlatformError(
ERROR_MAP["jd"].get(code, ErrorKind.BUSINESS),
code, e.get("zh_desc") or e.get("en_desc", ""), "jd")
return None
def _log(self, method: str, cost_ms: float, status: int):
# 接你现有的日志系统；接入期建议先打到 stdout
print(f"[{self.platform}] {method} {status} {cost_ms:.0f}ms")
async def close(self):
await self.http.aclose()
```
这个客户端里，**最值钱的是 `ERROR_MAP` 和错误分类**。
接入期你一定会遇到「接口偶尔失败」，如果不做分类，你只能无脑重试。而重试签名错误是纯浪费——签名错了重试一百次还是错，只是把配额烧光。
分成五类之后，策略就清楚了：`AUTH` 和 `PARAM` 立刻抛、让人看；`RATE_LIMIT` 退避；`SERVER` 重试；`BUSINESS` 交给业务层判断。
## 五、三大资源的具体封装
```python
# resources.py
from decimal import Decimal
from client import PlatformClient
class ProductAPI:
def __init__(self, client: PlatformClient):
self.c = client
async def get(self, item_id: str) -> dict:
"""商品详情，三平台归一"""
p = self.c.platform
if p == "taobao":
d = await self.c.call("taobao.item.seller.get", {
"fields": "num_iid,title,price,num,quantity,pic_url",
"num_iid": item_id})
it = d["item_get_response"]["item"]
return {"item_id": str(it["num_iid"]), "title": it.get("title", ""),
"price": Decimal(str(it.get("price", "0"))),
"stock": int(it.get("quantity", 0))}
if p == "alibaba":
d = await self.c.call("com.alibaba.product.get", {
"productID": item_id, "webSite": "1688"})
r = d.get("result") or {}
ranges = (r.get("saleInfo") or {}).get("priceRanges") or []
return {"item_id": str(r.get("productID", item_id)),
"title": r.get("subject", ""),
"price": Decimal(str(ranges[0]["price"])) if ranges else Decimal("0"),
"stock": int((r.get("saleInfo") or {}).get("amountOnSale", 0))}
d = await self.c.call("jd.union.open.goods.query", {
"360buy_param_json": '{"skuIds":[%s]}' % item_id})
rows = ((d.get("jd_union_open_goods_query_responce") or {})
.get("result") or {}).get("data") or []
if not rows:
return {}
g = rows[0]
return {"item_id": str(g.get("skuId")), "title": g.get("skuName", ""),
"price": Decimal(str(g.get("price", "0"))), "stock": 0}
class OrderAPI:
def __init__(self, client: PlatformClient):
self.c = client
async def list_increment(self, start: str, end: str, page: int = 1) -> list[dict]:
"""
按修改时间增量拉。start/end 格式 YYYY-MM-DD HH:MM:SS
注意：用修改时间，不是创建时间
"""
p = self.c.platform
if p == "taobao":
d = await self.c.call("taobao.trades.sold.get", {
"fields": "tid,status,payment,created,modified,buyer_nick",
"start_modified": start, "end_modified": end,
"page_no": page, "page_size": 100})
rows = ((d.get("trades_sold_get_response") or {})
.get("trades") or {}).get("trade") or []
return [{"order_id": str(t["tid"]), "status": t.get("status", ""),
"pay_amount": Decimal(str(t.get("payment", "0"))),
"created_at": t.get("created"),
"buyer": t.get("buyer_nick", "")} for t in rows]
if p == "alibaba":
d = await self.c.call("com.alibaba.trade/get.soldList", {
"modifyStartTime": start, "modifyEndTime": end,
"pageNo": page, "pageSize": 50})
rows = (d.get("result") or {}).get("orderList") or []
return [{"order_id": str(o.get("orderId")),
"status": o.get("orderStatus", ""),
"pay_amount": Decimal(str(o.get("sumPayment") or "0")),
"created_at": o.get("createTime"),
"buyer": o.get("toLoginId", "")} for o in rows]
d = await self.c.call("jingdong.pop.order.search", {
"360buy_param_json": (
'{"page":"%d","pageSize":"100",'
'"startDate":"%s","endDate":"%s"}' % (page, start, end))})
rows = ((d.get("jingdong_pop_order_search_responce") or {})
.get("searchorderinfo_result") or {}).get("orderInfoList") or []
return [{"order_id": str(o.get("orderId")),
"status": str(o.get("orderState", "")),
"pay_amount": Decimal(str(o.get("orderPayment") or "0")),
"created_at": o.get("orderStartTime"),
"buyer": o.get("pin", "")} for o in rows]
class LogisticsAPI:
def __init__(self, client: PlatformClient):
self.c = client
async def trace(self, waybill: str, company: str) -> list[dict]:
"""物流轨迹。接入期先用轮询验证通路，生产环境要改订阅"""
p = self.c.platform
if p == "taobao":
d = await self.c.call("taobao.logistics.trace.search", {
"tid": waybill})
elif p == "alibaba":
d = await self.c.call("com.alibaba.logistics/getLogisticsTraceInfo", {
"logisticsId": waybill})
else:
d = await self.c.call("jingdong.etms.trace.get", {
"360buy_param_json": '{"waybillCode":"%s"}' % waybill})
# 三个平台返回结构不同，只取「时间 + 描述」两个字段
nodes = (d.get("traces") or d.get("traceList")
or d.get("data") or [])
return [{"time": n.get("time") or n.get("acceptTime"),
"desc": n.get("desc") or n.get("standerDesc") or n.get("content")}
for n in nodes]
```
注意 `LogisticsAPI.trace` 里那段节点提取——用 `or` 链式兜底。这种写法不好看，但接入期很实用：你在没有真实返回样本的情况下，先把几种可能的字段名都兜住，跑通之后再根据实际返回精简。
## 六、Day 3 的验收：冒烟测试
这是第三天必须写的东西。没有它，你永远不知道「接通了」是接通了几个接口。
```python
# smoke_test.py
import asyncio
import sys
from client import PlatformClient
from resources import ProductAPI, OrderAPI, LogisticsAPI
CHECKS = [
# (平台, 资源, 说明)
("taobao", "product", "商品详情"),
("taobao", "order", "订单增量"),
("taobao", "logistics", "物流轨迹"),
("alibaba", "product", "商品详情"),
("alibaba", "order", "订单增量"),
("alibaba", "logistics", "物流轨迹"),
("jd", "product", "商品详情"),
("jd", "order", "订单增量"),
("jd", "logistics", "物流轨迹"),
]
CONFIG = {
"taobao": {"key": "your_key", "secret": "your_secret",
"item": "654321098765", "waybill": "SF1234567890"},
"alibaba": {"key": "your_key", "secret": "your_secret",
"item": "612345678901", "waybill": "YT9876543210"},
"jd": {"key": "your_key", "secret": "your_secret",
"item": "100012043978", "waybill": "JD0001234567"},
}
async def check_one(platform: str, resource: str) -> tuple[bool, str]:
cfg = CONFIG[platform]
client = PlatformClient(platform, cfg["key"], cfg["secret"])
try:
if resource == "product":
r = await ProductAPI(client).get(cfg["item"])
ok = bool(r.get("item_id"))
return ok, f"标题={r.get('title', '')[:20]} 价格={r.get('price')}"
if resource == "order":
r = await OrderAPI(client).list_increment(
"2026-10-15 00:00:00", "2026-10-22 00:00:00")
return True, f"拉到 {len(r)} 单"
r = await LogisticsAPI(client).trace(cfg["waybill"], "SF")
return True, f"{len(r)} 个轨迹节点"
except Exception as e:
return False, str(e)[:80]
finally:
await client.close()
async def main():
print(f"{'平台':<10}{'资源':<12}{'结果':<6}详情")
print("-" * 70)
failed = 0
for platform, resource, _ in CHECKS:
ok, detail = await check_one(platform, resource)
mark = "PASS" if ok else "FAIL"
if not ok:
failed += 1
print(f"{platform:<10}{resource:<12}{mark:<6}{detail}")
print("-" * 70)
print(f"通过 {len(CHECKS) - failed}/{len(CHECKS)}")
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(asyncio.run(main()))
```
这个脚本每天跑一次，接到监控里。**它能提前发现三类问题**：授权过期、平台改字段、限流额度变了。这三件事都是「不主动检查就不知道，等业务报错才发现」的。
## 七、真实时间分配
回头看这三天，时间实际花在哪（一个做过三次接入的人的经验值）：
| 环节 | 预计 | 实际 | 差异原因 |
| --------- | ----- | ----- | ------------------------- |
| 签名调试 | 0.5 天 | 1 天 | 1688 的 HMAC 和文档示例不一致，调了半天 |
| 接口联调 | 1 天 | 1.5 天 | 没有真实测试数据，返回空数组无从验证 |
| 字段映射 | 0.5 天 | 0.5 天 | 符合预期 |
| 物流 | 0.5 天 | 0.5 天 | 简单 |
| 错误处理 + 冒烟 | 0.5 天 | 1 天 | 错误码文档不全，靠试出来的 |
签名和错误码这两块，是预估最容易失准的地方。**预留 1 天余量给这两件事**，3 天才真的成立。
## 八、五个接入期必踩的坑
1. **权限没下来就开始写代码，导致最后交付延期三周。** 正确做法：Day 0 提交申请，同时用沙箱把代码写完。
2. **1688 的签名不是 MD5。** 淘宝和京东都是 `MD5(secret + k+v + secret)`，1688 是 HMAC-SHA1。三个长得像，很容易抄错。
3. **订单增量用创建时间，漏掉状态变化。** 客户半夜改地址、改价、退款，`gmt_create` 不变但 `gmt_modified` 变了。必须用修改时间。
4. **重试没有分类，把配额烧光。** 签名错误重试一百次还是错，只是更浪费。先分类，再决定重不重试。
5. **测试店铺是空的。** 没有真实商品和订单，接口返回 `[]`，你无法验证解析逻辑。接入前一定要准备测试数据。
## 九、最后
三天接通的核心不是手速，是三件事：
**提前申请权限**（让它和写代码并行）、**先通通道再写业务**（第一天只调一个最轻的接口）、**把基础设施一次做对**（错误分类、重试、限流、冒烟测试）。
这三件事做对了，三天是够的。做错了，三周也接不完。
代码在上面，`PlatformClient` 那个类可以直接用——加第四个平台，只需要在 `_build` 和 `ERROR_MAP` 里加一段，业务层一行不用改。
如果你不想在签名、限流、错误码映射这些地方花掉一两天（尤其 1688 的签名和三个平台各自的报错结构），也可以用现成的三平台 API 接口服务，统一格式输出，把三天里的 Day 1 直接省掉，从 Day 2 开始做业务。
有在接的可以聊聊，你们接入花了多久，卡在哪一步。
---
**标签**：`Python` `电商` `API` `系统集成` `架构设计`
