# 刚起步的电商小团队，第一批该接哪几个接口（附 Python 源码）

> 起步团队最容易犯的错，是把「接口清单」当成任务清单，一条条往下接。
>
> 三平台文档里有上千个接口，你真正需要的第一批只有 **4 个**。剩下的不是不重要，是**现在接它们会拖慢你赚到第一笔钱的时间**。
>
> 这篇给一个明确的顺序：第一批 4 个、第二批 3 个、第三批 4 个，以及三个「现在千万别接」的。判断标准只有一条——**这个接口能不能让货卖出去、或者让钱收回来**。
>
> 代码是一个 150 行的单文件，跑完就是一条完整闭环：商品上架 → 客户下单 → 订单进库 → 扣库存 → 日终对账。

## 一、先定一个判断标准

接口的价值不是「技术上重要」，是**它对收入的贡献**。按这个标准，起步期能碰的接口只分三类：

| 类型      | 作用        | 起步期优先级     |
| ------- | --------- | ---------- |
| **换收入** | 商品能上架、能卖  | 最高，第一批     |
| **保现金** | 订单不漏、账对得上 | 最高，第一批     |
| **补体验** | 物流可查、售后顺畅 | 第二批，等有单了再说 |
| **纯优化** | 评价、竞品、素材  | 第三批，或者永远不接 |

一个反例：很多团队第一批就去接「物流轨迹查询」，理由是「客户会问快递到哪了」。但起步期你一天可能就 5 单，客户问了你手动查一下，30 秒的事。**为一天 5 次的查询，写一套订阅 + 去重 + 状态机，是典型的提前优化**（第 3 篇那套方案是对的，但不是现在该做的）。

## 二、第一批：4 个接口，按这个顺序接

| 顺序 | 接口                      | 为什么在这个位置                                |
| -- | ----------------------- | --------------------------------------- |
| ①  | **商品详情**（含 SKU、标价、可售库存） | 没商品就没得卖。三平台的详情接口都一次返回 SKU 列表和标价，别拆成四次请求 |
| ②  | **订单列表**（按更新时间增量）       | 收入的证据。也是「有没有单」的唯一可靠来源                   |
| ③  | **订单详情**（含收货地址、支付时间）    | 发货必需。**列表接口通常不含完整地址**，这个坑第一天就会撞上        |
| ④  | **库存**（可售库存）            | 防超卖。但起步期用「简单账本」撑住就行，别做实时同步              |

### 为什么「先订单、后库存」

这是这篇最反直觉的一条，值得单独说。

按技术难度排，库存比订单简单（一个数字 vs 一堆状态）。但按**收益**排，订单在前：

- **订单**接进来，你立刻能发货、能算账、能知道今天赚了多少。这是正反馈。
- **库存**接进来，你防的是「超卖」。而起步期 SKU 少、单量小，超卖的实际损失远小于「商品还没上架」的损失。

所以正确的做法是：**库存先用静态值撑住，下单即扣本地账本，宁可少卖**。具体就是三行逻辑：

```
可卖数 = 平台可售 − 本地已占      （查询时算，不写库）
下单   → 够就 reserved += n，不够就拦住这笔单
日终   → 平台库存重新拉过之后，reserved 归零重算
```

这么做会偶尔少卖（账面比实际紧），但不会超卖，而且**没有并发问题**——因为你从来不做「实时同步」这件难事。

等单量上来了（比如日单过百，或者开始做多仓），再把第 3 篇那套 `total - reserved - safety` 加上事件校准换上。那时候你已经有收入了，花两周做对它是合理的。

反过来做的团队我见过不少：花两周把库存并发扣减做对，商品还没上架，一分钱没赚。**技术正确，商业上错了。**

## 三、第二批：有第一笔订单之后

大概在第一笔订单之后的 2~4 周，接这三个：

| 接口          | 触发条件                  | 不接的代价                  |
| ----------- | --------------------- | ---------------------- |
| **物流轨迹**    | 客户开始问「到哪了」，且每天超过 10 次 | 客服时间被吃掉，但还不致命          |
| **售后 / 退款** | 出现第一笔退款               | **账永远对不上**，你会以为是接口有问题  |
| **发货回写**    | 你做代发，或对接了 ERP         | 只能在平台后台手工点发货，一天几十单就撑不住 |

退款这个特别值得说：起步团队常常只拉「已付款」的订单，退款不管。结果月底一算，GMV 和实际到账差一截，财务来问，你查三天。

**退款不是异常，是正常业务**。它该在第二批，不该在「以后再说」。

## 四、第三批：月订单过千再看

