# 做 ERP 的看过来：三平台的商品订单接口，不用再一家家啃文档（附 Python 源码）

> 标题里「不用再一家家啃文档」要诚实地拆开说：**文档躲不掉，但可以从 300 页降到每平台 30 页**。
>
> 降法不是划重点，是换读法：ERP 只该面对一套「动作」，三个平台怎么实现这些动作，是适配器的事。你真正要啃的只有**差异**——而差异是可以列出来的，这篇就把它列全。
>
> 先划清和之前那篇 ERP 文（第 15 篇）的分工：那篇讲数据进了 ERP 之后怎么消化（金额拆分、账期、状态映射、超时窗口），全是账务语义；这篇只讲接口层——怎么把三个平台的东西拉进来、推出去，以及 ERP 场景特有的那道门槛：**多店铺**。

## 一、ERP 对接的真实工作量：不是学三个 SDK，是管理差异

先破一个误区。多数 ERP 团队评估对接工期时算的是「学三个平台的开发文档」，这不是主要成本。主要成本是**同一个业务动作在三个平台的实现差异**，而且这些差异文档里不会并排放着，你要自己发现。

按 ERP 最关心的五个域，把差异摊开：

| 域  | 方向     | 增量参数差异                                        | 分页差异                   | 最贵的坑                                    |
| -- | ------ | --------------------------------------------- | ---------------------- | --------------------------------------- |
| 商品 | 下行为主   | 淘宝按 `modified`，1688 按更新时间，京东部分接口**只支持按天**     | 三平台游标/页码不一，京东有**分页漂移** | 只同步 SPU 不同步 SKU（第 1 篇的老坑，ERP 里表现为库存对不上） |
| 订单 | 下行为主   | 淘宝按 `modified`，京东**按天且建议双时间字段**，1688 窗口重叠要自己做 | 页码式，深翻页丢数据             | 用创建时间不支付时间做增量 → 跨月订单归错账期（第 15 篇）        |
| 库存 | **双向** | 无增量可言，靠推送 + 定时校准                              | —                      | 上行回写是**写操作**：覆盖式同步会把 ERP 里的预占抹掉（第 3 篇）  |
| 物流 | 上行为主   | —                                             | —                      | 发货回写错单号 = **真发货事故**，这是 ERP 所有接口里责任最重的一个 |
| 售后 | 下行为主   | 三平台退款状态机完全不同（淘宝 7 态、京东 6 态、1688 退款分层级）        | 页码式                    | 仅退款 / 部分退款 / 退货入库成本重算，接口能拉到但语义要自己拼      |

方向那一列值得多看一眼：**ERP 不是「拉数据」，是双向同步**。下行占接口数量的 80%，但上行（库存回写、发货回写）占责任的 80%——下行拉错数据顶多报表难看，上行推错是真实的发货事故和库存事故。

## 二、统一动作层：ERP 只面对 12 个动作

解法就是第 8 篇那套适配器架构，但 ERP 版要按 ERP 的动作习惯重新切。我们把三平台的所有接口收敛成 12 个动作：

```
下行（拉取）                       上行（推送）
item.pull        商品资料          stock.push       库存回写
item_sku.pull    SKU 明细          ship.push        发货回写
buyer.pull       收货人信息        invoice.push     开票回写
order.pull       订单              refund.push      售后审核回写
stock.pull       库存快照          memo.push        卖家备注回写
refund.pull      售后单
logistics.pull   物流轨迹
```

两个纪律：

**① 动作是 ERP 定义的，不是平台定义的。** 平台的接口按它自己的业务切（淘宝一个「订单」接口里混着交易和履约信息），ERP 按自己的单据流切。适配器的活就是把平台切法翻译成 ERP 切法——**翻译层一旦出现 `if 订单金额 > 100` 这种业务判断，说明业务逻辑漏进来了**（第 8 篇的原话，ERP 场景下更常见，因为 ERP 的诱惑就是「顺手在适配层里算个账」）。

**② 能力声明必须显式。** 三个平台不是每个动作都支持。1688 没有按 SKU 的库存概念（按起订量阶梯），京东开票回写的权限包单独申请。**假装都支持是最贵的错误**：调用方传 sku_id，适配器默默返回了商品总库存，不报错，三个月后客户对不上账才被发现。

启动自检（第 8 篇的 `check_capabilities`，ERP 版）：

```python
REQUIRED = {
    "taobao":  ["item.pull", "order.pull", "stock.push", "ship.push", "refund.pull"],
    "alibaba": ["item.pull", "order.pull", "stock.pull", "ship.push"],   # 没有 stock.push
    "jd":      ["item.pull", "order.pull", "stock.push", "ship.push", "refund.pull"],
}

def check_capabilities(shops: list["Shop"]) -> None:
    for shop in shops:
        missing = [a for a in REQUIRED[shop.platform]
                   if a not in CAPABILITIES[shop.platform]]
        if missing:
            # 缺能力是配置错误，启动就该死，别等客户对账时发现
            raise SystemExit(f"{shop.name}({shop.platform}) 缺动作: {missing}")
```

注意 1688 那行的写法：`REQUIRED` 里**不包含** `stock.push`，`CAPABILITIES` 里也没有——它不是缺失，是**不存在**。这两份清单的区别就是「没申请权限」和「平台没这个概念」，混着写会误导后来的人去申请一个不存在的权限包。

## 三、代码：动作路由 + 三平台订单增量 + 发货幂等

`pip install httpx` 即可。代码按 ERP 最疼的两个动作给全：`order.pull`（下行最复杂）和 `ship.push`（上行责任最重）。

### 3.1 统一模型与动作路由

```python
# erp/models.py
from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal


@dataclass
class OrderLine:
    sku_id: str
    title: str
    qty: int
    price: Decimal            # 单价，永远 Decimal（第 4 篇：float 累加对不上账）
    pay_amount: Decimal       # 实付，分摊后的行金额（分摊算法见第 15 篇）


@dataclass
class Order:
    platform: str
    order_id: str             # 平台单号
    shop_id: str              # 多店铺：这单属于哪家店（见第五节）
    status: str               # 归一后的 ERP 状态，不是平台原始状态
    paid_at: str | None       # 支付时间。ERP 按这个记账期，不是 created_at
    lines: list[OrderLine] = field(default_factory=list)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

class NotSupported(Exception):
    """显式报错 > 静默返回错数据。允许调用方 catch 后降级，不允许适配器假装成功"""
```

```python
# erp/base.py
from abc import ABC, abstractmethod
from erp.models import Order, NotSupported

CAPABILITIES = {
    "taobao":  {"item.pull", "item_sku.pull", "buyer.pull", "order.pull",
                "stock.pull", "stock.push", "ship.push", "memo.push", "refund.pull"},
    "alibaba": {"item.pull", "item_sku.pull", "order.pull",
                "stock.pull", "ship.push"},
    "jd":      {"item.pull", "item_sku.pull", "buyer.pull", "order.pull",
                "stock.pull", "stock.push", "ship.push",
                "invoice.push", "refund.push", "refund.pull"},
}


class Adapter(ABC):
    platform: str

    def __init__(self, shop):
        self.shop = shop

    def can(self, action: str) -> bool:
        return action in CAPABILITIES[self.platform]

    @abstractmethod
    async def order_pull(self, since: int, cursor: dict) -> tuple[list[Order], dict]:
        """返回 (订单列表, 游标)。游标是 opaque 的：调用方不解析，原样存取"""

    async def query_order(self, order_id: str) -> Order:
        """ship.push 前的状态复核用，各平台自行实现"""
        raise NotImplementedError

    async def do_ship_push(self, order_id: str, waybill: str,
                           company: str, idem_key: str) -> str:
        """实际回写调用。idem_key 作为平台幂等键传入"""
        raise NotImplementedError

    async def ship_push(self, order_id: str, waybill: str, company: str) -> str:
        raise NotSupported(f"{self.platform} 未开通 ship.push")
```

游标设计成 opaque 的 dict 是刻意的：**调用方不许解析游标内容**。三个平台的游标内部结构完全不同（淘宝存 `modified` 时间戳，京东要存日期 + 页码，1688 存 `last_id`），如果调用方解析了游标，适配器内部就再也不能改增量策略了。

### 3.2 三平台 `order.pull`：差异全在这三个类里

