# 鉴权、签名、字段映射：三平台对接最容易翻车的三个环节（附 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、价格、库存，四个高频接口的正确调用姿势》讲了怎么调；这篇讲接入时怎么不翻车。*

# 商品详情、SKU、价格、库存，四个高频接口的正确调用姿势（附 Python 源码）

先给一个数字：这四个接口，占我们平台**总调用量的 85%**。

这个占比意味着两件事。第一，任何调用姿势上的坏习惯，都会被放大 85%——配额浪费、数据错位、深夜告警，基本都出在这四个接口上。第二，优化它们的收益也是 85% 的，比优化任何冷门接口都值。

这篇是手册型文章，和第 11 篇的对照表配套使用：那篇讲三个平台"有什么不同"，这篇讲"每个接口该怎么调"。四个接口各一份错误姿势清单，最后一个总纲。

---

## 〇、总纲：四个接口不是四件事，是一棵树

最常见的错误，是把它们当成四个平级接口、用同一个轮询周期去调。

但它们有严格的依赖关系：

```
商品详情（根）
   └── SKU 集合（挂在商品上）
          ├── 价格（挂在 SKU 上）
          └── 库存（挂在 SKU 上）
```

这个依赖关系推出两条铁律：

**铁律一：按变更频率分级调度。** 详情一天变不了几次，库存每秒都在变。捆在一个轮询周期里，要么详情白调（浪费配额），要么库存延迟（业务事故）。

**铁律二：能用一次详情带出的，不要拆四次请求。** 三个平台的详情接口几乎都会返回 SKU 列表和标价——一次请求拿三样东西，这是最大的配额节省。真正需要单独调的只有库存（它要实时）。

| 数据       | 变更频率 | 来源          | 同步方式          |
| -------- | ---- | ----------- | ------------- |
| 标题、主图、类目 | 天级   | 详情接口        | 每日增量          |
| SKU 集合   | 小时级  | 详情接口内       | 集合 diff       |
| 价格       | 分钟级  | 详情 / 价格接口   | 定时 + 改价回执     |
| 库存       | 秒级   | 库存接口 / 消息推送 | **事件优先，轮询兜底** |

下面四个接口的"正确姿势"，全部从这两条铁律推出来。

---

## 一、商品详情：它是根，不是流水账

详情接口的错误姿势，集中在"把它当唯一的真相来源"。

| 错误姿势             | 后果                     | 正确姿势                 |
| ---------------- | ---------------------- | -------------------- |
| 全量轮询所有商品详情       | 配额爆炸，且 90% 请求拿回的是没变的数据 | 列表接口增量筛出变更 ID，再按需调详情 |
| 不传 `fields`，全字段拉 | 响应体积差 10 倍，解析和带宽都翻倍    | 按需声明字段，用不到的不拉        |
| 把详情里的价格/库存当业务数据源 | 数据滞后几分钟到几小时            | 详情只管"慢字段"，快字段走分级调度   |
| 下架商品按正常结构解析      | 静默拿到空数据或 KeyError      | 先判状态位，软删除单独处理        |

四条里最贵的是第二条。很多人不知道详情接口支持字段裁剪，把整个响应体解析一遍，结果：**每次多传 80% 用不到的字节，配额按调用次数算倒是不亏，但带宽、解析时间、落库量全都翻倍**。字段白名单就一行的事：

```python
DETAIL_FIELDS = [
    "num_iid", "title", "pic_url", "cid",
    "skus", "price",           # SKU 集合和标价跟着详情一起拿（铁律二）
    "list_time", "delist_time",
]

params = {"num_iid": item_id, "fields": ",".join(DETAIL_FIELDS)}
```

第四条单独说一句：**下架商品在不同平台的表现不同**——有的返回错误码，有的返回成功但核心字段为空，有的直接不返回。用第 12 篇的 `integrity.validate` 兜底，加上一个状态位判断，别让下架商品在你的库里"看起来还活着"。