| 接口        | 什么时候值得                            |
| --------- | --------------------------------- |
| 活动价 / 优惠券 | 你开始做促销，且发现「到手价」算不对（第 13 篇的价格对象建模） |
| 类目属性      | 你开始批量铺货、被审核驳回（第 5 篇）              |
| 评价        | 你要做选品或口碑分析（第 16 篇）                |
| 素材 / 图片   | 你要做自动生成主图、多平台分发                   |

这四个都是「业务推着你接」，不是「文档里有所以接」。

## 五、三个现在千万别接的

**① TMC 消息推送。**

听起来很美：订单来了实时推给你。实际要维护长连接、断线重连、消息确认、重复消费——四件事都能坏，坏了都要你查。起步期用「5 分钟轮询 + 时间窗重叠」完全够，而且失败就是这一轮没跑，下一轮补上，心智负担接近于零。

等你日单过千、或者对时效真的有要求了再换。切换成本是一次可控的重构，不是重写。

**② 批量 / 异步接口。**

批量接口的特点是：提交任务 → 拿 task_id → 轮询结果 → 处理部分失败。这套东西要状态机、要重试、要存储中间态。起步期你的商品可能就 200 个，循环调 200 次详情，10 分钟跑完，够用。

**③ 全站 / 竞品数据。**

先把自己店的货卖好。比价、竞品监控是第二阶段的事（第 16 篇），而且它需要的数据源和自营接口是两套逻辑（联盟接口 / 聚合服务），混在一起做会让你的第一版架构变形。

## 六、代码：150 行，一条闭环

起步版的原则：**小到一屏能看懂**。不要配置层、不要抽象层、不要依赖注入——这些东西等你有了第二个平台再谈。