```python
# erp/adapters.py
import asyncio
import time
import httpx

from erp.base import Adapter
from erp.models import Order, OrderLine, NotSupported
from decimal import Decimal

STATUS_MAP = {
    # 平台原始状态 → ERP 归一状态。映射表放适配器里（是翻译，不是业务）
    "taobao":  {"TRADE_FINISHED": "DONE", "WAIT_SELLER_SEND_GOODS": "TO_SHIP"},
    "jd":      {"FINISHED": "DONE", "WAIT_DELIVERY": "TO_SHIP"},
    "alibaba": {"success": "DONE", "waitbuyerpay": "UNPAID"},
}


class TaobaoAdapter(Adapter):
    platform = "taobao"

    async def order_pull(self, since: int, cursor: dict) -> tuple[list[Order], dict]:
        # 游标：modified 时间戳 + last_id 双字段（秒级精度一秒多单会撞，第 3 篇）
        modified = cursor.get("modified", since)
        last_id = cursor.get("last_id", "")
        orders: list[Order] = []
        last_row = None

        async with httpx.AsyncClient(timeout=10) as client:
            while True:
                r = await client.get("/v1/taobao/trades", params={
                    "modified_start": modified,
                    "page_size": 100,
                    # 重叠加窗 60 秒：宁可重复不可遗漏（重复靠幂等吸收）
                })
                r.raise_for_status()
                body = r.json()
                rows = body.get("trades", [])
                for row in rows:
                    key = f"{row['modified']}:{row['tid']}"
                    if key <= last_id:
                        continue  # 双键排序，跳过已处理的
                    orders.append(self._to_order(row))
                    last_row = row
                if not body.get("has_next"):
                    break

        # 游标只推进到最后一条已处理的：结构对调用方 opaque
        if last_row:
            cursor = {"modified": last_row["modified"],
                      "last_id": f"{last_row['modified']}:{last_row['tid']}"}
        return orders, cursor

    def _to_order(self, row: dict) -> Order:
        # 淘宝 HTTP 永远 200，错误藏在 error_response（第 11 篇）——网关已剥过一层
        return Order(
            platform=self.platform,
            order_id=str(row["tid"]),
            shop_id=self.shop.id,
            status=STATUS_MAP[self.platform].get(row["status"], "UNKNOWN"),
            paid_at=row.get("pay_time"),
            lines=[OrderLine(
                sku_id=str(s["sku_id"]),
                title=s["title"],
                qty=int(s["num"]),
                price=Decimal(str(s["price"])),
                pay_amount=Decimal(str(s["payment"])),
            ) for s in row.get("orders", [])],
        )


class JdAdapter(Adapter):
    platform = "jd"

    async def order_pull(self, since: int, cursor: dict) -> tuple[list[Order], dict]:
        # 京东部分订单接口只支持按天（第 3 篇的硬约束）：
        # 增量单位从「小时」降级为「天」，用 last_modified + 去重吸收重复
        day = cursor.get("day") or time.strftime("%Y-%m-%d", time.localtime(since - 86400))
        orders: list[Order] = []

        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.post("/v1/jd/order_search", content=_jd_param_json({
                "startDate": day, "endDate": day, "page": cursor.get("page", 1),
            }))
            r.raise_for_status()
            body = jd_unwrap(r.json())  # 按接口名剥壳，key 写法还不统一（第 11 篇）
            for row in body.get("orderList", []):
                orders.append(self._to_order(row))

        # 按天拉：游标只推进「日期」，重复行由调用方按 order_id 幂等吸收
        return orders, {"day": day, "page": cursor.get("page", 1) + 1}


def _jd_param_json(payload: dict) -> str:
    # 京东的坑：360buy_param_json 里的中文不能被 \uXXXX 转义，
    # 签名按原文算（ensure_ascii=False，第 11 篇那个卡一天半的坑）
    import json
    return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))


def jd_unwrap(body: dict) -> dict:
    for k, v in body.items():
        if isinstance(v, dict) and "result" in v:
            return v["result"]
    return body


class AlibabaAdapter(Adapter):
    platform = "alibaba"

    async def order_pull(self, since: int, cursor: dict) -> tuple[list[Order], dict]:
        last_id = cursor.get("last_id", 0)
        orders: list[Order] = []
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.get("/v1/alibaba/trades", params={
                "last_id": last_id, "size": 50})
            r.raise_for_status()
            for row in r.json().get("result", []):
                orders.append(self._to_order(row))
        return orders, {"last_id": max((o.order_id for o in orders), default=last_id)}

    def _to_order(self, row):
        return Order(
            platform=self.platform, order_id=str(row["id"]), shop_id=self.shop.id,
            status=STATUS_MAP[self.platform].get(row["status"], "UNKNOWN"),
            paid_at=row.get("payTime"),
            lines=[OrderLine(sku_id=str(s["productID"]), title=s["name"],
                             qty=int(s["itemCount"]), price=Decimal(str(s["price"])),
                             pay_amount=Decimal(str(s["itemTotal"])))
                   for s in row.get("orderEntrys", [])],
        )
```

这段代码里每个类的写法都不一样，**这是特性不是缺陷**：淘宝是「时间游标 + 双键排序」，京东是「按天 + 幂等吸收」，1688 是「ID 游标」。差异被关在三个类里，调用方永远只看到 `order_pull(since, cursor) -> (orders, cursor)`。

### 3.3 `ship.push`：责任最重的一个动作

```python
# erp/ship.py
import hashlib

from erp.base import Adapter
from erp.models import NotSupported

# 模块级连接，进程启动时初始化（结构见 store 的 SCHEMA 思路，第 20 篇）
db = None  # type: sqlite3.Connection

SCHEMA_SHIP = """
create table if not exists ship_push(
  idem_key text primary key, platform text, order_id text,
  waybill text, result text, pushed_at integer);
"""

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def _already_pushed(idem: str) -> bool:
    return db.execute("select 1 from ship_push where idem_key=?",
                      (idem,)).fetchone() is not None


async def _mark_pushed(idem: str, adapter: Adapter, order_id: str,
                       waybill: str, result: str):
    import time
    db.execute(
        "insert or ignore into ship_push values(?,?,?,?,?,?)",
        (idem, adapter.platform, order_id, waybill, result, int(time.time())))
    db.commit()


async def ship_push(adapter: Adapter, order_id: str, waybill: str, company: str) -> str:
    """发货回写。三重防呆，缺一个都可能造成真发货事故：

    1. 本地幂等键：同一单号+运单号只发一次（重试安全）
    2. 先查后推：本地未确认发货的才允许推
    3. 平台侧幂等：用 order_id 做平台幂等键，重复推送返回首次结果
    """
    idem = hashlib.sha1(f"{adapter.platform}:{order_id}:{waybill}".encode()).hexdigest()
    if await _already_pushed(idem):
        return "DUPLICATE"

    if not adapter.can("ship.push"):
        raise NotSupported(f"{adapter.platform} 未开通 ship.push")

    # 先查后推：ERP 状态推进和平台状态之间有时间差，推之前再确认一次
    remote = await adapter.query_order(order_id)
    if remote.status not in ("TO_SHIP",):
        raise RuntimeError(f"平台状态 {remote.status} 不可发货，拒绝回写")

    result = await adapter.do_ship_push(order_id, waybill, company, idem_key=idem)
    await _mark_pushed(idem, adapter, order_id, waybill, result)
    return result
```


为什么发货回写要三重防呆，库存回写只要两重？因为**发货回写触发的是物理世界**：平台收到回写就真的安排仓库发货了。重试导致的重复发货、错状态导致的提前发货，都是客户仓库里的实际损失——这类接口的容错原则和下行完全相反：**下行情愿重复，上行情愿少做**。

## 四、多店铺：ERP 的第一道门槛，别到最后才处理

这是 ERP 和自用脚本最本质的区别：**ERP 一个实例服务几十家客户店**。前十九篇里的「店铺」是单数，ERP 里必须是复数。这一节把多店铺的地基讲了，深度展开（存储加密、几十家店并发刷新的工程细节）留给下一篇。

三件事必须从第一天做对：

### 4.1 token 按店铺隔离，谁的 token 只能查谁的数据

第 14 篇讲过的最隐蔽的死法在多店铺下是**必死**：混淆 app 级与用户级凭证，接口全绿，拉到的是别人店铺的数据。ERP 里这不叫 bug，叫**事故**——客户 A 的订单进了客户 B 的库。

```python
# erp/shops.py
import asyncio
import time
import sqlite3

import httpx


class Shop:
    def __init__(self, shop_id: str, platform: str, name: str):
        self.id = shop_id
        self.platform = platform
        self.name = name
        self.user_key = ""       # 店铺级授权凭证，绝不入库表名混用
        self._tokens: dict[str, tuple[str, float]] = {}  # platform -> (token, expire)
        self._refresh_lock = asyncio.Lock()   # 单飞锁，per shop
        self.quarantined_until = 0.0          # 过期隔离（见 4.3）

    async def token(self, app_key: str, secret: str) -> str:
        tok = self._tokens.get(self.platform)
        if tok and tok[1] > time.time() + 300:
            return tok[0]
        async with self._refresh_lock:        # 双检：拿到锁再查一次
            tok = self._tokens.get(self.platform)
            if tok and tok[1] > time.time() + 300:
                return tok[0]
            new_tok, expires_in = await self._do_refresh(app_key, secret)
            self._tokens[self.platform] = (new_tok, time.time() + expires_in)
            return new_tok
```

`_refresh_lock` 是 **per shop** 的，这是多店铺最容易犯的错：用一把全局锁，店铺 A 刷新时店铺 B 的正常调用也跟着排队；或者反过来，没有锁，同一家店两个协程同时刷新——京东和 1688 都是「后刷作废先刷」，两个协程会把自己刷死自己。

### 4.2 授权链路：ERP 走「他用型（ISV）」应用 + 商家授权

第 17 篇讲过应用类型决定权限天花板，ERP 场景再补三层：

- **ERP 是 ISV 应用**，自用型 appkey 根本申请不到多店铺需要的权限包。类型选错要重建应用重审，**这个决定要在写第一行代码之前做**。
- 每家客户店要**店主自己扫码授权**（生成各自的 session/token），ERP 拿的是「店铺级」凭证，不是平台级。客户解绑通常无通知，token 会悄悄失效——靠每日摘要（第 20 篇那套）或每店心跳发现。
- 给客户的说辞要按第 17 篇的场景描述写法来：「系统为该商家管理其自有店铺的订单与库存」，别出现「聚合」「采集」。

### 4.3 单店过期隔离：一家店挂了，不能炸全部

```python
# erp/runner.py
import time


async def run_round(shops: list, adapters: dict, db):
    """一轮同步。核心纪律：单店失败隔离，绝不 raise 到顶层"""
    for shop in shops:
        if time.time() < shop.quarantined_until:
            continue  # 隔离中，跳过，等摘要邮件喊人
        adapter = adapters[shop.platform](shop)
        try:
            cursor = load_cursor(db, shop.id, "order.pull")
            orders, cursor = await adapter.order_pull(since=last_round(db), cursor=cursor)
            save_orders(db, orders)          # 按 order_id 幂等 upsert
            save_cursor(db, shop.id, "order.pull", cursor)
        except Exception as e:
            # 连续失败才隔离：单次失败可能只是网络抖动
            shop.fail_count = getattr(shop, "fail_count", 0) + 1
            if shop.fail_count >= 3:
                # 401/403 大概率是客户解绑或 token 过期 → 隔离 1 小时
                shop.quarantined_until = time.time() + 3600
            record_failure(db, shop.id, repr(e))   # 进每日摘要
    # 循环正常走完：8 家店，1 家授权过期，其余 7 家照常出数据
```