---

## 二、SKU：它是会消失的

SKU 是四个接口里坑最深的，因为它的坑不在"怎么调"，在"调回来之后怎么管"。

| 错误姿势                      | 后果                | 正确姿势                               |
| ------------------------- | ----------------- | ---------------------------------- |
| 只拿 `sku_id` 当主键           | 多店铺/多平台必然撞键       | 主键 = `(platform, item_id, sku_id)` |
| 覆盖式更新 SKU 表               | 商家删掉的 SKU 永远留在你库里 | 集合 diff，消失的标记 inactive             |
| `props.split(":")` 硬拆销售属性 | 属性名里含冒号就炸         | 结构化解析（第 2 篇的 `_parse_props`）       |
| 假设每个商品都有 SKU              | 单规格商品返回空          | 先判商品规格类型                           |

第二条是重灾区。商家改规格是日常操作：把"红色 L"删了换成"红色 XL"。如果你每次拉到 SKU 列表就整表覆盖，或者只往上插入新 SKU——**被删掉的那个 SKU 会永远留在你库里，价格库存还挂着旧数据**。客户拿它下单，就是一次客诉。

正确做法是集合 diff：

```python
# sku_diff.py
def sku_diff(old_ids: set[str], new_ids: set[str]) -> tuple[set[str], set[str]]:
    """
    返回 (新增, 消失)。
    消失的 SKU 千万别物理删除——历史订单还引用它。
    标记 inactive，保留可查。
    """
    return new_ids - old_ids, old_ids - new_ids


async def sync_skus(store, item_id: str, new_skus: list[dict]):
    old_ids = set(store.sku_ids(item_id))
    new_ids = {s["sku_id"] for s in new_skus}

    added, removed = sku_diff(old_ids, new_ids)

    for s in new_skus:
        if s["sku_id"] in added:
            store.insert_sku(item_id, s)
        else:
            store.update_sku(item_id, s)          # 属性或价格变了

    for sku_id in removed:
        store.mark_inactive(item_id, sku_id)      # 软删，订单引用不断
```

另外主键那条：`sku_id` 只在单个商品内唯一。做三平台、多店铺的库，主键从第一天就是三元组，**后期改主键的痛苦是接入期的十倍**（第 11 篇说 1688 连 SKU 都没有，那它的"规格"要单独建表，别硬凑）。

---

## 三、价格：它不是一个字段，是一组字段

价格的问题出在"你以为价格是一个数"。

| 错误姿势       | 后果          | 正确姿势                   |
| ---------- | ----------- | ---------------------- |
| 只取一个价格字段   | 拿到标价，不是成交价  | 价格组归一，算"生效价"           |
| float 存价格  | 账对不上，差几分钱   | `Decimal`，最后一步才转 float |
| 改价后立刻读回来验证 | 读到旧缓存，以为没改成 | 接受几分钟延迟，或以改价接口回执为准     |
| 1688 只取一个价 | 阶梯价拿错档位     | 按起订量取档                 |
| 促销价无限期生效   | 过期促销价当现价    | 促销价必须带时效字段             |

第一和第五条组合起来，是最经典的线上事故：**拿到促销价 → 没记过期时间 → 促销结束后继续按促销价计算**。利润表和实际对不上，查一周。

正确建模是把价格当成一个带时效的小对象：

```python
# price_model.py
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from zoneinfo import ZoneInfo

TZ = ZoneInfo("Asia/Shanghai")


@dataclass
class Price:
    sku_id: str
    list_price: Decimal                    # 标价 / 一口价
    promo_price: Decimal | None = None     # 促销价
    promo_end: datetime | None = None      # 促销截止时间

    def effective(self, now: datetime | None = None) -> Decimal:
        """生效价：促销在有效期内才生效，过期自动回落标价"""
        now = now or datetime.now(TZ)
        if self.promo_price is not None:
            if self.promo_end is None or self.promo_end > now:
                return self.promo_price
        return self.list_price
```