```python
# mvp.py —— 起步版：4 个接口，1 条闭环
# 依赖：pip install httpx
import asyncio
import sqlite3
import time

import httpx

BASE = "https://api.example.com"          # 换成你的网关
KEY = "your_app_key"
SECRET = "your_app_secret"
SHOP = "shop_001"

SCHEMA = """
pragma journal_mode = wal;
create table if not exists items(
  item_id text primary key, title text, price real,
  stock int, skus text, updated_at int);
create table if not exists orders(
  tid text primary key, status text, pay_time int,
  amount real, receiver text, updated_at int);
create table if not exists ledger(
  item_id text primary key, reserved int default 0, last_sync int);
"""

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def call(client, path, **params):
    """一个函数搞定所有接口：拼参数、判错、返回 data"""
    r = await client.get(path, params={"shop_id": SHOP, **params})
    r.raise_for_status()
    body = r.json()
    if body.get("code") not in (0, 200, "0", None):
        raise RuntimeError(f"{body.get('code')} {body.get('msg')}")
    return body.get("data", body)


async def pull_items(client, db):
    """① 商品详情：一次带回 SKU、标价、可售库存。
    三平台详情接口都支持，别拆成 SKU 接口 + 价格接口 + 库存接口"""
    ids = [r[0] for r in db.execute("select item_id from items")]
    if not ids:                       # 第一次：从列表接口拿 ID
        lst = await call(client, "/v1/item/list", page_size=100)
        ids = [it["item_id"] for it in lst.get("items", [])]

    for item_id in ids:
        d = await call(client, "/v1/item", item_id=item_id)
        skus = d.get("skus") or []
        stock = sum(int(s.get("stock") or 0) for s in skus) if skus \
            else int(d.get("stock") or 0)
        db.execute(
            "insert into items(item_id,title,price,stock,skus,updated_at) "
            "values(?,?,?,?,?,?) on conflict(item_id) do update set "
            "title=excluded.title, price=excluded.price, stock=excluded.stock, "
            "skus=excluded.skus, updated_at=excluded.updated_at",
            (item_id, d.get("title", ""), float(d.get("price") or 0),
             stock, str(len(skus)), int(time.time())))
    db.commit()
    return len(ids)


async def pull_orders(client, db, since: int):
    """② 订单列表：按更新时间增量，重叠加窗（第 3 篇的边界坑）"""
    start = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(since - 600))
    page, got = 1, 0
    while True:
        lst = await call(client, "/v1/orders",
                         updated_start=start, page=page, page_size=100)
        rows = lst.get("orders", [])
        if not rows:
            break
        for o in rows:
            db.execute(
                "insert into orders(tid,status,pay_time,amount,updated_at) "
                "values(?,?,?,?,?) on conflict(tid) do update set "
                "status=excluded.status, updated_at=excluded.updated_at",
                (o["tid"], o["status"], int(o.get("pay_time") or 0),
                 float(o.get("payment") or 0), int(time.time())))
            got += 1
        page += 1
    db.commit()
    return got


async def fill_receiver(client, db):
    """③ 订单详情：列表接口没有地址，发货前补。
    只补「待发货」的，别全量拉——这一步最容易把额度烧光"""
    todo = db.execute(
        "select tid from orders where receiver is null and status='WAIT_SELLER_SEND'"
    ).fetchall()
    for (tid,) in todo:
        d = await call(client, "/v1/order", tid=tid)
        addr = d.get("receiver") or {}
        db.execute("update orders set receiver=? where tid=?",
                   (f"{addr.get('name','')}|{addr.get('mobile','')}|"
                    f"{addr.get('address','')}", tid))
    db.commit()
    return len(todo)


def available(db, item_id: str) -> int:
    """④ 可卖数 = 平台可售 − 本地已占。
    查询时算，不写库：`items.stock` 是平台镜像，每次拉取覆盖；
    本地占用量单独记在 ledger，两者相减才是能卖的数。
    不写回的好处是不存在「谁覆盖谁」的问题——第 3 篇那个
    「校准取 min」的原则，是给「要把库存写回平台」的场景用的，
    起步期我们不写回，所以用不上。"""
    item = db.execute("select stock from items where item_id=?",
                      (item_id,)).fetchone()
    if item is None:
        return 0
    row = db.execute("select reserved from ledger where item_id=?",
                     (item_id,)).fetchone()
    return max(0, item[0] - (row[0] if row else 0))


def reserve_stock(db, item_id: str, num: int = 1) -> bool:
    """下单占库存：够就扣，不够就拦住这笔单（宁可少卖，不超卖）。
    注意先判后扣——先扣再判，被拦下的单也会留下占用，账面越推越紧。"""
    if available(db, item_id) < num:
        return False

    row = db.execute("select reserved from ledger where item_id=?",
                     (item_id,)).fetchone()
    if row is None:
        db.execute("insert into ledger(item_id,reserved,last_sync) "
                   "values(?,?,?)", (item_id, num, int(time.time())))
    else:
        db.execute("update ledger set reserved=reserved+?, last_sync=? "
                   "where item_id=?", (num, int(time.time()), item_id))
    db.commit()
    return True


def reset_reserved(db):
    """日终：平台库存已经在 pull_items 里重新拉过，本地占用量归零重算。
    必须在拉取成功之后调，否则会把当天的占用抹掉——那就是超卖。"""
    n = db.execute("update ledger set reserved=0 where reserved>0").rowcount
    db.commit()
    return n


async def main():
    db = sqlite3.connect("mvp.db")
    db.executescript(SCHEMA)
    async with httpx.AsyncClient(base_url=BASE, timeout=10,
                                 headers={"X-Secret": SECRET, "X-Key": KEY}) as c:
        n_items = await pull_items(c, db)          # 先拉最新库存
        n_orders = await pull_orders(c, db, int(time.time()) - 3600)
        n_addr = await fill_receiver(c, db)
        n_reset = reset_reserved(db)               # 拉成功之后才清零
    print(f"商品 {n_items} · 订单 {n_orders} · 补地址 {n_addr} · 归零 {n_reset}")


if __name__ == "__main__":
    asyncio.run(main())
```

这 150 行就是起步版的全部。它故意没有的东西，是按重要性排的：

| 没有        | 为什么现在不加             | 什么时候加            |
| --------- | ------------------- | ---------------- |
| 配置层（YAML） | 5 个常量改起来比读配置快       | 有第二个部署环境         |
| 多平台适配器    | 先接一个平台，等第二个出现重复代码   | 第二个平台接的时候（第 8 篇） |
| 并发 / 限流   | 200 个商品串行跑 4 分钟，能接受 | 商品上千，或单量上来       |
| 重试 / 熔断   | 失败就下一轮再跑，起步期能忍      | 开始有客户依赖你的数据      |
| 消息推送      | 轮询 5 分钟够用           | 日单过千             |

## 七、验收标准：一条能跑完的闭环

「接口接完了」不等于「能用了」。第一批的正确验收标准是一个动作：**把你自己的一个商品，从头到尾跑一遍**。

1. 商品信息从接口拉下来，落到你的库，页面能展示
2. 用另一个账号在你店里下一单（1 元商品，自己买）
3. 轮询把订单拉进来，地址补全，状态是「待发货」
4. 本地库存被扣掉 1，页面显示的可售数减 1
5. 你在平台后台点发货，回写状态（第二批才有，现在手工）

这五步跑通，第一批就算完成。**注意第 2 步必须真下单**——只用接口返回的模拟数据，你永远不知道字段口径对不对。1 元的成本，换来的是「整条链路真的通」，这是起步期最划算的一笔投入。