隔离的判断要区分**错误类型**：401/403（授权失效）隔离一小时等客户重新授权；429（限流）降速不隔离；500 连续三次也隔离。**把「哪家店失败了、为什么、隔离到几点」写进每日摘要**，实施顾问早上看一眼就知道今天该找哪家客户重新扫码。

## 五、给 ERP 厂商的实施清单

最后是一份可以贴在项目文档里的清单，按实施顺序：

**售前（决定做不做）**

- [ ] 应用类型确认为 ISV/他用型（选错要重建重审）
- [ ] 五个域 × 三平台的能力矩阵过一遍，缺失动作（如 1688 无 stock.push）已与产品确认替代方案
- [ ] 客户的行业类目在三个平台都有对应权限包（特殊类目权限收紧，第 17 篇）

**开发（决定做多久）**

- [ ] 只啃差异：鉴权一节 + 增量参数一节 + 五个域的接口页，每平台约 30 页
- [ ] 游标 opaque，调用方不解析
- [ ] 上行三重防呆（幂等键 / 先查后推 / 平台幂等）
- [ ] 状态映射表放适配器，业务规则不进翻译层

**运维（决定活多久）**

- [ ] token per shop 隔离 + 单飞锁
- [ ] 单店失败隔离 + 每日摘要含「失败店铺清单」
- [ ] `solo.db`/token 库 `chmod 600`，绝不进 git

---

顺带说一句给 ERP 厂商的账：这套适配层写完是一次性资产，但**它的维护成本是按客户数摊的**——一家客户分摊到的维护费可能撑不起一个全职人力，这就是为什么不少 ERP 团队最后把「平台对接」这一层整体外置（买接口服务，自己只做动作层以上的业务）。算不算得过账，取决于你的客户数，公式和第 12 篇那套 TCO 一样，把「人力」换成「实施人力」就行。

下一篇写多店铺授权的工程细节：token 落库加密、几十家店并发刷新的撞车处理、批量重新授权的引导流程。那是 ERP 接入的最后一坑。

# 写给独立开发者：一个人怎么把三平台数据接口接完（附 Python 源码）

> 先把边界划清：这篇和之前那篇「3 天接通三平台」不是一回事。
>
> 那篇默认你有一个小团队、有预算、有人分工。这篇的第一约束只有一条：**没有第二个人**。你病了没人值班，你忙别的项目时没人看日志，三个月后你自己都会忘记这个服务还活着。
>
> 所以单人版的做法和团队版大量相反。团队版里的最佳实践——事件驱动、统一抽象、熔断降级——对一个人来说，多数是负债：**它们都要花维护意愿，而你的维护意愿是最稀缺的资源**。
>
> 下面按「留什么、砍什么、作弊什么、什么时候放弃」四步讲，代码是一个完整可跑的单人版骨架。

## 一、先算一笔账：一个人接，到底要多少小时

团队版的账（第 6 篇算过）是 13 天。单人版不能按天算，要按小时算——因为你的时间是从晚上和周末里抠的。

诚实的拆法：

| 模块                   | 小时     | 说明                                |
| -------------------- | ------ | --------------------------------- |
| 签名 + token 刷新        | 6      | 三平台三套签名，1688 的 urlPath 参与签名最容易卡半天 |
| 限流 + HTTP 客户端        | 4      | 进程内令牌桶，见下文，不用 Redis               |
| 增量拉取 + 指纹比对          | 8      | 时间窗 + 指纹，SQLite 存                 |
| 失败重放 + 每日摘要          | 6      | 单人版的命根子，后面细讲                      |
| 第二个平台                | 8      | 第一个平台摸熟后，主要是字段差异                  |
| 第三个平台                | 6      | 同上                                |
| 自愈测试（拔网线、改错 key、杀进程） | 2      | 这 2 小时不能省                         |
| **合计**               | **40** | ≈ 5 个周末，或 2~3 周的晚上                |

注意这是**开发**。维护另算：每月 1~~2 小时（平台改字段、授权过期、配额规则变化），一年 12~~24 小时。这笔账第五节要用。

## 二、单人版架构：留什么、砍什么

对照团队版逐项过。原则只有一条：**单人无法持续维护的复杂度，一律砍掉**。

| 团队版做法                | 单人版                         | 为什么                                                   |
| -------------------- | --------------------------- | ----------------------------------------------------- |
| TMC 消息推送（事件驱动）       | **轮询 5~10 分钟 + 时间窗**        | 推送要维护长连接、断线重连、消息确认。轮询简单粗暴，对独立产品够用。规模上去再换，反着来一次的成本是可控的 |
| 三平台统一适配层（第 8 篇）      | **先直连一个平台，第二个平台出现重复代码时再抽象** | 统一层是给维护团队的投资；单人先抽象等于为还不存在的需求写代码。规矩是老的：事不过三            |
| Redis 分布式令牌桶（第 10 篇） | **进程内令牌桶**                  | 你只有一个进程。少一个 Redis，少一类故障，少一笔服务器钱                       |
| 舱壁隔离 + 熔断降级（第 9 篇）   | **超时 + 失败进重放表**             | 熔断的价值是保护其他业务线；你只有一条业务线，它挂了你反正要修                       |
| 监控告警系统               | **每天一封摘要邮件**                | 告警系统要配、要调阈值、要有人响应。摘要邮件零维护，早上看一眼                       |
| 微服务 / 消息队列           | **单进程 + cron**              | 一个进程，一个库，一个定时器。部署 = 一个 systemd 单元                     |
| MySQL / PG + 运维      | **SQLite（WAL 模式）**          | 零安装、零运维、备份就是拷一个文件。单人规模的量它完全扛得住                        |

三条单人的铁律，比上面这张表更重要：

1. **一切设计围绕「失联自愈」**。你一定会有一周忘了它的存在。挂掉的轮次要能自己补回来（重放表），而不是等你发现。
2. **宁可重复，不可抽象**。两个平台的相似代码各写各的，等第三个出现再合并。单人项目里，错误的抽象比重复代码贵十倍——因为改错抽象的人只有你。
3. **归一化模型只做你需要的字段**。第 2 篇那套统一 Item/Sku 模型是给多消费者系统用的。你的小工具只需要展示 6 个字段，那就只映射 6 个。

## 三、代码：一个进程、一个库、五个文件

`pip install httpx` 之外零依赖。目录结构：

```
solo/
├── token_box.py   # token 刷新，双检
├── gate.py        # 进程内令牌桶
├── store.py       # SQLite：指纹表 + 重放表
├── main.py        # 每小时一轮，cron 调
└── digest.py      # 每日摘要邮件
```

### 3.1 token 刷新：双检 + 提前量

第 14 篇的 `TokenManager` 用的是分布式锁——那是多实例场景。单进程用线程锁 + 双检就够了，逻辑一个类讲完：

```python
# solo/token_box.py
import time
import threading
import sqlite3

import httpx

REFRESH_AHEAD = 300  # 过期前 5 分钟就刷，别卡着点


class TokenBox:
    """单进程版 token 管理。SQLite 存 + 线程锁双检。
    注意：多实例部署时这套不够，要用分布式锁（见第 14 篇）。"""

    def __init__(self, db: sqlite3.Connection, platform: str,
                 app_key: str, secret: str, refresh_url: str):
        self.db = db
        self.platform = platform
        self.app_key = app_key
        self.secret = secret
        self.refresh_url = refresh_url
        self._lock = threading.Lock()
        db.execute(
            "create table if not exists tokens("
            "platform text primary key, access_token text, expire_at real)")

    def get(self) -> str:
        row = self.db.execute(
            "select access_token, expire_at from tokens where platform=?",
            (self.platform,)).fetchone()
        if row and row[1] > time.time() + REFRESH_AHEAD:
            return row[0]

        with self._lock:  # 拿到锁后再查一次，防止两个协程轮流刷
            row = self.db.execute(
                "select access_token, expire_at from tokens where platform=?",
                (self.platform,)).fetchone()
            if row and row[1] > time.time() + REFRESH_AHEAD:
                return row[0]

            token, expires_in = self._refresh()
            self.db.execute(
                "insert into tokens(platform, access_token, expire_at) "
                "values(?,?,?) on conflict(platform) do update set "
                "access_token=excluded.access_token, expire_at=excluded.expire_at",
                (self.platform, token, time.time() + expires_in))
            self.db.commit()
            return token

    def _refresh(self) -> tuple[str, int]:
        r = httpx.post(self.refresh_url, data={
            "app_key": self.app_key, "secret": self.secret})
        r.raise_for_status()
        body = r.json()
        # 京东和 1688 都有「后刷作废先刷」的行为，
        # 刷新失败时别把旧 token 删了——还能用就先用着
        return body["access_token"], int(body.get("expires_in", 86400))
```

最后那段注释是单人版特有的容忍：**刷新失败时保留旧 token 继续用**。团队版会立刻告警，你没空响应，那就让旧的多活一会儿，大部分情况它真的还能用。

### 3.2 限流：进程内令牌桶

第 10 篇的 Redis 版是给多实例共享额度用的。单进程，一个字典就够——**少一个依赖，少一类故障**：

```python
# solo/gate.py
import time
import threading


class Gate:
    """进程内令牌桶，按 '平台:接口' 分桶（第 10 篇讲过为什么必须分桶：
    平台的限流维度是 接口 × 应用，不是全局）"""

    def __init__(self):
        self._buckets: dict[str, tuple[float, float]] = {}  # key -> (tokens, last)
        self._lock = threading.Lock()

    def acquire(self, key: str, rate: float, burst: int | None = None):
        burst = burst or max(2, int(rate * 2))
        while True:
            with self._lock:
                now = time.monotonic()
                tokens, last = self._buckets.get(key, (float(burst), now))
                tokens = min(float(burst), tokens + (now - last) * rate)
                if tokens >= 1:
                    self._buckets[key] = (tokens - 1, now)
                    return
                wait = (1 - tokens) / rate
            time.sleep(wait)
```

默认 `burst = rate × 2` 是保守起点——第 10 篇里 `burst=500` 撞 429 的教训，单人版更要远离。你没有人手盯限流告警。

### 3.3 存储：SQLite + WAL + 重放表

