# 商品详情、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 小时都够。**关键是分级这件事本身，不是每级的具体数字。**