跑完你会发现至少两个问题：某个字段为空（通常是权限没开）、某个时间格式不对（时区）。这两个坑每个团队都会踩，早踩比上线后踩好。

## 八、起步期三个最常见的错误

**① 三个平台一起接。**

看起来平行推进效率高，实际是把「三个平台各自的怪癖」同时砸在你头上。第 2 篇那个坑表里的东西——1688 没有 SKU 库存、京东返回体多套一层、淘宝错误藏在 body 里——同时遇到三个，你会怀疑人生。

正确顺序：**一个平台跑通闭环 → 再复制到第二个**。第二个平台的成本大约是第一个的 1/3，因为业务逻辑已经验证过了。

**② 一开始就做增量和高并发。**

起步期商品 200 个、订单一天 20 个，全量拉一遍 4 分钟。这个阶段做增量调度、做并发优化，是把第 7 篇那套 60 倍提速用在 4 分钟上——**优化一个你根本感知不到的问题**。

**③ 一开始就自建。**

第 12 篇算过账：自建爬虫三年 37 万，官方 API 17 万。起步团队还有一笔更贵的隐藏成本：**你在写同步代码的那两周，没有在卖货**。

起步期最该花钱买的就是数据接口，因为它把你从「修管道」里解放出来，去做只有你能做的事——选品、内容、投放。这笔账在收入为零的时候最容易被算错。

## 九、时间预算

按这篇的顺序做，第一周（约 20 小时）应该能到：

| 天     | 内容               | 小时 |
| ----- | ---------------- | -- |
| D1    | 申请、拿 key、跑通商品详情  | 4  |
| D2    | 商品落库 + 页面展示      | 4  |
| D3-D4 | 订单增量 + 地址补全 + 幂等 | 8  |
| D5    | 库存账本 + 日终校准      | 4  |

对比「全都接」的版本：三平台 + 物流 + 售后 + 推送，两个月起步，而且大概率在第二个月放弃。

**起步团队真正的资源不是技术能力，是「还没被复杂度拖住」的那段时间。**

---

这一篇放在系列里的位置比较特殊：前面二十一篇讲的是「怎么把系统做对」，这篇讲「什么时候该做多少」。如果你现在只有一个店、一天十几单，上面 150 行代码就够了——第 3 篇的库存账本、第 7 篇的并发、第 9 篇的五道关，都等你需要的时候再看，那时候你会看得更懂。

# 代运营公司怎么管 20 个店铺的数据？我们是这样做的（附 Python 源码）

> 代运营的数据问题和卖家不一样。
>
> 卖家的问题只有一个店：取到数据、别出错、别断。代运营的问题是一堆店：**授权会过期、数据不能串、客户要各自的报表、一个店出问题不能影响其他十九个**。
>
> 我们踩过的坑基本都在这四件事上，跟接口本身关系不大。20 个店铺、47 个有效授权（有的店只在两个平台开店）、三个平台三套授权体系——这套东西现在是五个文件的代码在跑，人工每天 10 分钟看一张汇总表。
>
> 这篇把授权、隔离、汇报三块拆开讲，代码可以直接拿去改。

## 一、先把「店铺」当成一级实体

大部分多店铺系统出问题，根子在建模：**把店铺当成了配置，而不是资源**。

配置是没有生命周期的——写进去就不管了。但一个店铺授权是有生命周期的：客户签进来 → 授权 → 用着 → 快到期 → 客户换人/改密码 → 授权失效 → 客户走了 → 撤权。**每一步都有状态，每一步都需要代码处理**。

正确的建模是两层：

```
店铺（shop）
  └── 授权（shop_id + platform）  ← 唯一的凭证单元
```

20 家店 × 3 平台，不是 20 条记录，是 47 个授权（有 7 家店只在两个平台开店）。凭证表的主键必须是 `(shop_id, platform)`，不能是 `shop_id`——**否则一家店在三个平台的 token 只能存一个，后写的覆盖先写的，你会以为授权好了，其实只好了最后一个平台**。这个 bug 我们见过不止一次。

授权有五个状态，别用布尔值糊过去：

| 状态         | 含义            | 系统该做什么       |
| ---------- | ------------- | ------------ |
| `pending`  | 已建店，授权还没完成    | 催授权，不进调度队列   |
| `active`   | 正常            | 正常调度         |
| `expiring` | 剩余 < 48 小时    | 提前预警，任务照跑    |
| `expired`  | 已过期           | 停止调度，标记客户待处理 |
| `revoked`  | 客户主动解绑 / 平台收回 | 停止调度，通知商务    |