整个单人版的灵魂在 `replay` 表。**它替代的是团队版的值班制度**：

```python
# solo/store.py
import hashlib
import json
import time
import sqlite3

SCHEMA = """
pragma journal_mode = wal;  -- 读写不互斥，单人版最关键的一行配置

create table if not exists items(
  platform    text not null,
  item_id     text not null,
  fingerprint text not null,
  payload     text not null,
  updated_at  integer not null,
  primary key(platform, item_id)
);

create table if not exists replay(
  id          integer primary key autoincrement,
  platform    text not null,
  kind        text not null,
  params      text not null,        -- json
  next_try_at integer not null,
  tries       integer not null default 0,
  last_error  text
);
"""


def open_db(path: str) -> sqlite3.Connection:
    db = sqlite3.connect(path)
    db.executescript(SCHEMA)
    return db


def fingerprint(payload: dict) -> str:
    """指纹比对（第 1 篇的思路）：变了才写库，省掉 90% 的无效写"""
    keys = ("price", "stock", "title", "status")  # 只盯你关心的字段
    raw = json.dumps({k: payload.get(k) for k in keys},
                     ensure_ascii=False, sort_keys=True)
    return hashlib.sha1(raw.encode()).hexdigest()

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def upsert_if_changed(db, platform, item_id, payload) -> bool:
    fp = fingerprint(payload)
    row = db.execute(
        "select fingerprint from items where platform=? and item_id=?",
        (platform, str(item_id))).fetchone()
    if row and row[0] == fp:
        return False
    db.execute(
        "insert into items(platform,item_id,fingerprint,payload,updated_at) "
        "values(?,?,?,?,?) on conflict(platform,item_id) do update set "
        "fingerprint=excluded.fingerprint, payload=excluded.payload, "
        "updated_at=excluded.updated_at",
        (platform, str(item_id), fp, json.dumps(payload, ensure_ascii=False),
         int(time.time())))
    return True


def add_replay(db, platform, kind, params, last_error: str):
    """失败的拉取进重放表，指数退避：1h → 2h → 4h → 封顶 12h。
    已有同任务行则原地更新，别插入新行——重试路径会反复走到这里"""
    p = json.dumps(params, sort_keys=True)
    row = db.execute(
        "select tries from replay where platform=? and kind=? and params=?",
        (platform, kind, p)).fetchone()
    tries = (row[0] if row else 0) + 1
    delay = min(3600 * 2 ** (tries - 1), 3600 * 12)
    if row:
        db.execute(
            "update replay set next_try_at=?, tries=?, last_error=? "
            "where platform=? and kind=? and params=?",
            (int(time.time()) + delay, tries, last_error[:300],
             platform, kind, p))
    else:
        db.execute(
            "insert into replay(platform,kind,params,next_try_at,tries,last_error) "
            "values(?,?,?,?,?,?)",
            (platform, kind, p, int(time.time()) + delay, 1, last_error[:300]))


def due_replays(db, limit=50):
    return db.execute(
        "select id, platform, kind, params, tries from replay "
        "where next_try_at <= ? order by next_try_at limit ?",
        (int(time.time()), limit)).fetchall()


def done_replay(db, replay_id: int):
    db.execute("delete from replay where id=?", (replay_id,))
```

为什么要重放表，而不是「失败就重试三次」？因为**你要防的不是网络抖动，是你自己**。某天你改服务器配置把服务弄挂了，两天后才想起来——这两天拉的增量全丢了，重放表会把它们补回来，不需要你做任何事。团队版靠值班兜底，单人版靠表结构兜底。

另外，`updated_at` 加时间戳还有个隐藏用途：摘要邮件要报「最老的数据是多久之前的」，这个数字比「跑没跑成功」更能说明系统是不是悄悄停了。

### 3.4 主循环：每小时一轮

```python
# solo/main.py
"""cron: 0 * * * * cd /opt/solo && /usr/bin/python3 main.py >> run.log 2>&1"""
import asyncio
import json
import sys
import time

import httpx

from gate import Gate
from store import open_db, upsert_if_changed, add_replay, due_replays, done_replay

WINDOW_MIN = 10      # 每轮往前多看 10 分钟，重叠窗口防边界漏单（第 3 篇的思路）
RATE = {
    "taobao":  {"item_list": 5.0},
    "alibaba": {"item_list": 3.0},
    "jd":      {"item_list": 2.0},
}


async def fetch_page(client, platform, updated_start) -> list[dict]:
    """各平台的增量列表接口。示例以淘宝为例，其余平台换参数即可"""
    r = await client.get(f"/v1/{platform}/item_list", params={
        "updated_start": updated_start,
        "page_size": 100,
    })
    r.raise_for_status()
    return r.json().get("items", [])

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def sync_platform(platform: str, since: int, gate: Gate, db) -> tuple[int, int]:
    ok, fail = 0, 0
    updated_start = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(since))
    async with httpx.AsyncClient(
            base_url="https://api.example.com",
            headers={"X-App-Key": platform},   # 按你的网关实际方式传
            timeout=10) as client:
        try:
            items = await fetch_page(client, platform, updated_start)
            for it in items:
                gate.acquire(f"{platform}:item_list", RATE[platform]["item_list"])
                if upsert_if_changed(db, platform, it["item_id"], it):
                    ok += 1
            db.commit()
        except Exception as e:
            fail += 1
            add_replay(db, platform, "item_list",
                       {"updated_start": updated_start}, repr(e))
            db.commit()
    return ok, fail


async def replay_due(db):
    """启动先还账：上次没跑完的先补"""
    for rid, platform, kind, params, tries in due_replays(db):
        p = json.loads(params)
        try:
            async with httpx.AsyncClient(
                    base_url="https://api.example.com",
                    headers={"X-App-Key": platform}, timeout=10) as client:
                items = await fetch_page(client, platform, p["updated_start"])
                for it in items:
                    upsert_if_changed(db, platform, it["item_id"], it)
                done_replay(db, rid)
                db.commit()
        except Exception as e:
            # tries 会 +1，退避继续；退到 12h 还不行的，摘要邮件会报「卡住的重放」
            add_replay(db, platform, kind, p, repr(e))
            db.commit()


async def main():
    db = open_db("solo.db")
    gate = Gate()
    await replay_due(db)

    since = int(time.time()) - 3600 - WINDOW_MIN * 60
    stats = {}
    for platform in ("taobao", "alibaba", "jd"):
        ok, fail = await sync_platform(platform, since, gate, db)
        stats[platform] = (ok, fail)

    # 结果写文件，digest.py 读它（比塞邮件逻辑进主循环干净）
    json.dump({"ts": int(time.time()), "stats": stats},
              open("last_run.json", "w"))


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception:
        # cron 环境里 print 没人看，异常落日志，明天摘要邮件会报「今天没跑」
        import traceback
        traceback.print_exc(file=open("run.log", "a"))
        sys.exit(1)
```


几个刻意的简化，别当成偷懒：

- **不做并发抓取**（第 7 篇那套 20 路并发）。单平台每小时一次增量，量级撑得起串行。等你真的慢了再优化，那时你已经知道瓶颈在哪了。
- **不做配置文件**。三个平台的参数直接写在常量里。独立开发者的项目，YAML 配置层多数是自我安慰。
- **cron 而不是常驻进程**。进程挂了 cron 下个小时自己会拉起来——这是免费的兜底，比 systemd 的 `Restart=always` 心智负担更小（两个都上也行）。

### 3.5 每日摘要：单人版的「监控」

```python
# solo/digest.py
"""cron: 0 21 * * * python3 digest.py
每天一封邮件。内容只有四个数字，30 秒看完。"""
import json
import sqlite3
import smtplib
import time
from email.mime.text import MIMEText
from datetime import datetime, timedelta, timezone

CN = timezone(timedelta(hours=8))


def build(db) -> str:
    run = json.load(open("last_run.json"))
    age = datetime.now(CN) - datetime.fromtimestamp(run["ts"], CN)

    oldest = db.execute("select min(updated_at) from items").fetchone()[0]
    oldest_days = (time.time() - oldest) / 86400 if oldest else -1
    stuck = db.execute(
        "select count(*), max(tries) from replay where tries >= 4").fetchone()

    lines = [
        f"昨日变更: " + ", ".join(
            f"{p} {ok}条" for p, (ok, f) in run["stats"].items()),
        f"最后成功时间: {age.total_seconds()/3600:.1f} 小时前"
        + ("  ← 超过 2 小时，检查服务" if age.total_seconds() > 7200 else ""),
        f"最老数据: {oldest_days:.1f} 天前",
        f"卡住的重放: {stuck[0]} 条（最多重试 {stuck[1] or 0} 次）",
    ]
    return "\n".join(lines)


if __name__ == "__main__":
    db = sqlite3.connect("solo.db")
    body = build(db)
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = f"[solo-sync] {datetime.now(CN):%m-%d} 摘要"
    msg["From"] = "me@example.com"
    msg["To"] = "me@example.com"
    with smtplib.SMTP_SSL("smtp.example.com", 465) as s:
        s.login("me@example.com", "your_password")
        s.send_message(msg)
    print(body)
```

四个数字各管一类静默故障：

- **最后成功时间 > 2 小时** → 服务挂了（cron 没跑、密钥过期、服务器欠费）
- **最老数据 > 3 天** → 某个平台的增量在悄悄失败，但整体流程还在「成功」
- **卡住的重放 > 0** → 有接口连续 4 次以上失败，需要人看一眼了
- **昨日变更数** → 和你对业务的预期对不上时（比如明明在搞活动，变更却很少），说明数据链路可能有洞

这就是单人版的监控哲学：**不做实时告警，做每天 30 秒的例行检查**。告警系统你配完头一周会认真看，第三周开始划掉通知，第三个月忘了密码。一封邮件你至少会扫一眼标题。

## 四、部署：一小时的收尾，别跳过

代码写完只算一半。独立开发者的项目死于「在我电脑上是好的」，这四步一小时搞定：