1688 的价格是另一个物种——阶梯价，按起订量分档：

```python
def alibaba_price_tiers(raw: list[dict]) -> list[tuple[int, Decimal]]:
    """
    1688 priceRange：[{beginAmount: 2, price: "8.50"}, {beginAmount: 100, price: "7.20"}]
    按起订量升序排列，采购 N 件时取满足 N 的最后一档。
    """
    tiers = sorted(
        ((int(x["beginAmount"]), Decimal(x["price"])) for x in raw),
        key=lambda t: t[0],
    )
    return tiers

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def tier_for(tiers: list[tuple[int, Decimal]], qty: int) -> Decimal:
    price = tiers[0][1]
    for begin, p in tiers:
        if qty >= begin:
            price = p
    return price
```

还有改价那条：**改价接口返回成功 ≠ 价格已生效**。平台侧有缓存延迟，短则几秒长则几分钟。以改价接口的响应做业务判断，别改完立刻读回来验证——你会拿到旧价格，然后误判"改价失败"再改一次。

---

## 四、库存：只有"可售库存"有意义

库存的问题在于：平台给你好几个数，你用了错的那个。

| 错误姿势         | 后果           | 正确姿势             |
| ------------ | ------------ | ---------------- |
| 用"总库存"当可卖数量  | 超卖（锁定的库存又卖了） | 用**可售库存**字段      |
| 平台库存直接覆盖本地库存 | 抹掉本地扣减，二次超卖  | 本地账本 + 校准（第 3 篇） |
| 和详情同频率轮询     | 延迟大还烧配额      | 事件推送优先，轮询兜底      |
| 库存为 0 就当售罄   | 可能是下架或区域性无货  | 状态位 + 库存一起判断     |
| 硬凑 1688 的库存  | 它没有 SKU 库存概念 | 可售状态 + 阶梯价建模     |

第三条是这篇的落地重点。库存是唯一值得上**事件推送**的字段——淘宝的 TMC 有库存变更消息，订阅它，库存事件毫秒级到达，配额零消耗。轮询只做兜底（比如每 5 分钟对账一次，防丢消息）：

```python
# stock_sync.py —— 事件优先，轮询兜底
async def on_stock_event(msg: dict):
    """TMC 库存变更消息：毫秒级，零配额"""
    await apply_stock(msg["item_id"], msg["sku_id"], msg["quantity"])

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def reconcile_loop(gw, store, item_ids: list[str], interval: float = 300):
    """兜底对账：防消息丢失。只查本地 5 分钟内有订单变动的商品，别全量"""
    while True:
        hot = store.items_with_recent_orders(minutes=5)
        for iid in hot:
            raw = await gw.call("stock.get", item_id=iid)
            local = store.local_stock(iid)
            # 校准取 min：平台还没反映本地刚扣的量（第 3 篇的规则）
            safe = min(int(raw["saleable"]), local)
            if store.remote_stock(iid) != safe:
                store.update_remote_stock(iid, safe)
        await asyncio.sleep(interval)
```

注意兜底循环里那句 `hot = items_with_recent_orders(minutes=5)`：**只对账"最近有单"的商品**。全量对账是最容易写出来的版本，也是把库存轮询成本放大一百倍的版本——而真正需要对账的（刚发生扣减、可能不一致的）永远只占极小一部分。

---

## 五、把四个接口串起来：分级调度器

最后给一个可跑的调度骨架。它的职责只有一件事：**保证每个字段按自己的频率被同步，不多不少**。