`expiring` 这个状态是最值钱的一个。**代运营每月有 3~5 个授权到期**，等过期了再处理，意味着那家店的日报当天就断了，客户第二天早上会来问。提前 48 小时预警，实际处理时间是两天，从容得多。

## 二、token 存储：加密与否，是责任问题

47 个店铺的 refresh token 躺在数据库里，这就是 47 张店铺的钥匙。数据库被拖、备份文件进了 git、开发同学的笔记本被偷——任何一个环节泄露，客户的全店数据就没了。

**这不是「应该加密」，是「必须加密」**，而且要在写代码的第一天做，后面补的成本高得多（要迁移存量、要处理轮换）。

```python
# ops/vault.py
import base64
import os

from cryptography.hazmat.primitives.ciphers.aead import AESGCM


class Vault:
    """token 加解密：AES-GCM，密文入库，密钥永远不进数据库。

    keys 是 {key_id: 32字节密钥}，current 是当前用来加密的 key_id。
    存 key_id 的目的只有一个：轮换。换密钥时老密文还能用老 key 解开，
    解密成功的记录顺手用新 key 重新加密，慢慢滚过去。
    """

    def __init__(self, keys: dict[str, bytes], current: str):
        self.keys = keys
        self.current = current
        for k in keys.values():
            if len(k) != 32:
                raise ValueError("AES-256 需要 32 字节密钥")

    def seal(self, plaintext: str) -> tuple[str, str, str]:
        """返回 (密文, nonce, key_id)，三列一起入库"""
        nonce = os.urandom(12)
        ct = AESGCM(self.keys[self.current]).encrypt(
            nonce, plaintext.encode("utf-8"), None)
        return (base64.b64encode(ct).decode(),
                base64.b64encode(nonce).decode(),
                self.current)

    def open(self, ct_b64: str, nonce_b64: str, key_id: str) -> str:
        return AESGCM(self.keys[key_id]).decrypt(
            base64.b64decode(nonce_b64),
            base64.b64decode(ct_b64),
            None).decode("utf-8")
```

密钥从哪来？三个档，按你的规模选：

| 方式             | 适合         | 注意                          |
| -------------- | ---------- | --------------------------- |
| 环境变量           | 小团队、单机     | 进程能读到，宿主机被拿就没了；至少别写进代码和 git |
| 云 KMS / 密钥管理服务 | 十家店以上、多机部署 | 每次解密一次网络调用，要配本地短缓存（5 分钟）    |
| 硬件加密机          | 有等保要求      | 代运营基本用不上，成本不划算              |

我们用的是云 KMS + 进程内 5 分钟缓存。缓存的是**解密后的 token**，不是密钥——这个区别很重要，密钥永不落地。

**一个容易漏的地方**：`refresh_token` 也要加密，而且它比 `access_token` 更重要。`access_token` 两小时就过期了，`refresh_token` 能换出无数个新的。只加密前者不加密后者，等于没加密。

## 三、并发刷新撞车：多店铺场景的第一次翻车

单个店铺的时候你不会遇到这个问题。20 个店铺、多个任务并发跑的时候，它一定会来。

场景长这样：某个店的 token 在 10:00:00 过期。10:00:01，订单同步任务和商品同步任务同时发现过期，同时发起刷新。平台的处理是**后刷作废先刷**（京东和 1688 都有这个行为），于是：

- 任务 A 拿到的 token 已经作废了 → 401
- 任务 B 拿到的 token 有效 → 继续跑
- 更糟的情况：两个都拿到「刷新成功」，但只有一个真正有效，另一个在下一次调用时才暴露

结果就是「同一个店，一半任务成功一半失败，重试之后好了」——这种故障最难查，因为它不稳定复现。

解法是**单飞（single-flight）**：同一个 `(shop_id, platform)` 在同一时刻只能有一个刷新在跑，其他协程等它的结果。