```bash
# 1. systemd 用户级单元（比系统级安全，不用 root）
mkdir -p ~/.config/systemd/user
cat > ~/.config/systemd/user/solo-digest.timer <<'EOF'
[Unit]
Description=solo digest daily
[Timer]
OnCalendar=*-*-* 21:00:00
Persistent=true
[Install]
WantedBy=timers.target
EOF
systemctl --user enable --now solo-digest.timer

# 2. 数据库备份：一天一次，拷文件就行（SQLite 的好处）
echo '0 22 * * * cp /opt/solo/solo.db /opt/solo/backup/solo-$(date +\%m\%d).db' | crontab -

# 3. 数据库目录权限：600，里面有 token
chmod 600 /opt/solo/solo.db

# 4. 密钥不进代码：环境变量或独立文件，git 里只有占位符
```

第 3 步单独说一下：`solo.db` 里有 `tokens` 表，**泄露 = 三家店铺的授权全丢**。单人项目最容易死在「数据库和代码一起扔进 git 仓库」。

## 五、放弃线：什么时候该停手

这篇如果只写「怎么接」，不说「什么时候别接」，就不诚实了。三个信号，碰到任何一个就该停：

**① 收入覆盖不了维护时间。** 用第一节的账算：开发 40 小时 + 维护 12~24 小时/年。按独立开发者的时薪折成钱，再加服务器、再加你的注意力成本。**如果你的产品月收入低于这个数的一半，直接买现成的接口服务**——不是你不行，是这笔账本来就不成立。

**② 你开始需要第二个消费方。** 数据一旦要喂两个产品（比如一个网页 + 一个小程序），归一化模型、统一接口层这些东西就从负债变资产了——那是第 2 篇和第 8 篇的领域，说明你该按团队版重做了。

**③ 权限天花板到了。** 个人主体能申请的权限包是有限的（第 17 篇讲过应用类型决定上限）。很多商品、订单类权限对企业主体都要审，个人主体经常直接不可见。**这一条不是技术问题，是资质问题，重写多少代码都解决不了**——也是独立开发者最终走向「买服务」或「注册公司」最常见的分叉点。

三平台接口的门槛里，最硬的一直不是代码。

---

二十篇到这里，正好把「一个人」这个视角补上了。前面十九篇默认读者背后有团队，这篇把约束换成「只有你」，砍法就完全不同了：**团队版优化的是吞吐和稳定，单人版优化的是维护意愿**。你接下来要维护的不是这套代码，是你自己看它的兴趣。

# 免费试用 7 天：三平台接口文档 + 示例代码全给（附 Python 源码）

> 先说清楚一件事：**7 天试用最容易白费**。
>
> 我们后台看得到试用账号的调用曲线，一个很规律的现象是——第 1 天调几十次，把「能不能通」验证完了，然后就没有然后了；第 6 天想起来还剩一天，慌忙压一下并发，第 7 天到期，什么结论也没拿到。
>
> 所以这篇不讲「我们多好用」，讲**这 7 天该怎么用**：先验什么、后验什么、每一验用什么代码、什么数字算过、什么数字算不过。文末的四个脚本你可以直接拿去跑，跑完会生成一份决策报告——**那份报告才是试用的产出，不是「跑通了」这三个字**。
>
> 顺便说，这套脚本对任何一家接口服务商都适用。你要是正在对比几家，拿它挨个跑一遍，比听销售讲一小时有用。

## 一、先划清 7 天能做什么、不能做什么

试用不是「免费版产品」，它的边界必须提前讲明，否则第 3 天你就会撞上：

| 能做                | 做不到                       |
| ----------------- | ------------------------- |
| 调通全部核心接口，验证字段是否够用 | 全量数据迁移（额度不够，也不该在试用期做）     |
| 抽样验证数据准确性（对商家后台）  | 长期稳定性验证（7 天看不出季度级故障）      |
| 测出限流拐点和延迟分布       | 拿到和正式开通完全一致的权限（试用权限通常是子集） |
| 走通一个真实业务场景的端到端流程  | 把试用期的数据当生产依赖（到期即停，别埋雷）    |

一条经验：**试用的目标是拿到「买不买」的决策依据，不是「先跑起来再说」**。这两件事的顺序反了，7 天一定白费。

## 二、文档：三平台接口文档，先看这四节

接口文档动辄几百页、上千个接口，全看完是浪费时间。真正决定你能不能落地的只有四节，按这个顺序看：

| 顺序 | 看什么             | 为什么先看这个                                        |
| -- | --------------- | ---------------------------------------------- |
| ①  | **接口列表 + 权限归属** | 先确认「你要的接口存不存在、在哪个权限包里」。这一步能筛掉一半方案              |
| ②  | **请求参数与返回字段**   | 重点看返回字段有没有 `null` 之外的「缺失」——缺 key 通常是权限问题，不是没数据 |
| ③  | **错误码表**        | 只看前 15 个高频码。剩下的你一辈子不会遇到                        |
| ④  | **限流与配额说明**     | 很多文档只写「有频率限制」，具体数值要问。这一条决定你的调度策略               |

三个平台的文档风格差异很大，顺着看会踩坑：

|      | 淘宝开放平台             | 1688 开放平台         | 京东宙斯                         |
| ---- | ------------------ | ----------------- | ---------------------------- |
| 文档组织 | 按**权限包**分组，接口散在各包下 | 按**业务域**分组，中文描述更细 | 按**接口名**平铺，中文文档偏薄            |
| 参数命名 | 下划线（`num_iid`）     | 下划线，字段偏业务语义       | 驼峰 + `360buy_param_json` 包一层 |
| 返回结构 | 顶层直接给业务字段          | 与淘宝接近             | **按接口名再套一层**，key 写法还不统一      |
| 最该注意 | 沙箱常与线上不同步          | 部分接口只有中文文档，字段含义要试 | 中文序列化转义问题（见第 11 篇）           |

我们把这些整理成了一份**接口对照索引**：三个平台的核心接口（商品、SKU、价格、库存、订单、物流，各平台 12 个左右）按「同一业务动作」横向对齐，左边是动作，右边是三平台的接口名和关键差异。你按业务动作查，不用记三套命名。文档和示例代码一起给，不用申请就能看。

## 三、示例代码：四个脚本，覆盖 7 天全部验证

代码是一套的：`client.py` 打底，另外四个脚本各管一件事。全部用 `httpx` + `asyncio`，`pip install httpx` 就能跑。

### 3.1 统一的试用客户端

三个平台签名逻辑不同（淘宝和京东是 MD5 包夹，1688 是 HMAC-SHA1 且 urlPath 参与签名），但试用期你不需要关心这些——网关做了归一，你只面对一套参数和一套返回。

```python
# trial/client.py
from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass, field
from typing import Any

import httpx


@dataclass(frozen=True)
class TrialConfig:
    base_url: str                 # 试用网关地址
    app_key: str
    app_secret: str
    daily_quota: int = 5000       # 试用额度，每天重置
    timeout: float = 8.0


@dataclass
class Result:
    ok: bool
    data: Any = None
    code: str = ""                # 归一后的错误码，见下表
    message: str = ""
    cost_ms: int = 0
    request_id: str = ""          # 出问题时报给服务商用
    raw: dict = field(default_factory=dict)

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TrialClient:
    """试用期最小客户端：只管调用、计时、错误归一，不含业务逻辑。"""

    def __init__(self, cfg: TrialConfig):
        self.cfg = cfg
        self._client = httpx.AsyncClient(
            base_url=cfg.base_url,
            timeout=httpx.Timeout(cfg.timeout),
            limits=httpx.Limits(max_connections=32, max_keepalive_connections=16),
        )
        self.calls = 0

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        await self._client.aclose()

    async def get(self, path: str, **params) -> Result:
        self.calls += 1
        t0 = time.monotonic()
        try:
            r = await self._client.get(
                path,
                params={"app_key": self.cfg.app_key, **params},
                headers={"X-Trial-Secret": self.cfg.app_secret},
            )
        except httpx.TimeoutException:
            return Result(False, code="TIMEOUT",
                          message=f">{self.cfg.timeout}s",
                          cost_ms=int((time.monotonic() - t0) * 1000))
        except httpx.HTTPError as e:
            return Result(False, code="NETWORK", message=str(e)[:120],
                          cost_ms=int((time.monotonic() - t0) * 1000))

        cost = int((time.monotonic() - t0) * 1000)
        body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}

        # 关键：HTTP 200 不等于成功。三个平台都有「200 + body 里报错」的返回
        code = str(body.get("error_code") or body.get("code") or "")
        if r.status_code >= 400 or (code and code not in ("0", "200", "success")):
            return Result(False, code=normalize(code, r.status_code),
                          message=str(body.get("error_msg") or body.get("message") or "")[:200],
                          cost_ms=cost, request_id=r.headers.get("x-request-id", ""),
                          raw=body)

        return Result(True, data=body.get("data", body), cost_ms=cost,
                      request_id=r.headers.get("x-request-id", ""), raw=body)


ERROR_KIND = {
    "AUTH":    {"401", "403", "TOKEN_EXPIRED", "SIGN_ERROR", "INVALID_SIGN"},
    "PARAM":   {"400", "PARAM_MISSING", "PARAM_INVALID"},
    "LIMIT":   {"429", "THROTTLED", "QUOTA_EXCEEDED"},
    "PLATFORM":{"500", "502", "503", "504", "PLATFORM_ERROR"},
    "NOT_SUPPORT": {"404", "API_NOT_FOUND", "CAPABILITY_MISSING"},
}


def normalize(code: str, status: int) -> str:
    raw = code or str(status)
    for kind, codes in ERROR_KIND.items():
        if raw in codes:
            return kind
    return "UNKNOWN"
```

`normalize` 这一层看着简单，但它是整个验证的前提——**如果错误码不归一，你没法统计「这 7 天里失败都是什么原因」**，最后只能说「偶尔报错」，等于没验。

### 3.2 `quickstart.py`：5 分钟跑通，输出一张对齐表

第一个脚本的目的不是「能调通」，而是**让你在 5 分钟内看到三个平台的数据长什么样、差在哪**。这比逐页读文档快得多。