```python
# scheduler.py
import asyncio
import time

LEVELS = {"detail": 3600, "sku": 3600, "price": 300, "stock": 30}


class DueScheduler:
    def __init__(self, r):
        self.r = r

    async def due(self, item_id: str, field: str) -> bool:
        last = await self.r.get(f"due:{field}:{item_id}")
        if last is None:
            return True                      # 从没同步过，先来一次
        return time.time() - float(last) >= LEVELS[field]

    async def mark(self, item_id: str, field: str):
        await self.r.set(f"due:{field}:{item_id}", time.time())


async def sync_item(gw, sched: DueScheduler, store, item_id: str):
    # ---- 1) 详情（小时级）：一次拉全 慢字段 + SKU 集合 + 标价 ----
    if await sched.due(item_id, "detail"):
        d = await gw.call("item.get", item_id=item_id,
                          fields=",".join(DETAIL_FIELDS))
        store.upsert_item(item_id, d)

        new_skus = d.get("skus", [])
        await sync_skus(store, item_id, new_skus)     # 内部做集合 diff
        await sched.mark(item_id, "sku")

        await sched.mark(item_id, "detail")

    # ---- 2) 价格（分钟级）：从详情缓存或价格接口取，归一成 Price 对象 ----
    if await sched.due(item_id, "price"):
        raw = await gw.call("price.get", item_id=item_id)
        for sku_price in normalize_prices(item_id, raw):
            store.save_price(sku_price)               # effective() 按时效算生效价
        await sched.mark(item_id, "price")

    # ---- 3) 库存（秒级）：事件推送是主通道，这里只是兜底标记 ----
    # 真正的兜底对账在 reconcile_loop，只扫最近有单的商品


async def main(gw, sched, store, item_ids):
    # 详情和价格按各自的频率到期才调；库存靠事件
    while True:
        for iid in item_ids:
            await sync_item(gw, sched, store, iid)
        await asyncio.sleep(30)     # 最细粒度的检查周期，但不是最细粒度的调用频率
```

跑起来的效果：同一个商品，详情接口一天调 24 次、价格 288 次、库存 0 次（走事件）——**对比"四个接口都 30 秒一轮"的写法，配额消耗约降到 1/300**，而库存的实时性反而更好（事件是毫秒级的）。

---

## 六、四个反直觉的结论

**1. 详情接口调得越勤，数据越不准。**  
因为详情是慢字段的来源，高频调它不会让标题更新更快（它一天就变一次），只会让你的价格库存看起来"应该也是新的"——然后掩盖它们其实滞后的事实。分开调度，延迟才藏不住。

**2. 消失的 SKU 比新增的危险。**  
新增最多是数据多一条；消失的 SKU 还挂着旧价格旧库存，客户能下单。覆盖式更新是这里的万恶之源。

**3. 促销价必须带时效，这是建模问题不是字段问题。**  
没有 `promo_end` 字段的促销价，就是一颗定时炸弹——促销结束那天准时爆。

**4. 库存的正确姿势是"少调"，不是"勤调"。**  
事件推送零配额、毫秒级；轮询全量、高延迟。轮询不是兜不了底，是兜底时只扫"最近有单"的商品——全量对账是库存成本失控的第一原因。

---

## 七、最后

这篇没有新架构，讲的全是**已有架构里每个接口的使用纪律**。

回看整个系列：第 8 篇给了骨架（统一接口层），第 9、10 篇给了运行时（稳定性、限流），第 12 篇给了选型（为什么用 API），第 11 篇给了差异（平台之间差在哪）。这篇补的是最后一层——**每个高频接口的调用纪律**。骨架、运行时、选型、差异、纪律，五层齐了。

下一篇写「多店铺授权管理」：token 的存储与加密、并发刷新撞车、单店授权过期的隔离。那是接入体系的最后一个坑区，写完这个系列就真正完整了。

`LEVELS` 里的数字是起点不是标准：详情 1 小时、价格 5 分钟、库存 30 秒，都是按"多数电商场景"取的。你们的业务如果价格敏感度高（比如比价类），价格调到 60 秒；如果纯铺货场景，详情调到 6 小时都够。**关键是分级这件事本身，不是每级的具体数字。**