```python
# ops/refresher.py
import time

# 抢锁靠这一行 update：只有拿到了刷新权的那个协程，rowcount 才是 1
CLAIM = """
update grants
   set locked_until = :until
 where shop_id = :shop_id and platform = :platform
   and (locked_until is null or locked_until < :now)
"""


def ensure_token(db, vault: Vault, shop_id: str, platform: str,
                 refresh_fn, ahead: int = 300) -> str:
    """返回可用的 access_token。

    ahead：提前多少秒视为「需要刷新」。别卡着过期点刷，
    网络一来一回加上平台处理，卡点必翻车。
    """
    for _ in range(3):  # 最多等三轮，够覆盖一次正常刷新
        row = db.execute(
            "select access_token, nonce, key_id, refresh_token, "
            "refresh_nonce, refresh_key_id, expire_at, state "
            "from grants where shop_id=? and platform=?",
            (shop_id, platform)).fetchone()
        if row is None:
            raise LookupError(f"未找到授权 {shop_id}/{platform}")

        (ct, nonce, key_id, rct, rnonce, rkey_id, expire_at, state) = row
        if state in ("revoked", "expired"):
            raise PermissionError(f"授权不可用 {shop_id}/{platform} state={state}")

        if expire_at > time.time() + ahead:
            return vault.open(ct, nonce, key_id)

        now = int(time.time())
        cur = db.execute(CLAIM, {"until": now + 30, "shop_id": shop_id,
                                 "platform": platform, "now": now})
        db.commit()

        if cur.rowcount == 0:
            # 别人正在刷，等一下再读它刷好的结果
            time.sleep(0.3)
            continue

        try:
            plain_rt = vault.open(rct, rnonce, rkey_id)
            new_token, expires_in, new_rt = refresh_fn(plain_rt)

            n_ct, n_nonce, n_kid = vault.seal(new_token)
            r_ct, r_nonce, r_kid = vault.seal(new_rt or plain_rt)
            db.execute(
                "update grants set access_token=?, nonce=?, key_id=?, "
                "refresh_token=?, refresh_nonce=?, refresh_key_id=?, "
                "expire_at=?, state='active', locked_until=null, "
                "updated_at=? where shop_id=? and platform=?",
                (n_ct, n_nonce, n_kid, r_ct, r_nonce, r_kid,
                 int(time.time()) + expires_in, now, shop_id, platform))
            db.commit()
            return new_token

        except Exception as e:
            # 刷新失败：不清旧 token，只标记状态。旧的可能还能用一会儿，
            # 清掉就是立刻断服——代运营断一家店，客户当天就知道。
            db.execute(
                "update grants set locked_until=null, last_error=?, "
                "state=case when expire_at < ? then 'expired' else state end "
                "where shop_id=? and platform=?",
                (repr(e)[:300], int(time.time()), shop_id, platform))
            db.commit()
            raise
        finally:
            # 释放锁。用条件更新而不是直接置 null：
            # 万一本进程卡了很久、锁已被别人重新抢到，别把别人的锁解掉
            db.execute(
                "update grants set locked_until=null "
                "where shop_id=? and platform=? and locked_until <= ?",
                (shop_id, platform, now + 30))
            db.commit()

    raise TimeoutError(f"等待刷新超时 {shop_id}/{platform}")
```

三个细节值得说：

1. **抢锁用的是 `update ... where locked_until is null or < now`**，没有引入 Redis。行级条件更新本身就是原子的，`rowcount == 1` 就是拿到锁。多店铺场景下这比引入一套分布式锁简单，也少一个故障点——除非你多机部署，那就换 Redis 或 PG 的 `select ... for update`。
2. **`locked_until` 是超时时间，不是布尔锁**。进程被 kill、机器重启，锁不会永久卡住——30 秒后自动可抢。用布尔锁的人一定会遇到「一次 OOM 之后这个店再也刷不了 token」。
3. **刷新失败不清旧 token**。这条和第 20 篇单人版一致，但理由不同：单人版是「没空管」，代运营是「断一家店客户当天就知道」。让旧 token 多活一会儿，比立刻 401 好。

## 四、隔离：一个店挂掉，不能带走其他十九个

这是多店铺和单店铺最本质的区别。20 家店的调度如果写成这样：

```python
for shop in shops:
    await sync(shop)      # 一家超时 60 秒，20 家就是 20 分钟
```

那么任何一家店出问题——授权过期、类目权限被收回、平台侧限流——整轮任务就停在那里。更隐蔽的是额度：**20 家店共用一个令牌桶，一家店做大促把额度烧光，其他 19 家的日报全部延迟**。

三件事必须按店铺拆开：

```python
# ops/fanout.py
import asyncio


class ShopWorker:
    """每个店铺一个独立上下文：独立限流桶、独立退避、独立错误计数。
    这就是第 9 篇「舱壁隔离」在代运营场景的具体形态。"""

    def __init__(self, shop_id: str, gate, backoff_s: float = 0):
        self.shop_id = shop_id
        self.gate = gate                 # 按 (shop, platform, api) 分桶
        self.backoff_s = backoff_s

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def run_round(workers: list[ShopWorker], task_fn, timeout: float = 300):
    """一轮调度：所有店铺并发，互不阻塞，单店异常只影响自己"""

    async def one(w: ShopWorker):
        try:
            # 单店整体超时：防止一家店拖住整轮
            return w.shop_id, await asyncio.wait_for(
                task_fn(w), timeout=timeout)
        except asyncio.TimeoutError:
            return w.shop_id, {"ok": False, "reason": "shop_timeout"}
        except PermissionError as e:
            return w.shop_id, {"ok": False, "reason": f"auth:{e}"}
        except Exception as e:
            return w.shop_id, {"ok": False, "reason": repr(e)[:160]}

    results = await asyncio.gather(*(one(w) for w in workers))
    ok = sum(1 for _, r in results if r.get("ok"))
    return {"ok": ok, "fail": len(results) - ok, "detail": dict(results)}
```