```python
# trial/quickstart.py
import asyncio

from client import TrialClient, TrialConfig

TARGETS = [
    # (平台, 商品/订单标识, 说明)
    ("taobao", "654321098765", "自家店在卖的"),
    ("alibaba", "678901234567", "同款或近似同款"),
    ("jd", "100012043978", "在售竞品"),
]

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def main(cfg: TrialConfig):
    async with TrialClient(cfg) as c:
        rows = []
        for platform, item_id, note in TARGETS:
            item = await c.get(f"/v1/{platform}/item", item_id=item_id)
            if not item.ok:
                rows.append([platform, item_id, f"失败:{item.code}", "", "", ""])
                continue

            d = item.data
            skus = d.get("skus") or []
            prices = [s.get("price") for s in skus if s.get("price") is not None]
            stocks = [s.get("stock") for s in skus if s.get("stock") is not None]
            rows.append([
                platform, item_id, str(d.get("title", ""))[:24],
                f"{min(prices) if prices else d.get('price', '-')}",
                f"{sum(stocks) if stocks else '无SKU库存'}",
                f"{len(skus)}个SKU / {item.cost_ms}ms",
            ])

        print(f"{'平台':<9}{'商品ID':<15}{'标题':<26}{'最低价':<10}{'总库存':<12}{'结构'}")
        print("-" * 88)
        for r in rows:
            print(f"{r[0]:<9}{r[1]:<15}{r[2]:<26}{r[3]:<10}{r[4]:<12}{r[5]}")
        print(f"\n本次消耗 {c.calls} 次额度")


if __name__ == "__main__":
    asyncio.run(main(TrialConfig(
        base_url="https://api.example.com",
        app_key="trial_xxxxxxxx",
        app_secret="trial_secret_xxxxxxxx",
    )))
```

跑完你会立刻发现三件事：1688 那个商品「总库存」是空的、京东的 SKU 结构和淘宝不一样、同一个商品三个平台的标题几乎无法直接比对。这三件事正是后面几天要处理的。

### 3.3 `smoke.py`：9 项冒烟，出一份可粘贴的报告

冒烟的意义不在于「今天通了」，在于**以后每天跑一次，能提前发现三类变化**：授权过期、平台改字段、额度规则调整。所以它必须输出结构化结果，不能只 `print("ok")`。

```python
# trial/smoke.py
import asyncio
import json
from dataclasses import dataclass, asdict

from client import TrialClient, TrialConfig

PROBES = [
    ("taobao",  "item",    {"item_id": "654321098765"}, ["title", "skus"]),
    ("taobao",  "orders",  {"start": "2026-10-01", "end": "2026-10-07"}, ["orders"]),
    ("taobao",  "logistics", {"tid": "1234567890123"}, ["status"]),
    ("alibaba", "item",    {"item_id": "678901234567"}, ["title", "price_ranges"]),
    ("alibaba", "orders",  {"start": "2026-10-01", "end": "2026-10-07"}, ["orders"]),
    ("alibaba", "logistics", {"tid": "1234567890"}, ["status"]),
    ("jd",      "item",    {"sku_id": "100012043978"}, ["name", "price"]),
    ("jd",      "orders",  {"start": "2026-10-01", "end": "2026-10-07"}, ["orders"]),
    ("jd",      "logistics", {"order_id": "23456789012"}, ["status"]),
]


@dataclass
class Probe:
    platform: str
    resource: str
    status: str        # PASS / FAIL / EMPTY / SKIP
    detail: str
    cost_ms: int


async def run(cfg: TrialConfig) -> list[Probe]:
    out = []
    async with TrialClient(cfg) as c:
        for platform, resource, params, required in PROBES:
            res = await c.get(f"/v1/{platform}/{resource}", **params)
            if not res.ok:
                out.append(Probe(platform, resource, "FAIL", res.code, res.cost_ms))
                continue

            # 空数组不等于成功：无数据、无权限、参数错都长这样
            payload = res.data if isinstance(res.data, dict) else {}
            missing = [k for k in required if k not in payload]
            if missing:
                out.append(Probe(platform, resource, "EMPTY",
                                 f"缺字段 {','.join(missing)}", res.cost_ms))
            elif not any(payload.get(k) for k in required):
                out.append(Probe(platform, resource, "EMPTY", "返回空", res.cost_ms))
            else:
                out.append(Probe(platform, resource, "PASS", "", res.cost_ms))
    return out


def report(probes: list[Probe]) -> str:
    lines = ["| 平台 | 资源 | 结果 | 说明 | 耗时 |", "|---|---|---|---|---|"]
    for p in probes:
        icon = {"PASS": "✅", "FAIL": "❌", "EMPTY": "⚠️", "SKIP": "—"}[p.status]
        lines.append(f"| {p.platform} | {p.resource} | {icon} {p.status} | "
                     f"{p.detail or '—'} | {p.cost_ms}ms |")
    return "\n".join(lines)


if __name__ == "__main__":
    cfg = TrialConfig(base_url="https://api.example.com",
                      app_key="trial_xxxxxxxx", app_secret="trial_secret_xxxxxxxx")
    probes = asyncio.run(run(cfg))
    print(report(probes))
    json.dump([asdict(p) for p in probes], open("smoke_result.json", "w"), ensure_ascii=False, indent=2)
```

`EMPTY` 这个状态是刻意加的：**「调用成功但没数据」比「调用失败」危险得多**。失败你会去看，空数据你会以为这个商品就是没库存。淘宝的 `trade_list` 在时间参数超出保留期时就是这么返回的。

### 3.4 `bench.py`：找限流拐点，不是测「最高 QPS」

这是四个脚本里最有价值的一个，也是 90% 的试用会漏掉的一个。

多数人压测是「一路加压，直到报 429，那个数值就是上限」。这错了——**429 出现的时候，你已经有很多请求在超时了**。生产上的表现不是「报限流错误」，是「莫名其妙的超时和变慢」。

正确的做法是**阶梯加压，找延迟开始抬升的那个点**：

```python
# trial/bench.py
import asyncio
import time
import statistics

from client import TrialClient, TrialConfig


async def one_stage(cfg, concurrency: int, seconds: float, path: str) -> dict:
    """固定并发压一段时间，收集延迟样本"""
    latencies, errors, wall_end = [], 0, time.monotonic() + seconds
    async with TrialClient(cfg) as c:
        sem = asyncio.Semaphore(concurrency)

        async def worker():
            nonlocal errors
            while time.monotonic() < wall_end:
                async with sem:
                    r = await c.get(path, item_id="654321098765")
                    latencies.append(r.cost_ms)
                    if not r.ok:
                        errors += 1

        await asyncio.gather(*[worker() for _ in range(concurrency)])
        total = c.calls

    lat = sorted(latencies)
    return {
        "concurrency": concurrency,
        "qps": round(total / seconds, 1),
        "p50": lat[int(len(lat) * 0.50)] if lat else 0,
        "p95": lat[int(len(lat) * 0.95)] if lat else 0,
        "p99": lat[int(len(lat) * 0.99)] if lat else 0,
        "err_rate": round(errors / total, 3) if total else 0,
        "samples": total,
    }


async def find_cliff(cfg: TrialConfig, path="/v1/taobao/item"):
    base = None
    rows = []
    for c in (1, 2, 4, 8, 16, 32):
        st = await one_stage(cfg, c, seconds=20, path=path)
        rows.append(st)
        if base is None:
            base = st["p95"]
        # 拐点判定：P95 涨到基线的 2 倍，或错误率超 1%
        if st["p95"] > base * 2 or st["err_rate"] > 0.01:
            break

    print(f"{'并发':<6}{'QPS':<8}{'P50':<8}{'P95':<8}{'P99':<8}{'错误率':<8}")
    for r in rows:
        print(f"{r['concurrency']:<6}{r['qps']:<8}{r['p50']:<8}{r['p95']:<8}"
              f"{r['p99']:<8}{r['err_rate']:<8}")
    safe = rows[-2]["qps"] if len(rows) > 1 else rows[0]["qps"]
    print(f"\n安全 QPS 建议：{safe}（拐点前一级，约占平台上限的 70%）")
    return rows, safe
```


跑出来的表长这样（我们内部跑的真实形态，数字随账号等级变）：

| 并发 | QPS | P50 | P95 | P99 | 错误率 |
|---|---|---|---|---|---|
| 1 | 3.4 | 280 | 310 | 420 | 0% |
| 2 | 6.7 | 290 | 330 | 460 | 0% |
| 4 | 13.1 | 300 | 350 | 520 | 0% |
| 8 | 25.6 | 320 | **420** | 980 | 0% |
| 16 | 38.2 | 410 | **1,850** | 4,200 | 0.4% |

注意 8 到 16 那一行：**错误率只有 0.4%，看着还行，但 P95 已经从 420ms 涨到 1,850ms**。如果你只看错误率，会以为 16 并发是安全的，上线之后就是一片超时告警。真正的安全值在 8 那级，甚至更低一点。

这也是我建议你**在试用期一定要跑压测**的原因：这个拐点每个账号都不一样，问销售也只能给个大概值，只有自己跑才知道。

### 3.5 `verify.py`：数据准不准，只能和商家后台对

前三步验的是「通不通、快不快」，这一步验的是**「对不对」**，也是最容易糊弄过去的一步——因为很多服务商的接口「返回 200、字段齐全、数值是错的」，你不做对账永远发现不了。

最硬的验证方式是：**让商家从自己后台导一份 CSV，跟接口返回逐条比**。抽样 50 个商品就够，覆盖有活动价、有多个 SKU、刚改过价的。

