# 鉴权、签名、字段映射：三平台对接最容易翻车的三个环节（附 python 源码）
> 接淘宝、1688、京东的开放平台接口，我们团队前后踩过的坑能写一本书。但真要按「浪费的人天」排个序，前三名毫无悬念：**鉴权、签名、字段映射**。
>
> 这三个环节有个共同的恶毒之处：**报错信息都不说人话**。淘宝返回 `Invalid signature`，京东返回 `signature error`，1688 返回 `error_code: 40`——没有一个告诉你错在哪一个参数、哪一步、哪个字符。很多人卡一天半，最后发现是时间戳用了秒而不是毫秒。
>
> 这篇不讲怎么接入（第 6、11 篇讲过），讲的是**翻车之后怎么定位**。建议先收藏，翻车的时候拿出来查。
## 一、为什么这三个环节最坑：错误都在「你自己的代码」和「平台的报错」之间
数据接口报错，通常是业务问题——库存不足、商品不存在，一眼能懂。而这三个环节的错误发生在**请求离开你的机器之前**，平台收到的东西和你以为发出去的东西**不是同一份**：
- 你以为发了 `timestamp=1730000000`，实际签名时用的是另一个值；
- 你以为参数按字典序排了，实际有一个 key 多了个空格；
- 你以为字段是整数，平台按字符串解析，静默截断。
**排查的核心方法论只有一句话：别猜，把「你实际发出去的东西」打出来，和「平台期望的东西」逐字符比对。** 后文所有代码都是围绕这句话展开的。
先给一张总览表，后面三节逐个展开：
| 环节 | 典型报错 | 报错的真实含义 | 平均排查耗时（我们的工单统计） |
| ---- | ------------------------------------------- | ------------------------------------------ | --------------- |
| 鉴权 | `session expired` / `lack of authorization` | token 过期、session 与 appkey 不匹配、token 没跟着请求走 | 2 小时 |
| 签名 | `Invalid signature` / `error_code: 40` | 参数没排对序、时间戳单位错、中文被转义、空值参与签名 | **1.5 天** |
| 字段映射 | 无报错（最危险） | 价格少两位、状态映射错、时间差 8 小时 | 3 天起，且常由客户先发现 |
注意字段映射那一行——**它经常没有报错**。这是三个环节里唯一可能酿成资损的一类，放最后一节重点讲。
## 二、鉴权：三种 token，三种死法
三平台的授权模型长得像，细节全不一样。先列差异：
| 平台 | 凭证结构 | 有效期 | 过期前能刷新吗 |
| ---- | ---------------------------------- | ----------------------- | ------------------------ |
| 淘宝 | `appkey` + `sessionKey`（绑定商家） | 长期但可被商家主动失效 | 会话失效后需商家重新授权，**无法程序化刷新** |
| 京东 | `app_key` + `access_token`（OAuth2） | 通常 1 年，refresh_token 更长 | 可以，用 refresh_token 换新 |
| 1688 | `appkey` + `access_token`（OAuth2） | 约 10 小时（各应用不同） | 可以，refresh_token 换新 |
**死法一：混淆 app 级凭证和用户级凭证。**
`appkey/secret` 只代表「你的应用」，调用商品、订单接口还必须带上「商家授权」的 token。最经典的事故：测试环境用 A 商家的 session，生产环境忘了换，结果测试数据全对、上线后拉的全是另一个店铺的商品——**接口返回 200，数据是别人的**。所以凭证要「应用 × 商家」二元组建模，别把 sessionKey 存成全局配置。
**死法二：拿本地时钟判断 token 过期。**
token 的过期时间是平台返回的 `expires_in`（剩余秒数），如果你存的是 `本地 now + expires_in`，服务器时钟漂移 5 分钟，token 就会提前或延后 5 分钟失效。**必须存「到期时间点」，且以收到响应那一刻为基准**，再留安全余量。
**死法三：淘宝的「商家悄悄解绑」。**
淘宝 session 没有程序化刷新路径，商家在千牛后台解除授权后，你的 token 立刻全挂，且**没有提前通知**。这是淘宝接入最阴的一雷。解法是健康探测（第 9 篇的 PROBES 思路）+ 授权状态监控，挂了第一时间找到商家重新授权。
代码上，一个最小可用的 `TokenManager`：
```python
# auth/tokens.py
import asyncio
import time
from dataclasses import dataclass, field
@dataclass
class Token:
value: str
refresh_value: str | None
expires_at: float # 平台响应到达那一刻 + expires_in，时间点而非时长
refresh_expires_at: float | None = None
def needs_refresh(self, margin_s: int = 600) -> bool:
"""提前 10 分钟就判为「该刷了」，余量吃掉网络延迟和时钟误差"""
return time.time() >= self.expires_at - margin_s
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TokenManager:
"""按 (platform, seller_id) 二元组管理，杜绝串店铺"""
def __init__(self, store):
self.store = store # 持久层：sqlite / redis，重启不丢
self._locks: dict[str, asyncio.Lock] = {}
self._mem: dict[str, Token] = {}
def _lock(self, key: str) -> asyncio.Lock:
if key not in self._locks:
self._locks[key] = asyncio.Lock()
return self._locks[key]
async def get(self, platform: str, seller_id: str) -> str:
key = f"{platform}:{seller_id}"
tok = self._mem.get(key) or await self.store.load(key)
if tok is None:
raise AuthError(f"{key} 无凭证，需要商家重新授权")
if tok.needs_refresh():
async with self._lock(key): # 单飞：并发请求只触发一次刷新
tok = self._mem.get(key) or tok
if tok.needs_refresh(): # 双检：拿锁后再确认一次
try:
tok = await self._refresh(platform, seller_id, tok)
except RefreshFailed:
# refresh_token 也失效：只能走重新授权，主动告警而不是等调用报错
await self.store.mark_broken(key)
raise AuthError(f"{key} refresh_token 已失效，需商家重新授权")
return tok.value
async def _refresh(self, platform, seller_id, tok) -> Token:
if platform == "taobao":
raise RefreshFailed("淘宝无程序化刷新，走重新授权流程")
# 京东 / 1688：POST token 端点，grant_type=refresh_token
resp = await oauth_refresh(platform, tok.refresh_value)
new = Token(
value=resp["access_token"],
refresh_value=resp.get("refresh_token", tok.refresh_value),
expires_at=time.time() + int(resp["expires_in"]),
refresh_expires_at=time.time() + int(resp.get("refresh_token_valid_time", 0)),
)
key = f"{platform}:{seller_id}"
self._mem[key] = new
await self.store.save(key, new)
return new
```
三个细节值得注意：
1. **`needs_refresh` 提前 10 分钟**——不是 token 到期才刷，余量吃掉时钟误差和刷新请求本身失败后的重试空间。
2. **单飞锁 + 双检**——高并发下 100 个请求同时发现 token 过期，只允许 1 个真正去刷新，其余等结果。没有锁就是「并发刷新撞车」：后发的刷新会把先发的 token 作废（京东和 1688 都有这个行为），自己把自己刷死。多店铺规模上去之后这块还要加密存储和分店隔离，是下一篇的主题。
3. **淘宝直接抛 `RefreshFailed`**——诚实地告诉调用方「这条路径不存在」，别写个永远成功的假刷新。
## 三、签名：报错只说错了，不说错在哪
签名错误是接入期**单点耗时最长**的环节。原因很简单：平台不会告诉你它算签名的输入是什么，你只能猜自己哪个参数「脏」了。
先给三平台签名算法速查（实现细节第 11 篇有完整代码，这里只列排障需要的差异点）：
| 平台 | 算法 | 关键差异 | 最高频错误 |
| ---- | ---------------------------- | ---------------------------- | -------------------- |
| 淘宝 | MD5（`secret + 排序串 + secret`） | GET/POST 均按参数排序拼 | 参数没按 ASCII 排序 |
| 京东 | MD5（`secret + 排序串 + secret`） | 业务参数整体塞进 `360buy_param_json` | JSON 中文被转义成 `\uXXXX` |
| 1688 | HMAC-SHA1 | **urlPath 参与签名** + **毫秒时间戳** | 时间戳用了秒 |
### 签名错误 Top 5（按我们工单频次排序）
**① 时间戳单位或时区错。** 1688 要毫秒，淘宝/京东用秒也能过但会有诡异边界；服务器是 UTC 时区的话，某些按「分钟窗口」校验时间戳的接口直接全挂。**解法：时间戳永远显式生成，别用框架默认值。**
**② 参数没按 ASCII 字典序排。** Python 的 `sorted()` 恰好就是 ASCII 序，所以纯 Python 代码很少踩；踩坑的都是「从别的语言抄的签名代码」或者「自己手写了拼接顺序」。
**③ 中文被转义。** 京东最典型：`360buy_param_json` 里的中文商品名，`json.dumps` 默认 `ensure_ascii=True` 会转成 `\uXXXX`——签名按转义后的算，服务端按原文验，**永远对不上**。一行 `ensure_ascii=False` 解决，但不知道的时候能卡一天半。
**④ 空值参与了签名。** 某些平台规定空参数不参与签名，你把 `""` 也拼进去了。规则各平台不同，**拿不准就试：先全拼，报错再去掉空值试一次**。
**⑤ secret 泄露后的轮换灾难。** secret 提交到公开仓库后去平台后台重置，结果忘了旧 secret 还配在测试机 cron 里——旧任务报签名错，新任务正常，两边互相干扰极难定位。**secret 必须集中存放、一处引用**。
### 排查方法：把「实际发出去的」打出来比对
下面这个 `sign_debug.py` 是我们内部排查签名问题的工具，思路是：**本地复算签名，同时把参与签名的 canonical string 完整打出来**。拿它和平台的报错对照，90% 的签名错误肉眼就能看出来：
```python
# sign_debug.py —— 签名排障工具：复算 + canonical string 可视化
import hashlib
import hmac
import json
import time
from urllib.parse import quote
def canonical(params: dict) -> str:
"""淘宝/京东式 canonical string：ASCII 序 k=v 直接相连，跳过空值和签名本身"""
pairs = sorted((k, v) for k, v in params.items()
if v not in ("", None) and k != "sign")
return "".join(f"{k}{v}" for k, v in pairs)
def sign_taobao(params: dict, secret: str) -> str:
raw = f"{secret}{canonical(params)}{secret}"
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def sign_jd(params: dict, secret: str, body: dict | None = None) -> str:
if body is not None:
# 关键：ensure_ascii=False，中文按原文参与签名
params = {**params, "360buy_param_json": json.dumps(
body, ensure_ascii=False, separators=(",", ":"))}
raw = f"{secret}{canonical(params)}{secret}"
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def sign_alibaba(params: dict, secret: str, url_path: str) -> str:
# 1688：urlPath + 排序串，HMAC-SHA1，毫秒时间戳
params = {**params, "_aop_timestamp": str(int(time.time() * 1000))}
ordered = sorted((k, v) for k, v in params.items() if v not in ("", None))
raw = url_path + "".join(f"{k}{v}" for k, v in ordered)
digest = hmac.new(secret.encode(), raw.encode(), hashlib.sha1).hexdigest().upper()
return digest
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def debug_dump(name: str, params: dict, secret: str, url_path: str = ""):
"""把参与签名的每一步都打出来，和平台的报错对照"""
cs = url_path + canonical(params)
print(f"=== {name} ===")
for k, v in sorted(params.items()):
print(f" {k!r:28} -> {v!r}") # repr 能暴露首尾空格、全角字符
print(f" canonical ({len(cs)} chars):")
print(f" {cs}")
print(f" sign: {sign_taobao(params, secret)}")
if __name__ == "__main__":
debug_dump("taobao.item.get", {
"method": "taobao.item.get",
"app_key": "your_appkey",
"timestamp": "2026-10-29 15:00:00",
"num_iid": "683625971234",
"fields": "num_iid,title,price",
"session": "your_session",
"note ": "value ends with a space", # 故意放一个带尾空格的参数做演示
}, "your_secret")
```
用法：报签名错的时候，把**你实际发请求用的那份参数**（不是你以为的那份——从 httpx 的请求日志或抓包里拿）灌进 `debug_dump`，然后：
1. 看 `repr` 输出里有没有**首尾空格、全角字符、大小写不一致**的 key；
2. 看 canonical string 的长度对不对——多一段少一段，二分着删参数试；
3. 时间戳单独确认单位（秒/毫秒）和格式。
**一个能省 80% 排查时间的习惯**：签名函数写好后立刻用官方文档里的「签名示例」跑一遍断言（三平台文档都附了带预期结果的示例参数），断言不过说明你的实现有问题，别等真请求了才发现。
## 四、字段映射：不报错，只出事
前两个环节的错误平台会告诉你（虽然说得含糊），字段映射的错误**平台根本不知道错了**——你拉回来的就是一份合法响应，只是你读它的方式不对。这类问题的共同特征：**上线数天后由运营或客户发现，排查时谁也不怀疑数据管道**。
### 三类错误 + 一个语义陷阱
**① 数值类型。** 三个平台的价格给法不一样：淘宝给字符串 `"59.90"`，京东给「分」为单位的整数 `5990`，1688 阶梯价给嵌套数组。最恶劣的是 `5990`——它既能当「分」也能当「元」，**错了不报错，只差 100 倍**。铁律：入库前统一转 `Decimal`，转换时必须显式指定来源单位。
**② 时间字段。** 秒还是毫秒、UTC 还是北京时间、字符串还是整数——三平台三样，甚至同平台不同接口都不一样。差 8 小时的时区错误会让「今日订单」报表永远差一半，而财务通常在月底才对出来。铁律：**收到即归一**，全部转成 UTC 毫秒时间戳存库，展示层再转本地。
**③ 状态枚举。** 订单状态三平台完全不同：淘宝 `WAIT_SELLER_SEND_GOODS`，京东 `WAIT_SELLER_STOCK_OUT`，1688 用数字码。直接把平台状态存库，后面每写一个业务逻辑都要写一遍三平台的 if-else。铁律：**入库即映射到自有状态机**（第 3 篇的订单状态机就是那套字典）。
**④ 空值语义陷阱（最隐蔽）。** `"price": null`、`"price": ""`、key 整个不出现——三种「没有价格」的含义完全不同：null 通常是「该字段不适用」，空字符串可能是「平台接口有 bug」，key 缺失可能是「你没申请这个字段的权限」。**不区分这三种，价格同步会把一批「查询权限没申请」的商品当成「无货」处理**，这直接就是资损。
### 声明式映射：把「翻译规则」和数据代码分开
解决思路是老三样（第 2 篇的统一模型），这里给出映射层的可运行骨架，重点是**把单位、时区、枚举、空值四类规则显式声明出来**，而不是散落在业务代码里：
```python
# mapping.py —— 声明式字段映射 + 空值语义
from dataclasses import dataclass
from decimal import Decimal
@dataclass
class FieldSpec:
path: str # 平台响应里的取值路径
to: str # 统一模型字段名
unit: str = "raw" # "raw" | "cent" | "second" | "millisecond"
enum: dict | None = None # 平台状态 -> 自有状态
required: bool = True
def pick(data: dict, path: str):
"""按 a.b.c 取值；区分三种空：键缺失 / null / 空串"""
cur = data
for part in path.split("."):
if not isinstance(cur, dict) or part not in cur:
return Missing.ABSENT # key 缺失：多为权限未申请
cur = cur[part]
if cur is None:
return Missing.NULL # 明确的 null：字段不适用
if cur == "":
return Missing.EMPTY # 空串：疑似平台侧异常，单独上报
return cur
class Missing:
ABSENT = object(); NULL = object(); EMPTY = object()
def normalize(raw, spec: FieldSpec):
v = pick(raw, spec.path)
if isinstance(v, type(Missing.ABSENT)): # 用 is 判定哨兵对象
if spec.required:
raise MappingError(f"{spec.path} 缺失：检查字段权限是否已申请")
return None
if v is Missing.NULL:
return None
if v is Missing.EMPTY:
raise MappingError(f"{spec.path} 为空串：疑似接口异常，人工确认")
if spec.unit == "cent": # 京东价格：分 -> 元
v = Decimal(v) / 100
elif spec.unit == "second": # 秒 -> 毫秒
v = int(v) * 1000
if spec.enum: # 状态映射，未收录的显式报错
if v not in spec.enum:
raise MappingError(f"未收录状态 {v!r}，先补映射表再上线")
v = spec.enum[v]
return v
# ---- 三平台的映射表：规则集中在这里，业务代码只见统一模型 ----
TAOBAO_TRADE = [
FieldSpec("tid", "order_no"),
FieldSpec("created", "created_at", unit="second"),
FieldSpec("status", "status",
enum={"WAIT_SELLER_SEND_GOODS": "TO_SHIP",
"TRADE_FINISHED": "DONE"}),
FieldSpec("payment", "paid_amount"), # 淘宝字符串元
]
JD_TRADE = [
FieldSpec("orderId", "order_no"),
FieldSpec("orderStartTime", "created_at", unit="millisecond"),
FieldSpec("orderState", "status",
enum={"WAIT_SELLER_STOCK_OUT": "TO_SHIP",
"FINISHED": "DONE"}),
FieldSpec("orderPrice", "paid_amount", unit="cent"), # 京东分
]
def map_trade(raw: dict, table: list[FieldSpec]) -> dict:
return {spec.to: normalize(raw, spec) for spec in table}
```
```
这套结构的价值在三个「显式」：
1. **单位显式**——`unit="cent"` 写在映射表里，永远不会有人把京东的价格直接当元用；
2. **枚举显式**——平台新增一个状态码，映射层当场报错拦下，而不是静默落一个未知状态进库；
3. **空值显式**——三种「没有」三种处理，权限没申请的问题在管道层就炸出来，不会变成「无货」。
再配一个 **round-trip 自检**，映射表改坏了 CI 直接红：
```python
def test_roundtrip():
sample = {"orderId": "JD123", "orderStartTime": 1730000000000,
"orderState": "WAIT_SELLER_STOCK_OUT", "orderPrice": 5990}
out = map_trade(sample, JD_TRADE)
assert out["order_no"] == "JD123"
assert out["status"] == "TO_SHIP"
assert out["paid_amount"] == Decimal("59.90") # 分 -> 元，别差 100 倍
```
## 五、翻车速查表：先定位环节，再查原因
最后把三个环节收成一张速查表。翻车的时候**先对症状定环节**，再去对应小节查原因，比盲目重试省一个工作日：
| 症状 | 环节 | 最可能的原因（按概率排序） |
|---|---|---|
| 所有请求统一报鉴权/签名错，昨天还好好的 | 鉴权 | token 过期 / 商家解绑授权 / secret 被轮换 |
| 只有部分商家报鉴权错 | 鉴权 | 个别商家 session 失效，串店铺配置 |
| 签名错，参数肉眼检查全对 | 签名 | 中文被转义（京东）/ 时间戳单位（1688 毫秒）|
| 签名错，加一个参数就开始报错 | 签名 | 新参数含特殊字符或空值参与规则不同 |
| 无报错，但数据「看起来不对」 | 字段映射 | 价格单位（分/元）、时间时区（8 小时）|
| 某类商品的数据总是缺字段 | 字段映射 | 字段权限未申请，pick 到 ABSENT |
| 报表数字对不上，接口日志全绿 | 字段映射 | 状态映射漏了新枚举 / 空值被当 0 |
三个环节的共同预防措施其实就一条：**接入期写冒烟测试（第 6 篇的 9 项），把签名断言、token 刷新、字段映射 round-trip 都放进 CI**。翻车不可怕，可怕的是翻车发生在客户通知你之后。
---
这三件事做扎实，接入期 80% 的坑就避掉了。剩下的坑集中在**多店铺授权管理**——token 加密存储、几十家店并发刷新撞车、单店过期不拖累其他店——那是下一篇的主题，也是整个接入系列的收尾。
如果不想自己踩这些坑，也可以直接用我们封装好的接口服务，鉴权续期、签名托管、字段归一这些都是现成的，欢迎评论区或私信聊。
---
*系列文章：上一篇《商品详情、SKU、价格、库存，四个高频接口的正确调用姿势》讲了怎么调；这篇讲接入时怎么不翻车。*