关键就一句：`gather(..., return_exceptions=True)` 的效果——**用 `wait_for` 包住单店 + 逐个捕获**，这样一家店超时，其他十九家的结果照常返回。别用 `gather` 的默认行为，它会让第一个异常直接冒出，剩下的协程被取消，等于全挂。

额度也要按店分：

```python
# 每个店铺每天一个额度配额，用完就停，不占别人的
gate = Gate(per_key_quota={
    "shop:taobao:item_list": 800,
    "shop:jd:item_list": 500,
})
```

这样一家店做活动把额度打满，只是它自己停，其他店照跑。**这个设计在 618、双 11 那种日子价值极高**——那一天你一定会遇到某家店的数据量翻十倍。

## 五、数据隔离：靠 review 记住加 `where` 一定会漏

技术隔离是内部问题，数据隔离是**对客户的责任问题**。你的系统里同时有 20 家店的数据，客户 A 登录进来，绝不能看到客户 B 的任何一行。

两种做法，差别很大：

**做法一（错）**：业务代码里手写 `where shop_id = ?`。写的时候都记得，三个月后新加一个查询，忘了。这类事故在代运营行业不是罕见事件。

**做法二（对）**：在数据访问层强制注入，业务层拿不到「不带 shop 的入口」。

```python
# ops/tenant.py
class TenantRepo:
    """所有查询强制带 shop_id。业务代码只有这一个入口，
    想查全库得绕过这个类——那就要 review 了，能拦住。"""

    def __init__(self, db, shop_id: str):
        if not shop_id:
            raise ValueError("shop_id 不能为空")   # 空值 = 全库查询，直接炸
        self.db, self.shop_id = db, shop_id

    def items(self, platform: str, limit: int = 500):
        return self.db.execute(
            "select item_id, payload, updated_at from items "
            "where shop_id=? and platform=? order by updated_at desc limit ?",
            (self.shop_id, platform, limit)).fetchall()

    def orders(self, platform: str, start: int, end: int):
        return self.db.execute(
            "select * from orders where shop_id=? and platform=? "
            "and created_at between ? and ?",
            (self.shop_id, platform, start, end)).fetchall()

    # 需要跨店聚合的（内部报表），单独一个类，命名上区分开
```

`shop_id` 为空就抛异常那两行是关键：**空值等于「查全库」**，历史上无数越权事故就是这么来的（参数没传、前端选错、定时任务忘了带）。让它炸，比让它查全库安全。

再配一条测试，比任何 review 都可靠：

```python
def test_no_cross_shop_leak(db):
    a = TenantRepo(db, "shop_a").items("taobao")
    b_ids = {r[0] for r in TenantRepo(db, "shop_b").items("taobao")}
    assert not ({r[0] for r in a} & b_ids), "A 的查询结果里出现了 B 的商品"
```

这条测试跑在 CI 里，任何人写出串店查询都会被拦下。

## 六、店铺生命周期：进来和走的时候最容易乱

20 家店的代运营，每个月都有新客户进来、老客户离开。这两件事没有做成流程，就会留下垃圾和风险。

**进来（onboarding）**，25 分钟能搞定，但要有清单：

| 步骤        | 容易漏的                                     |
| --------- | ---------------------------------------- |
| 建店记录、绑定平台 | 一个客户在三个平台开店，要建 3 个授权，别只建一个               |
| 商家授权      | **必须商家本人扫码**，不能用员工账号代授——员工离职那天，20 家店一起失效 |
| 核对权限包     | 商品、订单、物流、售后四个包分别确认，缺哪个当场说                |
| 拉一次样本     | 抽 5 个商品 + 3 个订单，跟商家后台对一遍，确认字段和口径         |
| 配日报       | 收件人、时间、关注字段；先发一次给客户确认格式                  |
| 记入到期日历    | 授权到期日 + 48 小时预警，写进系统而不是某人的日历             |

那次「员工用老板账号授权」值得单独说。代运营公司里常见做法是运营用客户的主账号扫码授权，图省事。后果是：客户改密码、老板换手机、员工离职交接不清，任何一种都会让授权失效，而且**失效的时候你不会收到通知**，只会发现数据不更新了。正确做法是让商家用自己的账号授权给你的服务商应用，然后给你开子账号。