```python
# trial/verify.py
import csv
from decimal import Decimal

from client import TrialClient, TrialConfig


def load_truth(path: str) -> dict[str, dict]:
    """商家后台导出的 CSV：item_id,title,price,stock"""
    truth = {}
    with open(path, encoding="utf-8-sig") as f:
        for row in csv.DictReader(f):
            truth[row["item_id"].strip()] = {
                "title": row["title"].strip(),
                "price": Decimal(row["price"]),
                "stock": int(row["stock"]),
            }
    return truth


async def verify(cfg: TrialConfig, truth: dict[str, dict], platform="taobao"):
    ok, diffs = 0, []
    async with TrialClient(cfg) as c:
        for item_id, expect in truth.items():
            res = await c.get(f"/v1/{platform}/item", item_id=item_id)
            if not res.ok:
                diffs.append((item_id, "调用失败", res.code, ""))
                continue

            got_price = Decimal(str(res.data.get("price", "0")))
            got_stock = int(res.data.get("stock", 0) or 0)

            if got_price != expect["price"]:
                kind = "价格口径" if abs(got_price - expect["price"]) < expect["price"] * Decimal("0.2") else "价格错误"
                diffs.append((item_id, kind, str(expect["price"]), str(got_price)))
            elif got_stock != expect["stock"]:
                # 库存差 5% 以内通常是时点差，不算错
                if abs(got_stock - expect["stock"]) > max(3, expect["stock"] * 0.05):
                    diffs.append((item_id, "库存时点", str(expect["stock"]), str(got_stock)))
                else:
                    ok += 1
            else:
                ok += 1

    total = len(truth)
    print(f"准确率 {(ok / total):.2%}  ({ok}/{total})")
    for d in diffs[:15]:
        print(f"  {d[0]:<15}{d[1]:<8}预期{d[2]:<10}实际{d[3]}")
    return ok / total, diffs
```

差异分类比准确率本身更有用：

- **价格口径**（差 20% 以内）——通常是券后价 / 活动价 / 含运费的差异。这不算错，是口径没对齐，谈的时候问清楚就行。
- **价格错误**（差异很大）——真问题，直接排除。
- **库存时点**——你导 CSV 和接口取值差了几分钟，正常。差 5% 以上才需要查。

我们内部的验收线是 **准确率 ≥ 99.5%，且「价格错误」为 0**。低于这个数，接入之后每天都会有人找你。

## 四、7 天怎么排

四个脚本写完了，剩下的就是排期。这张表可以直接照着做：

| 天 | 做什么 | 用哪个脚本 | 产出 |
|---|---|---|---|
| D1 | 跑通 + 冒烟 | `quickstart.py` / `smoke.py` | 9 项探针结果，知道哪些接口能用 |
| D2 | 文档对表 | — | 你要的 12 个接口，哪几个缺、缺在哪个权限包 |
| D3 | 数据准确性 | `verify.py` | 准确率 + 差异清单（抽样 50 个） |
| D4 | 限流与延迟 | `bench.py` | 安全 QPS、P95 基线 |
| D5 | 真实场景端到端 | 自己写 | 一次全量铺货，或一份日报，跑完 |
| D6 | 异常与降级 | 手动 | 改错 key、断网、超配额，看返回能不能解释 |
| D7 | 出报告、做决策 | `trial_report.py` | 一页纸：过 / 不过 / 有条件过 |

额度大概这么分（按 5,000 次/天算）：冒烟 60 次、验准 300 次、压测 2,000 次、真实场景 2,500 次。**第一天别把额度烧完**，这是最常见的错误。

D6 那天值得单独说：故意把 `app_secret` 改错一位，看返回是 `AUTH` 还是 `UNKNOWN`；故意把商品 ID 改成一个不存在的，看是 `PARAM` 还是空数据。**错误可解释性决定了你以后排障要花多少小时**——返回 `UNKNOWN` 的服务商，出问题只能找人问，一问就是半天。

## 五、`trial_report.py`：把结果汇总成一页决策报告

7 天结束，你不该拿着一堆 json 去汇报。这个脚本把前面所有结果拼成一份 markdown，带明确的判定：

```python
# trial/trial_report.py
import json
from datetime import date

# 验收阈值（可按业务调，但建议不要放宽「价格错误为 0」）
THRESHOLDS = {
    "accuracy": 0.995,      # 数据准确率
    "p95_ms": 800,          # 常规接口 P95
    "safe_qps_ratio": 1.5,  # 安全 QPS / 业务峰值
    "smoke_pass": 9,        # 冒烟通过项
}


def build(smoke: list[dict], verify_acc: float, price_errors: int,
          bench: list[dict], safe_qps: float, peak_qps: float) -> str:
    passed = sum(1 for p in smoke if p["status"] == "PASS")
    p95 = bench[-2]["p95"] if len(bench) > 1 else bench[-1]["p95"]

    checks = [
        ("冒烟通过", f"{passed}/9", passed >= THRESHOLDS["smoke_pass"]),
        ("数据准确率", f"{verify_acc:.2%}", verify_acc >= THRESHOLDS["accuracy"]),
        ("价格错误", str(price_errors), price_errors == 0),
        ("P95 延迟", f"{p95}ms", p95 <= THRESHOLDS["p95_ms"]),
        ("QPS 余量", f"{safe_qps / peak_qps:.1f}x",
         safe_qps >= peak_qps * THRESHOLDS["safe_qps_ratio"]),
    ]
    verdict = "通过" if all(c[2] for c in checks) else (
        "有条件通过" if sum(c[2] for c in checks) >= 4 else "不通过")

    lines = [f"# 接口试用评估报告（{date.today()}）", "",
             f"**结论：{verdict}**", "", "| 检查项 | 实测 | 判定 |", "|---|---|---|"]
    for name, val, ok in checks:
        lines.append(f"| {name} | {val} | {'✅' if ok else '❌'} |")

    lines += ["", "## 未通过项的处理建议", ""]
    for name, val, ok in checks:
        if not ok:
            lines.append(f"- **{name}**：实测 {val}。"
                         "若是口径差异请与服务商确认定义；若是真差异，换方案比接入后打补丁便宜。")
    return "\n".join(lines)
```

这份报告有个副作用：**它同时是给老板看的材料**。技术人做选型最难的不是判断，是说服——「我觉得这家还行」没有说服力，「准确率 99.8%、P95 420ms、QPS 余量 3.2 倍、五个检查项全过」有。

## 六、试用结束后最容易被忽略的两个坑

**① 试用期的数据不能当生产依赖。** 到期停服，你的任务表里那些外键、缓存里那些商品、报表里那些历史，全部失去来源。正确的做法是试用期只做「读 + 验证」，不做数据落地。要落，就落在你自己库里并标清来源和有效期（见第 18 篇那个 `provenance` 字段）。

**② 试用权限通常是正式权限的子集。** 你试用期能调的接口，正式开通不一定一次全给（要看资质和权限包）；反过来，试用期调不通的，也可能是权限没开而不是接口没有。**这一点必须在 D2 那天和商务确认清楚**，别等到接入时才发现「哦这个要另外申请」。

## 七、文中用到的东西

- 六个文件：`client.py`（打底客户端）、`quickstart.py`、`smoke.py`、`bench.py`、`verify.py`、`trial_report.py`（可直接跑，改配置即可）
- 三平台接口对照索引：核心业务动作 → 三平台接口名 + 关键差异，一页表
- 三平台错误码对照：把三套错误码归一成 5 类，判断「该重试还是该改代码」
- 试用账号：7 天，5,000 次/天，核心接口全开，不需要企业资质

我们把这些打包成了一份试用包，文档和代码不申请也能先看。**建议的顺序是：先看文档对表，确认接口够用，再领试用账号跑脚本**——反过来做，容易在第 2 天发现最关键的接口没有，白折腾。

---

前面十八篇讲的是「怎么把这件事做成」，这篇讲「怎么判断手上的工具行不行」。判断标准我写在代码里了，阈值你可以改成自己的——**但「价格错误为 0」这条建议别改**，它是唯一一个接入之后没法补救的指标。

# 官方授权通道和灰色爬虫，做数据服务该选哪条路（附 python 源码）

> 第 12 篇算过一笔自用的账：自建爬虫三年 37 万，官方 API 17.2 万。那篇的结论是「自用场景下，风险是一票否决项」。
>
> 但很多人问的不是自用——他们想做**数据服务**：把多平台的价格、商品数据聚合起来，卖给商家或开发者。这个问题必须单独回答，因为答案比自用场景更一边倒，而且原因完全不同。
>
> 一句话：**自用，灰色爬虫是成本问题；做成服务，是生存问题。**
>
> 自用时风险由你自己承担，封了就封了，损失可控。做成服务，你的每一个客户都跟你签了 SLA——你的上游哪天断了，你就是在对所有客户违约。这篇讲清楚为什么，以及走官方授权这条路要付的真实代价。

先划清边界：这篇不讨论怎么爬、怎么绕。爬虫的攻防不是技术选型，是法律风险——绕过技术措施可能触及刑法第 285 条，有真实获刑判例。这里只讲两条路的**商业与合规结构**，以及走合规路线需要的工程设施。

## 一、先精确定义「灰」在哪：四条边界

「灰色爬虫」不是一个笼统的概念，它压着四条线。做自用和做服务，压线的深度完全不同：

| 边界    | 说什么                    | 自用         | 做服务                 |
| ----- | ---------------------- | ---------- | ------------------- |
| 技术措施  | 绕过反爬、破解接口 = 刑法 285 条风险 | 不碰就相对安全    | 同样不能碰               |
| 个人信息  | 买家昵称、评价内容、地址都算         | 容易顺手采集     | **对外提供 = 违规放大**     |
| 实质性替代 | 向第三方提供平台数据的替代视图        | 小规模通常够不上   | **必然够得上**——你就是在做这件事 |
| 平台协议  | robots、用户协议、接口条款       | 违约成本是封你一个号 | 封你，然后你违约于所有客户       |

关键差异在第三条。**做数据服务，天然就是「向不特定第三方提供平台数据的聚合视图」**——这正是反不正当竞争法里「实质性替代」的典型形态。爬公开数据自己看，和法律风险之间还有距离；聚合起来卖，距离就没有了。

所以不要问「爬公开数据违不违法」这种笼统问题，要问：**我提供的这个东西，是不是平台功能的替代品，我有没有权利向第三方提供它。**

## 二、做服务，风险是传导的

这是本篇的核心论证。画一条链：

```
上游数据源 → 你的服务 → SLA 合同 → 客户
```

客户看到的、签合同的、付钱的，都是**你**。但你的服务质量，实际上由**上游**决定：

- **官方授权上游断了**：有公告、有工单、有恢复预期，多数故障在小时级恢复，你甚至能提前收到字段变更通知。
- **灰色上游断了**：一夜之间全挂——封号、IP 段拉黑、验证码升级，**没有公告、没有申诉渠道、往往不可恢复**。你连告诉客户「什么时候恢复」都做不到，因为没有「恢复」这回事。

而你和客户签的合同不会写「因为我的上游是灰色的」。SLA 写的是 99.9%、数据准确率、赔偿条款。上游断一天，就是全员违约：赔偿是月费的几倍，但真正的损失是客户流失和行业口碑——数据服务的圈子很小，一次大面积断供，名字就传开了。

**所以数据服务的第一性问题不是「哪条路便宜」，而是：你的上游有没有给你「对客户履约的能力」。** 灰色爬虫在这条上是结构性缺失，不是运营努力能补的。

顺带一个现实：灰色上游的故障是**突发的、全量的**——不存在「降级到 50%」，只有「可用」和「归零」两种状态。这让 SLA 的数学根本不成立。

## 三、官方授权通道的真实形态：不是一条路，是一组上游

「官方授权」也常被误解成「自己去申请开放平台」。做服务时，合法上游其实有四种，各有各的位置：

| 上游                  | 拿到什么         | 适合         |
| ------------------- | ------------ | ---------- |
| **平台开放平台 / ISV 授权** | 你有权限范围内最全的字段 | 自家店铺数据、精修层 |
| **联盟接口**（淘宝客、京东联盟）  | 全站商品的价格、券、佣金 | 覆盖层，门槛低    |
| **持牌数据商采购**         | 归一化后的多平台数据   | 补充层，要尽调合同  |
| **人工与授权补录**         | 长尾与特殊字段      | 量小，但必须留痕   |

三点实操建议：

1. **混合上游是常态，不是妥协。** 没有单一上游能同时满足覆盖、深度、时效。成熟的数据服务商都是「官方为主 + 采购补充 + 人工兜底」——重要的是每一条数据知道自己从哪来（下一节的代码就是干这个的）。
2. **采购持牌数据商要做尽调。** 问清授权链路（第 17 篇那五问直接适用）：是直连官方授权还是二次倒卖？授权范围含不含「对外提供」？合同里把数据合规责任写进去。
3. **「客户授权」也是授权。** 商家把店铺 session 授权给你（他用型 ISV 模式），你拿到的是**该店铺自己数据**的读取权——用它去查别的商家的商品，是越权。私域和行情的边界，第 17 篇讲过，这里只提醒一句：这个错误在自用时是隐患，在服务场景里是事故。

## 四、数据溯源：合规不是纪律，是代码

合规这件事，靠「团队自觉」一定会出事——半年后新人接手，从某个来源抓了批数据直接进了对外接口，没人知道。

所以正确做法是**把授权状态做进数据本身**：每条数据带溯源字段，未验证来源的数据在代码层面就出不了 API。

```python
# provenance.py —— 数据溯源：让「灰数据进不了产品」成为系统属性
from dataclasses import dataclass
from enum import Enum


class SourceType(str, Enum):
    OFFICIAL = "official"        # 平台开放平台 / ISV 授权接口
    AFFILIATE = "affiliate"      # 联盟接口（淘宝客、京东联盟）
    PROCURED = "procured"        # 向持牌数据商采购（有合同）
    MANUAL = "manual"            # 人工录入 / 授权补录
    UNVERIFIED = "unverified"    # 来源未验证——禁止对外提供


@dataclass
class Provenance:
    source_type: SourceType
    authorization_ref: str       # 授权依据：ISV 合同号 / 采购合同编号 / 联盟备案号
    fetched_at: str              # 抓取时间（ISO 8601）
    license_scope: str           # 允许用途：internal_only / client_delivery / redistribution
    upstream_version: str = ""   # 上游接口版本——平台改字段时能圈定影响面


class ComplianceError(Exception):
    pass


DELIVERABLE = {SourceType.OFFICIAL, SourceType.AFFILIATE,
               SourceType.PROCURED, SourceType.MANUAL}

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def assert_deliverable(p: Provenance) -> None:
    """对外提供前强制检查。这个函数在 API 出口处调用，不是在文档里写一遍就完了"""
    if p.source_type is SourceType.UNVERIFIED:
        raise ComplianceError("来源未验证的数据禁止对外提供（internal_only）")
    if p.source_type not in DELIVERABLE:
        raise ComplianceError(f"来源类型 {p.source_type} 不允许对外提供")
    if "client_delivery" not in p.license_scope:
        raise ComplianceError(f"来源 {p.source_type.value} 的授权范围不含对外提供")
```

三个设计点，每一个都有真实用途：

1. **`UNVERIFIED` 不是「不准用」，是「不准出」。** 临时来源的数据可以进内部分析库做研究，但 `assert_deliverable` 挂在 API 出口，它永远发不出去。研究自由和交付合规分开管。
2. **`license_scope` 比来源类型更重要。** 同样是采购数据，合同里授权了 `internal_only` 和 `client_delivery` 是两个世界。签合同时看清楚，代码里对应上。
3. **`upstream_version` 是救命字段。** 平台改字段（第 14 篇讲过字段映射的坑）时，你能精确回答「受影响的数据是从哪个上游、哪个版本抓的」，圈定影响面、定向修复。没有它，就是全量重刷。

## 五、合规自查：定期扫自己的系统

上游合规之外，还要防**自己人**越界。两类常见事故，都可以用脚本定期扫：

**事故一：越权调用。** 拿 A 店的授权去查 B 店的商品——接口返回 200，但这是第 17 篇说的越权，在服务场景里是事故。

**事故二：个人信息入库。** 抓商品数据时顺手把评价里的买家昵称、地址存下来了。单条看是脏数据，攒起来就是一个个人信息库——而你没有处理它的合法性基础。

```python
# audit.py —— 两类自查：越权调用模式 + 疑似个人信息字段
import re

# ---- 自查一：session 与查询对象不匹配 = 越权 ----
def scan_unauthorized_calls(call_log: list[dict],
                            token_owner: dict[str, str]) -> list[dict]:
    """
    call_log:    [{"token_id": "...", "seller_queried": "...", "api": "..."}]
    token_owner: token_id -> 该授权店铺的 seller_id
    返回所有「用 A 店的 token 查 B 店数据」的调用
    """
    alerts = []
    for call in call_log:
        owner = token_owner.get(call["token_id"])
        if owner and call.get("seller_queried") not in (None, "", owner):
            alerts.append({**call,
                           "reason": f"token 属于店铺 {owner}，却查询了 {call['seller_queried']}"})
    return alerts


# ---- 自查二：疑似个人信息字段 ----
PI_FIELDS = {"buyer_nick", "buyernick", "receiver_name", "receivername",
             "receiver_mobile", "receivermobile", "mobile", "id_card",
             "address", "detail_address"}

PI_PATTERNS = {
    "手机号": re.compile(r"\b1[3-9]\d{9}\b"),
    "身份证": re.compile(r"\b\d{17}[\dXx]\b"),
}

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def scan_pi_fields(schema: dict[str, str]) -> list[dict]:
    """
    schema: {"orders.receiver_name": "收货人姓名", ...}
    返回疑似个人信息字段清单及命中原因，逐条评估去留
    """
    hits = []
    for key, desc in (schema or {}).items():
        name = key.split(".")[-1].lower()
        if name in PI_FIELDS:
            hits.append({"field": key, "why": "字段名命中 PI 清单"})
        elif desc and any(p.search(desc) for p in PI_PATTERNS.values()):
            hits.append({"field": key, "why": "字段说明命中格式特征"})
    return hits


def run_audit(call_log, token_owner, schema) -> str:
    pi = scan_pi_fields(schema)
    cross = scan_unauthorized_calls(call_log, token_owner)
    report = [f"[审计] 疑似 PI 字段 {len(pi)} 个，越权调用 {len(cross)} 笔"]
    for h in pi:
        report.append(f"  PI 字段: {h['field']}（{h['why']}）")
    for c in cross[:20]:
        report.append(f"  越权: {c['api']} {c['reason']}")
    return "\n".join(report)
```

这个脚本建议**每月跑一次、接到 CI 里**，输出直接发到内部告警群。它防的不是黑客，是自己系统的熵增：新同事加了张表、新接了个来源、某个字段顺手存了明细——每一次单看都无害，攒起来就是合规事故。

## 六、决策框架：三个问题定路线

不绕弯子，三个问题问完，答案自己就出来了：

| 问题      | 答案是 A      | 答案是 B             |
| ------- | ---------- | ----------------- |
| ① 给谁用？  | 自己/公司内部    | 对外客户（SaaS、API、报告） |
| ② 数据去哪？ | 内部分析，不对外分发 | 聚合后对外提供           |
| ③ 出事谁赔？ | 自己承担，损失可控  | 合同违约，全员赔偿         |

**三问里只要有一个落在 B 列，灰色爬虫就不是选项**——不是「风险高低」的权衡，是商业模式结构上不成立：你无法对客户承诺一个随时可能归零的上游。

只有一个特殊情况值得说：**自用研究、一次性分析、无开放能力的长尾站点**（第 12 篇列过的爬虫合理场景），这些可以做，但数据必须标记为 `UNVERIFIED`——按第四节的机制，它们进得了你的分析库，出不了你的产品。

## 七、一句话收尾

两条路的差别，不在技术，在**商业模式的地基**：

> 灰色爬虫的商业模式建立在「平台没发现」上，每一天都在倒数；官方授权的建立在「平台允许」上，虽然慢，但没有倒计时。

做服务的人选哪条路，其实不是技术选型——是你打算把「客户信任」建在哪块地上。数据的生意是长生意，上游稳，才能谈 SLA；SLA 稳，才有复购。

---

*我们自己做三平台接口服务，走的就是全授权链路：开放平台 + 联盟 + 持牌合作，每个字段可溯源。如果你的选品、比价、监控系统不想在数据来源上冒风险，评论区或私信聊。*