**离开（offboarding）**，比进来更重要，因为涉及数据责任：

1. **立即 revoke**：调平台的解绑接口，别只是把状态改成 `revoked`——平台侧的授权还在，只是你不用，等于钥匙还在门上。
2. **数据保留期写进合同**：常见是 30 天，到期物理删除。别口头说「我们会删」。
3. **密钥销毁**：这家店的 token 密文连同 `key_id` 一起删。留着解不开也没意义，但审计时会问。
4. **归档报表**：客户可能要历史数据，导出成文件交给他，然后删库。

第 1 条我们踩过：客户走了三个月，我们只是标了 `revoked`，平台侧授权还在。客户的 IT 后来做安全审计，发现我们的应用还在他的授权列表里，直接投诉到了平台。**这件事的后果不是技术问题，是信任问题。**

## 七、20 家店的日报：一张表看完，异常行标记

最后是每天真正要看的东西。47 个授权，不可能一个个点进去看，所以汇总成一张表，一屏装下：

```python
# ops/dashboard.py
import time

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def daily_board(db) -> list[dict]:
    """一行 = 一个授权。异常行带 reason，前端标红。"""
    rows = db.execute("""
        select g.shop_id, g.platform, g.state,
               (g.expire_at - strftime('%s','now')) / 86400.0  as days_left,
               (select count(*) from items i
                 where i.shop_id=g.shop_id and i.platform=g.platform
                   and i.updated_at > strftime('%s','now') - 86400) as changed,
               (select max(updated_at) from items i
                 where i.shop_id=g.shop_id and i.platform=g.platform) as last_ok
          from grants g
         order by g.state desc, days_left asc
    """).fetchall()

    out = []
    for shop_id, platform, state, days_left, changed, last_ok in rows:
        reason = ""
        if state in ("expired", "revoked"):
            reason = "授权不可用"
        elif days_left is not None and days_left < 2:
            reason = f"授权 {days_left:.1f} 天后到期"
        elif last_ok is None or (time.time() - last_ok) > 86400 * 2:
            reason = "超过 2 天没有新数据"
        elif changed == 0:
            reason = "昨日无变更"      # 不一定是错，但值得看一眼
        out.append({"shop": shop_id, "platform": platform, "state": state,
                    "days_left": round(days_left, 1) if days_left else None,
                    "changed": changed, "reason": reason})
    return out
```

这张表每天 9:00 推到运营群，20 行，异常行带原因。运营的实际工作量从「逐店巡检 2 小时」变成「看 10 分钟，处理 1~3 个异常」。

要注意 `changed == 0` 这一条：**它不是错误，但它是信号**。一家正常营业的店，一天商品和订单一定有变化。连续两天为 0，八成是拉取失败但没报错（第 14 篇讲过的空值问题）。这条规则帮我们抓到过两次静默故障。

## 八、六个坑，按踩到的频率排

| 坑                  | 症状                        | 解法                          |
| ------------------ | ------------------------- | --------------------------- |
| 员工用客户主账号授权         | 员工离职 / 客户改密码，一批店同时失效，且无通知 | 商家本人授权 + 子账号操作              |
| 凭证表主键用 `shop_id`   | 一家店三个平台只有一个 token 生效      | 主键必须是 `(shop_id, platform)` |
| 只加密 access_token   | 数据库泄露 = 永久授权泄露            | refresh_token 一起加密          |
| 全局令牌桶              | 一家店大促把额度吃光，其他店全延迟         | 按店铺分桶                       |
| 手写 `where shop_id` | 某次查询漏了，串店                 | 数据层强制注入 + CI 测试             |
| 离开时只标状态不解绑         | 客户审计发现授权还在                | 调平台解绑接口                     |

## 九、这套东西的规模效应

47 个授权，每天跑一轮增量：

- 全量拉取：约 18,800 次/天（47 × 平均 400 商品）
- 改增量后：约 2,300 次/天（只有变动的商品和订单）

人工部分：巡检从 2 小时/天降到 10 分钟/天，月底对账从半天降到 1 小时。真正省下的不是时间，&#x662F;**「哪家店可能出问题」这件事不再依赖某个人记得**。

---

多店铺管理这件事，技术上没有什么新鲜东西——加密、单飞、隔离、强制注入，都是老办法。它难的地方在于：**每一个细节都对应一个真实客户的损失**。串店是信任问题，断服是续费问题，授权过期是关系问题。

所以这套代码里我写得最仔细的，不是并发和性能，是那几行「失败时不清旧 token」「shop_id 为空就抛异常」「离开时必须真的解绑」。