令人惊叹的 旅游专栏, 保持 充满灵感。感谢! 安布拉斯城堡 我高度评价, 这里...
免费测试大麦网API接口大麦网是国内领先的演出票务平台，涵盖演唱会、话剧、体育赛事等票务服务。由于大麦网未正式开放公共 API，以下分析基于其移动端 / 网页端的非官方接口（通过抓包分析），并提供 Python 调用方案。需注意：非官方 API 可能存在稳定性问题，且使用需遵守平...
期待你的精彩评论 加入1831人围观未分类
做 ERP 的看过来：三平台的商品订单接口，不用再一家家啃文档（附 Python 源码）
万邦科技Lex 发表于2026-11-03 浏览21 评论0
# 做 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 源码）
万邦科技Lex 发表于2026-11-03 浏览22 评论0
# 写给独立开发者：一个人怎么把三平台数据接口接完（附 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 源码）
万邦科技Lex 发表于2026-11-02 浏览37 评论0
# 免费试用 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 源码）
万邦科技Lex 发表于2026-11-02 浏览38 评论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 稳，才有复购。
---
*我们自己做三平台接口服务，走的就是全授权链路：开放平台 + 联盟 + 持牌合作，每个字段可溯源。如果你的选品、比价、监控系统不想在数据来源上冒风险，评论区或私信聊。*未分类
淘宝商品 API 申请不下来？三个常见卡点和对策（附 python 源码）
万邦科技Lex 发表于2026-10-31 浏览67 评论0
# 淘宝商品 API 申请不下来？三个常见卡点和对策（附 python 源码）
> 先说结论：申请不下来，绝大多数时候**不是你的技术方案有问题，是材料和场景没写对**。
>
> 淘宝开放平台的接口权限是「按权限包申请 + 人工审核」的机制，审核员看的是三件事：**你是谁（主体）、你要用哪种应用调（类型）、你拿数据干什么（场景）**。这三件事任何一件不对，材料写得再漂亮也会被驳回。
>
> 更麻烦的是，三种失败长得完全不一样：有的是提交就被拒，有的是审核过了但调用报「权限不足」，有的是调用能通但过几个月权限被回收。混在一起处理，只会反复提交、反复被拒。
>
> 这篇按三个卡点拆开讲，最后给一个**权限自检脚本**——申请下来之后第一件事应该是跑它，而不是直接写业务代码。
## 一、先把三种失败分清
| 表现 | 卡在哪 | 谁在拦你 |
| -------------------------------------------- | ---------------- | ------------------ |
| 提交申请就被驳回，附一句「资质不符」 | 主体与资质 | 资质审核 |
| 应用创建了，但权限包里**找不到**你要的接口 | 应用类型 | 平台的产品设计（改不了，只能换类型） |
| 权限申请通过了，调用还是报 `Insufficient isv permissions` | 权限包 / 白名单 / 调用方式 | 运行时鉴权 |
| 用了几个月，突然全部报无权限 | 场景复核 | 风控（调用行为与申报不符） |
第二种最坑：你以为「没权限」，其实是「没资格」。后面细说。
## 二、卡点一：主体与资质——最容易被硬驳回的地方
三个必须过的关，缺一个都卡：
1. **企业主体 + 实名**。个人开发者账号能拿到的商品类接口极少，做商用几乎必然要企业主体（营业执照、统一社会信用代码、法人信息）。
2. **对公账户验证**。企业实名一般要走支付宝对公打款或授权验证，这一步卡住的人不少，因为财务流程慢。
3. **经营范围**。这是最容易忽略、也最致命的一条：**营业执照的经营范围要能覆盖你的申请场景**。你去申请「数据服务」类接口，经营范围里只有「日用百货销售」，大概率直接驳回。
我们的建议很具体：
- 经营范围里补上**信息技术、数据服务、软件开发、计算机系统服务**这几项（工商变更一般 3~5 个工作日，比反复申诉快）。
- 联系人手机号必须是**法人或已备案的经办人**，填错会被系统直接驳回，而你往往收不到明确提示。
- 授权店铺绑定要店主扫码，别让客户等——这一条经常是「材料都齐了但卡了三天」的真实原因。
## 三、卡点二：应用类型选错，权限天花板就定了
淘宝的应用类型大方向分两类：**自用型**（只服务自己的店）和**他用型 / ISV**（服务多个商家）。类型决定了你能申请哪些权限包——**而这是不可逆的**（要换只能新建应用、重新审核）。
所以最坑的一种情况是：你的 appkey 是自用型，去「功能场景」里找商品检索类接口的权限包，**根本找不到**。你会以为是自己没资格申请，实际上是应用类型不支持。
判断方法很简单，也有官方依据：接口文档最下方会列出「该接口所属权限包」和「可申请此权限包的应用类型」。对照一下就知道是哪种情况。
### 场景描述：审核员看的是业务，不是技术
场景描述被驳回，是最常见的软性失败。三条规则：
**① 不要出现技术词，更不要出现敏感词。**
- 「爬虫」「批量抓取」「爬取全网商品」——直接触发敏感词，这类表述会被判定为数据抓取用途。
- 「数据同步」「数据分析」——不是敏感词，但太笼统，审核员无法判断你要拿数据干什么，大概率打回。
**② 写「谁用、用来做什么、数据从哪来、不外传」四件事。**
对比一下：
```
❌ 数据同步
❌ 调用淘宝商品接口获取商品数据用于分析
✅ 为已授权的商家提供店铺商品管理：商家在系统中维护自己的商品，
系统通过商品接口读取本店商品的价格与库存，用于生成补货建议，
数据仅在本系统内使用，不向第三方提供，不用于跨店铺采集合集。
```
**③ 附证据。** 原型图、界面截图、数据流向图，比文字有用得多。审核员每天看成百份描述，有图的申请通过率明显高。
**④ 如果是他用型（ISV），还要说清服务对象。** 「服务哪些商家、怎么签约、数据隔离怎么做」——这三问不答，他会怀疑你在做数据转售。
## 四、卡点三：权限包和接口是多对多，不是一个「商品权限」
这是技术人最容易误解的一层。**「我申请了商品权限」不等于「我能调所有商品接口」。**
`taobao.item.seller.get`（本店商品详情）和 `items.list.get`（商品列表）很可能在**不同的权限包**里。权限包是按业务场景切的，不是按资源类型切的。
所以正确的申请顺序是**从接口倒推权限包**，而不是从类目正推：
1. 列出你最终要调的接口（不是类目，是接口名）；
2. 每个接口文档拉到底，看它属于哪个权限包；
3. 去控制台「应用管理 → 应用证书 → 功能场景」里找这些包，能申请的一次申请（但别一次申请 20 个，首申通过率低）；
4. 找不到的包 → 回到卡点二，是应用类型问题。
### 私域接口和行情接口是隔离的
这条要单独强调，因为它是**封号级**的错误：
- **私域接口**（需要 `session`）：只能读**该授权店铺自己**的数据。
- 拿店铺授权去批量查别的商家的商品——这是越权，不是「权限不够」。轻则权限回收，重则应用封禁、主体追责。
我们见过有人用客户店铺的 session 去扫竞品，理由是「接口返回成功了」。接口返回成功只代表你没超配额，不代表你合规。
### 五个子错误码，对应五种完全不同的动作
调用报 `Insufficient isv permissions` 时，真正的信息在**子错误码**里：
| 子错误码 | 含义 | 该做什么 |
| ---------------------------------------- | ------------- | --------------------------- |
| `isv.permission-api-package-limit` | 应用没有该接口权限 | 去功能场景申请对应权限包 |
| `isv.permission-api-package-empty` | 没关联任何访问包 | 按业务申请对应权限包 |
| `isv.permission-api-package-not-allowed` | 应用标签不允许这类 API | 检查应用标签（如买家应用不能调部分订单接口） |
| `isv.permission-ip-whitelist-limit` | IP 不在白名单 | 安全中心加白名单（生产机 IP 会变，别用动态 IP） |
| `isv.permission-api-widget-only-limit` | 仅允许组件方式访问 | 换调用方式，或换接口 |
把这五个当成同一个「无权限」去调参数，能调一整天。**看到 `parameters` 类子错误反而是好消息——说明权限是有的，只是参数不对。**
顺便提醒：新创建的应用，默认调用额度通常很低（几十次/天这个量级，具体以后台显示为准），要靠真实使用记录逐步提升。所以别一上来就设计成每天几十万次的全量扫描，额度根本不够，而且申报口径和实际调用对不上，会触发复核。
## 五、被拒之后：别重复提交同样的材料
最常见的错误动作是「原样再交一次」。系统有记录，重复提交同样的内容只会延长时间。
正确顺序：
1. **先问清理由**。工单或邮件问「具体是哪一项不满足」，绝大多数会给出方向（资质、场景、业绩证明之类）。
2. **逐条对应补材料**，补的时候在描述里写「针对 X 项，已补充 Y」，让审核员不用重新读一遍。
3. **申诉要证据化**：流程图 + 界面截图 + 数据用途说明 + 不外传承诺。空口说「我们只自己用」没有说服力。
4. **别同时开多个应用刷**。同一主体短期内多个相似应用，会被判定为试探，反而更严。
## 六、实在申请不下来，三条可走的路
| 路径 | 适合 | 代价 |
| --------------- | ------------- | ------------------------------ |
| **换平台先跑通** | 只是想验证业务模型 | 京东、1688 的资质门槛和权限结构不同，有时这边卡、那边通 |
| **走联盟侧接口**（淘宝客） | 做导购、比价、选品 | 需联盟备案；只能拿有推广计划的商品；SKU 明细常不全 |
| **用第三方授权接口服务** | 要快速上线、不想碰资质流程 | 按量付费；关键要问清授权链路和口径 |
**明确不推荐的两条**：借用别人的 appkey（违规，出事责任在你）、自建爬虫补数据（法律风险，第 12 篇算过账）。
## 七、申请通过之后，第一件事是跑权限自检
这是全文最实用的部分。很多人拿到 appkey 就直接写业务代码，结果上线才发现某几个接口没权限，报错还都是一个 `Insufficient isv permissions`。
正确做法：**把要用的接口全部探测一遍，按子错误码分类，生成一张权限清单**。哪些能用、哪些要申请、哪些是应用类型不支持（申请也没用）——一次跑清楚。
```python
# probe.py —— 权限自检：逐接口探测，按子错误码给出动作建议
import asyncio
import time
# 子错误码 -> 动作（顺序敏感：先匹配具体，再匹配泛化）
ACTIONS = {
"isv.permission-api-package-limit": "去「功能场景」申请该接口所属权限包",
"isv.permission-api-package-empty": "未关联访问包，按业务申请对应权限包",
"isv.permission-api-package-not-allowed": "应用标签不允许，检查应用类型/标签",
"isv.permission-ip-whitelist-limit": "安全中心添加 IP 白名单",
"isv.permission-api-widget-only-limit": "仅支持组件方式，改用 widget 或换接口",
"isv.parameters-error": "有权限！参数问题，检查必填与格式",
"isv.invalid-signature": "签名问题，见第 14 篇的排查工具",
"isv.app-call-limited": "超配额，降频或提额",
}
def classify(resp: dict) -> tuple[str, str]:
"""返回 (状态, 建议)。成功/无权限/参数错 三态要分清"""
if "error_response" not in resp:
return "OK", "可用"
err = resp["error_response"]
sub = (err.get("sub_code") or "").strip()
msg = err.get("sub_msg") or err.get("msg") or ""
for code, action in ACTIONS.items():
if sub.startswith(code):
return sub, action
if sub:
return sub, f"未收录子码，原文：{msg}"
return f"code={err.get('code')}", f"无子码，原文：{msg}"
async def probe_one(client, name: str, method: str, params: dict) -> dict:
"""用最小合法参数探测一个接口。参数错也说明有权限，这本身就够用了"""
req = {"method": method, "timestamp": str(int(time.time())), **params}
try:
resp = await client.post_taobao(req, timeout=10)
except Exception as e: # 网络异常不算结论，标记重试
return {"api": name, "method": method, "status": "NETWORK", "action": f"重试：{e}"}
status, action = classify(resp)
return {"api": name, "method": method, "status": status, "action": action}
async def probe_all(client, targets: list[dict]) -> str:
"""并发探测并输出 markdown 表，直接贴进方案文档或工单"""
rows = await asyncio.gather(*(probe_one(client, **t) for t in targets))
lines = ["| 接口 | method | 状态 | 建议 |", "|---|---|---|---|"]
for r in rows:
lines.append(f"| {r['api']} | {r['method']} | {r['status']} | {r['action']} |")
ok = sum(1 for r in rows if r["status"] == "OK")
limited = sum(1 for r in rows if r["status"].startswith("isv.permission"))
lines.append("")
lines.append(f"结论：可用 {ok} / 共 {len(rows)}；权限相关 {limited} 个待申请")
return "\n".join(lines)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ---- 用法：列你要用的接口，参数给最小合法值 ----
TARGETS = [
{"name": "本店商品详情", "method": "taobao.item.seller.get",
"params": {"fields": "num_iid,title,price", "num_iid": "683625971234"}},
{"name": "商品列表", "method": "taobao.items.onsale.get",
"params": {"fields": "num_iid,title", "page_size": "1"}},
{"name": "库存查询", "method": "taobao.item.quantity.get",
"params": {"fields": "num_iid,quantity", "num_iid": "683625971234"}},
]
if __name__ == "__main__":
print(asyncio.run(probe_all(make_client(), TARGETS)))
```
跑完你会得到一张这样的表，直接贴进工单或方案：
```
| 接口 | method | 状态 | 建议 |
| 本店商品详情 | taobao.item.seller.get | OK | 可用 |
| 商品列表 | taobao.items.onsale.get | isv.permission-api-package-limit | 去「功能场景」申请该接口所属权限包 |
| 库存查询 | taobao.item.quantity.get | isv.parameters-error | 有权限！参数问题，检查必填与格式 |
```
三个设计要点：
1. **用最小合法参数探测**。参数错没关系——`isv.parameters-error` 已经证明你有权限了，这比构造完美请求省事得多。
2. **先匹配具体子码，再匹配泛化**。`api-package-*` 有一串，用 `startswith` 保留全貌，未收录的码原样打出来，方便你回头补映射表。
3. **区分网络异常和权限结论**。超时不是「没权限」，别把它写进结论表误导自己。
## 八、一句话收尾
淘宝的接口审核，本质是**场景审核**，不是技术审核。你写的是给审核员看的业务流程，不是给机器看的调用链。
材料按「我是谁、我用哪种应用、我拿数据干什么、数据不出我系统」这四句写清楚，附上图和截图，通过率会明显不一样。
代码层面，申请下来之后**先跑权限自检，再写业务逻辑**——把「无权限」和「参数错」分开，能省掉至少半天。
---
*如果资质流程实在走不通，又急着上线，可以先用我们的三平台接口服务过渡：不需要你申请应用、不需要企业资质，按量付费，字段统一。等自建权限下来再切回去，代码改动很小。评论区或私信聊。*未分类
想做多渠道比价系统，数据源到底该从哪来
万邦科技Lex 发表于2026-10-31 浏览60 评论0
这问题比“用哪个接口”更靠前一层。先给结论：**比价系统的数据源不该只选一个，而应按用途分三层**。只选一个，必然在覆盖和精度之间二选一。
## 五类数据源的真实差异
未分类
1688 选品 + 淘宝比价 + 京东竞品监控，一条数据流跑起来（附 python 源码）
万邦科技Lex 发表于2026-10-30 浏览75 评论0
# 1688 选品 + 淘宝比价 + 京东竞品监控，一条数据流跑起来（附 python 源码）
> 前面写了很多篇怎么把数据拿下来：签名、限流、稳定性、字段对齐。但拿下来之后呢？
>
> 数据本身不产生价值。**用三平台的数据回答三个问题**才产生价值：这个货最低多少钱能拿到（1688）、同款现在卖多少钱（淘宝）、对手在做什么动作（京东）。
>
> 这篇把三个问题串成一条流，从采集到输出一张「能不能做」的清单。中间最难的环节不是采集——是**同款识别**。三个平台对同一个商品的描述方式完全不同，认错了，后面全错。
## 一、先说清这条流要输出什么
技术方案要服务于一个具体产物。我们做这条流，最后只输出一张表，每天早上推到企业微信：
| 商品 | 1688 拿货价 | 起订量 | 淘宝价格带（P25/中位/P75） | 京东在售竞品 | 预估净利率 | 退货率红线 | 建议 |
| ------------ | -------- | ------ | ----------------- | ------ | ----- | ----- | -- |
| 不锈钢保温杯 500ml | 28.0 | 2 件代发 | 49 / 65 / 89 | 7 家 | 28.8% | 125% | 可做 |
| 硅胶折叠水壶 | 41.5 | 50 件起批 | 39 / 45 / 52 | 12 家 | -2.1% | — | 放弃 |
| 玻璃冷水壶 1.2L | 19.8 | 5 件代发 | 45 / 58 / 79 | 3 家 | 31.5% | 88% | 重点 |
这张表的四列数据来源不同、更新频率不同，但**必须在同一行上对齐**——这是同款识别要解决的问题。
三段的职责：
- **1688（货源）**：成本价、起订量、是否支持一件代发。日级更新就够，价格不常动。
- **淘宝（比价）**：不是找最低价，是找**价格带**。最低价往往是不能做的货（亏本引流、规格不同、二手翻新）。
- **京东（竞品）**：不是看有多少竞品，是看竞品**做了什么动作**——降价、上新、断货。
## 二、最难的环节：同款识别
同一个保温杯，三个平台可能这样写：
```
淘宝： 【爆款】316不锈钢保温杯男女士大容量水杯便携 500ml 送礼定制
京东： 富光保温杯 316不锈钢 500ml 白色 FGL-500
1688： 厂家直销 316不锈钢真空保温杯 500ml 定制LOGO 2件起批
```
纯标题匹配会把这当成三个完全不同的商品——关键词几乎没有一个完全重合。而实际上它们就是同一个货。
### 三级匹配：类目 → 标题 → 主图
我们的做法是三道筛子，逐级加严：
| 级别 | 手段 | 作用 | 成本 |
| -- | -------------------- | ------------ | ------ |
| 一级 | 叶子类目 + 价格量级 | 粗筛，砍掉 90% 噪音 | 极低 |
| 二级 | 标题归一 + 字符 2-gram 相似度 | 找候选 | 低 |
| 三级 | 主图 dHash 汉明距离 | 定案 | 需要下载图片 |
实测数据：**纯标题匹配的误判率约 40%，加上主图后降到 8%**。所以主图那一步不能省——图片是唯一跨平台稳定的特征，同一张图在三个平台都会出现（哪怕是供应商给的白底图）。
### 标题归一：去营销词，但必须保留规格词
这一步的规则是**反直觉的**：营销词要全部删掉（它们纯属噪音），但规格词必须留下并归一化——`500ml`、`0.5L`、`500毫升` 是同一个东西，而 `500ml` 和 `750ml` 是两个不同的货，价格能差 30%。
```python
# matching.py —— 标题归一：去噪音，保留并统一规格
import re
NOISE = ["包邮", "爆款", "新款", "正品", "厂家直销", "一件代发", "买一送一",
"限时", "特价", "热销", "网红", "同款", "批发", "定制", "送礼", "便携"]
UNIT = {"ml": ("ml", 1), "毫升": ("ml", 1), "l": ("ml", 1000), "升": ("ml", 1000),
"g": ("g", 1), "克": ("g", 1), "kg": ("g", 1000), "千克": ("g", 1000)}
SPEC_RE = re.compile(r"(\d+(?:\.\d+)?)\s*(ml|毫升|l|升|g|克|kg|千克)")
def normalize_title(title: str) -> str:
t = title.lower()
for w in NOISE:
t = t.replace(w, "")
t = re.sub(r"[【】\[\]（）()《》\s\-_/|,，、!！?？+*~]+", " ", t)
def repl(m):
num = float(m.group(1))
unit, factor = UNIT[m.group(2)]
return f" {int(num * factor)}{unit} " # 0.5l / 500毫升 -> 500ml
return " ".join(SPEC_RE.sub(repl, t).split())
```
跑一下上面三个标题，归一后是这样：
```
不锈钢保温杯男女士大容量水杯 500ml
富光保温杯不锈钢 500ml白色 fgl-500
不锈钢真空保温杯 500ml logo
```
现在「保温杯」和「500ml」都浮出来了，2-gram 相似度就能算得比较准。
### 相似度：中文用字符 2-gram，不要用空格分词
中文商品标题没有空格分词，用 `split()` 分出来的往往是一整个长串。字符 2-gram 简单粗暴但效果好，而且零依赖：
```python
def bigrams(s: str) -> set[str]:
s = re.sub(r"\s+", "", s)
return {s[i:i + 2] for i in range(len(s) - 1)} if len(s) > 1 else {s}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def jaccard(a: str, b: str) -> float:
A, B = bigrams(a), bigrams(b)
return len(A & B) / len(A | B) if A | B else 0.0
```
### 主图 dHash：64 位指纹，汉明距离 ≤10 判相似
dHash 的思路是把图缩到 9×8 灰度，逐像素和右边邻居比较，得到一个 64 位整数。它只关心**明暗结构的梯度**，所以对缩放、轻微裁剪、压缩都不敏感——正好适合跨平台的主图。
```python
# image_hash.py
from PIL import Image
def dhash(path: str, size: int = 8) -> int:
img = Image.open(path).convert("L").resize((size + 1, size), Image.LANCZOS)
px = list(img.getdata())
bits = 0
for row in range(size):
for col in range(size):
i = row * (size + 1) + col
bits = (bits << 1) | (1 if px[i] > px[i + 1] else 0)
return bits
def hamming(a: int, b: int) -> int:
return bin(a ^ b).count("1")
```
`hamming ≤ 10` 判为相似（64 位里最多差 10 位）。这个阈值我们是拿 500 对人工标注样本调出来的，你可以按自己的品类重调——**服装类要放宽到 14**，因为模特姿势和背景差异更大。
### 组合打分：权重和阈值都要保守
```python
from decimal import Decimal
W = {"title": 0.55, "image": 0.35, "price": 0.10}
def price_plausible(cost: Decimal, retail: Decimal) -> float:
"""成本价与售价的倍数关系：正常 1.5~5 倍；<1 或 >8 基本是不同货或不同规格"""
if cost <= 0 or retail <= 0:
return 0.0
ratio = float(retail / cost)
if ratio < 1.0 or ratio > 8.0:
return 0.0
return 1.0
def match_score(title_sim: float, image_sim: float, price_score: float) -> float:
"""image_sim: 1 - hamming/64；无图时传 None 走保守降权"""
if image_sim is None:
# 没图不能直接当 0 分，按标题相似度打折，且总分封顶 0.7（进人工复核）
return min(title_sim * 0.9, 0.7)
return (W["title"] * title_sim + W["image"] * image_sim
+ W["price"] * price_score)
def is_same(score: float, threshold: float = 0.72) -> bool:
return score >= threshold
```
**阈值偏保守是有意为之。** 漏匹配的代价是少一个候选，人工还能补；误匹配的代价是按错的价格算利润，最后备货备错——后者贵得多。所以我们把 0.6~0.72 这段叫「灰区」，全部丢进人工复核队列，一天几十条，运营顺手就过了。
## 三、利润模型：把隐性成本算进去，否则算出来的都是幻觉
选品最容易犯的错是「售价减拿货价，毛利 40%，能做」。真实情况是：**毛利 36% 的大件，退货率 25% 的时候净利率只剩 10.7%**，再扣人工仓储就是负的。
四个必须进模型的隐性成本：
1. **平台佣金**（约 5%，各平台各品类不同）
2. **推广费**（新店起量期 10~20%，成熟期 5%）
3. **退货成本**——不只是退钱，是「发货运费 + 回程运费 + 包装折损 + 商品可能无法二次销售」
4. **资金占用**——批量拿货便宜 20%，但货压在仓里，两个月卖不掉就是负现金流
```python
# profit.py
from dataclasses import dataclass
from decimal import Decimal
@dataclass
class CostModel:
price: Decimal # 售价
supply: Decimal # 1688 拿货价
ship_out: Decimal # 发货运费
ship_back: Decimal # 退货回程运费
pack_loss: Decimal = Decimal("3") # 退货包装/折损
commission_rate: Decimal = Decimal("0.05")
ad_rate: Decimal = Decimal("0.10")
return_rate: Decimal = Decimal("0.15")
def breakdown(self) -> dict:
gross = self.price - self.supply - self.ship_out
commission = self.price * self.commission_rate
ad = self.price * self.ad_rate
per_return = self.ship_out + self.ship_back + self.pack_loss
return_loss = self.return_rate * per_return
net = gross - commission - ad - return_loss
return {
"毛利": gross,
"佣金": commission,
"推广": ad,
"退货损失": return_loss,
"净利": net,
"净利率": net / self.price if self.price else Decimal("0"),
}
def breakeven_return_rate(self) -> Decimal:
"""
退货率红线：净利归零时的退货率。
选品时看这个数比看净利率有用——它告诉你「这个货能承受多差的售后」
"""
gross = self.price - self.supply - self.ship_out
after_fixed = gross - self.price * (self.commission_rate + self.ad_rate)
per_return = self.ship_out + self.ship_back + self.pack_loss
if per_return <= 0:
return Decimal("1")
return max(Decimal("0"), after_fixed / per_return)
```
拿三个真实的品算一遍，都是我们做过或评估过的（费率统一按佣金 5%、推广 10%）：
```
【保温杯】售价 65，拿货 28，发 6，回 8，折损 3，退货率 15%
毛利 31.00 (47.7%)
佣金+推广 9.75
退货损失 2.55 (15% × 17)
净利 18.70 (28.8%) 退货率红线 125% → 健康
【服装】售价 129，拿货 45，发 5，回 9，折损 8，退货率 40%
毛利 79.00 (61.2%)
佣金+推广 19.35
退货损失 8.80 (40% × 22)
净利 50.85 (39.4%) 退货率红线 271% → 扛得住
【大件家居】售价 599，拿货 320，发 60，回 120，折损 80，退货率 25%
毛利 219.00 (36.6%)
佣金+推广 89.85
退货损失 65.00 (25% × 260)
净利 64.15 (10.7%) 退货率红线 49.7% → 危险
```
第三个才是真正的杀手。**它单件退货成本 260 元，比毛利 219 元还高**——退一单等于白卖一单还倒贴。所以退货率红线只有 50%，而大件的行业退货率经常就在 30% 上下，稍有波动就到线。
注意这三个品的**毛利率排序和净利率排序是反的**：服装毛利 61% 但净利 39%，大件毛利 37% 净利只剩 11%。只看「毛利率」选品，会系统性地偏向高毛利品类，而高毛利品类往往正是退货成本最高的。
所以 `breakeven_return_rate()` 的真正价值不是「算红线」，而是**逼你把退货率填对**。我们内部的做法：新品按类目历史退货率的 1.2 倍填，换供应商时重算，大件类目额外单列「退货物流成本」字段——因为它的运费可能比货值还高。
## 四、竞品监控：监控「变化」，不是监控「数据」
这是另一个常见的浪费：写个脚本每天把竞品价格、销量、评价数存一遍，看板上堆满数字，然后没人看。
**人要的不是数据，是「发生了什么」。** 所以我们只输出五类事件：
| 事件 | 触发条件 | 业务含义 |
| ------ | ---------------------------- | ---------------- |
| 降价 | 幅度 ≥3% **且** 超过该商品 30 天波动 3σ | 对手在抢量，可能要跟 |
| 涨价 | 同上（反向） | 对手在提利润，是机会窗口 |
| SKU 减少 | SKU 集合出现消失项 | 对手断货或砍款，可能是我们的机会 |
| 断货 | `in_stock` 由 true 变 false | 立刻可抢排名 |
| 评价增速异常 | 日增评价数的 z-score > 3 | 对手在冲量（刷单或大促） |
关键在**阈值要用该商品自己的历史波动做基线**，不能拍脑袋定 3%：
```python
# monitor.py
from decimal import Decimal
def detect_events(prev: dict, cur: dict, sigma: Decimal) -> list[tuple[str, dict]]:
"""prev/cur: {"price", "in_stock", "sku_ids", "review_count"}"""
events = []
# 价格：幅度门槛 + 基线门槛，双条件才报，避免 99 -> 98.9 这种噪音
if prev["price"] and cur["price"]:
delta = (cur["price"] - prev["price"]) / prev["price"]
if abs(delta) >= Decimal("0.03") and abs(delta) >= 3 * sigma:
events.append(("PRICE_DROP" if delta < 0 else "PRICE_UP",
{"delta": round(float(delta), 4),
"from": prev["price"], "to": cur["price"]}))
if prev["in_stock"] and not cur["in_stock"]:
events.append(("OUT_OF_STOCK", {"price": cur["price"]}))
gone = set(prev["sku_ids"]) - set(cur["sku_ids"])
if gone:
events.append(("SKU_REMOVED", {"sku_ids": sorted(gone)}))
# 评价增速：用「日增评价数」的 z-score，不是评价总数
prev_v, cur_v = prev.get("review_velocity"), cur.get("review_velocity")
review_sigma = prev.get("review_sigma") # 该商品历史日增评价数的标准差
if prev_v is not None and cur_v is not None and review_sigma:
z = (cur_v - prev_v) / review_sigma
if z > 3:
events.append(("REVIEW_SURGE", {"z": round(float(z), 2),
"velocity": cur_v}))
return events
```
这个设计救过我们一次：最初阈值定的是「价格变动 >1%」，结果一个竞品每天微调 1 块钱（99→98→97.5），一天推 300 条告警，运营把通知静音了，三周后对手直接降到 79 没人知道。加了「3σ 基线 + 3% 幅度」双条件之后，一天最多 5 条，每条都值得看。
**监控清单是人定的，不是全类目扫。** 一个类目全扫是几万次调用，其中 99% 你根本不关心。我们的做法是运营每周挑 20~60 个 SKU 进监控池，只盯这些。
## 五、把三段串起来：两级池 + 分级频率
调用量是这条流最大的成本，靠「两级池」收敛：
```
全类目候选 20,000 ──粗筛（日级，1 次/天）──▶ 候选池 800
│
┌─────────┴─────────┐
淘宝比价（4 次/天） 京东竞品（48 次/天，仅精选 60）
│ │
└────▶ 决策表 + 事件推送
```
- **候选池 800 个**：1688 日扫一次（800）+ 淘宝 6 小时比一次（3,200）→ 约 4,000 次/天
- **精选池 60 个**：运营确认过的，京东 30 分钟一次 → 2,880 次/天
- 合计约 6,900 次/天；如果全量按 30 分钟扫 2 万个商品，是 96 万次/天
**调用量降到约 1/140，监控频率反而从 12 小时提升到 30 分钟。** 这就是第 13 篇那条铁律的又一次应用：按变更频率分级调度，只对值得高频的对象高频。
调度骨架可以直接复用第 13 篇的 `DueScheduler`，只需换一张频率表：
```python
LEVELS = {
"supply": {"interval_s": 86400, "pool": "candidate"}, # 1688 货源
"retail": {"interval_s": 21600, "pool": "candidate"}, # 淘宝比价
"rival": {"interval_s": 1800, "pool": "selected"}, # 京东竞品
}
```
## 六、三个结论
1. **同款识别是地基。** 认错了，比价是错的、利润是错的、备货也是错的。宁可用保守阈值 + 人工复核，也不要让误匹配混进决策表。
2. **利润要算到净利，退货率是隐藏杀手。** 填 5% 退货率的模型得出的结论，不如不填；`breakeven_return_rate()` 的用处是逼你把假设填对。
3. **监控要监控事件，不要监控数字。** 每天的 300 条噪音会让运营关掉通知，那比不监控更糟。
最后一句合规提醒：这套流用的都是平台授权接口返回的公开商品数据，**不采集买家信息、不抓取评价内容原文**。竞品监控是看公开的挂牌价格和库存状态，这和抓取用户数据是两件事，别混在一起做。
---
*如果不想自己维护这套同款识别和分级调度，我们的三平台接口服务已经把同款聚合、价格带统计、竞品事件都做成了标准输出，接进来自带上面这张决策表。评论区或私信聊。*未分类
电商 ERP 对接三平台，90% 的问题都出在字段和超时上（附 python 源码）
万邦科技Lex 发表于2026-10-30 浏览62 评论0
# 电商 ERP 对接三平台，90% 的问题都出在字段和超时上（附 python 源码）
> 我们做过十几个 ERP 对接项目，复盘下来有个很稳定的分布：**字段问题 62%，超时问题 28%，剩下的 10% 才是签名、限流、权限这些**。
>
> 这个分布有点反直觉——工程师通常以为对接难在协议和加密，实际难的地方在**语义**和**时间窗口**。
>
> 原因也简单：普通的数据同步，是把平台的数据搬到自己库里；ERP 对接，是让**平台的交易记录**和**企业的会计账**在同一个时间窗口内闭合。前者错了解析字段重跑就行，后者错了要调账。
>
> 这篇讲两件事：为什么 ERP 的字段问题特别多，以及 ERP 说的「超时」到底指什么。
先说清楚边界：单位和时区那类错误（分/元、秒/毫秒）第 14 篇讲过，这里不再重复。这篇讲的是**更靠上的语义层**——字段含义对不上，以及业务时限兜不住。
## 一、ERP 与普通同步的根本差别：交易视图 vs 凭证视图
平台接口返回的是**交易视图**：这笔交易发生了什么，买家付了多少钱，货发到哪。
ERP 需要的是**凭证视图**：这笔业务记哪个科目、算哪个成本中心、归哪个账期、冲销哪张单。
两种视图之间不是字段改名，是**结构变形**。下面四个变形，是 62% 字段问题的来源（按出现频次排序）：
| # | 变形 | 平台给的 | ERP 要的 | 错了的后果 |
| - | ----------- | ---------------- | ------------------------ | ---------------- |
| 1 | 金额拆分 | 一个 `payment`（实付） | 商品金额 + 运费 + 优惠 + 应收 + 佣金 | 毛利算不对，优惠成本全压在订单上 |
| 2 | SKU → 物料 | 一个销售 SKU | 多个物料 + 数量（组合装） | 库存只扣一部分，静默虚高 |
| 3 | 时间 → 账期 | 创建时间、支付时间 | 业务日期（可配日结切点） | 跨月订单归错月，GMV 差几个点 |
| 4 | 订单状态 → 单据状态 | 5 个交易状态 | 审单 / 配货 / 出库 / 开票多张单据 | 状态机错乱，已出库单被重复发货 |
### 变形一：金额是一个数，但账需要五个数
平台给 `payment`（买家实付），ERP 要做三件事：拆出商品金额与运费；把订单级优惠**分摊到行**（否则算不出单品行毛利）；算出应收（扣平台佣金后的金额）。
其中最容易出错、也最能体现专业度的是**优惠分摊**：
```python
# erp/money.py
from decimal import Decimal, ROUND_HALF_UP
CENT = Decimal("0.01")
def allocate_discount(total_discount: Decimal, lines: list[dict]) -> list[Decimal]:
"""
把订单级优惠按行金额比例分摊到行。
lines: [{"line_id": ..., "amount": Decimal}] —— amount 是分摊前的行金额
返回: 与 lines 等长的分摊额列表（正数表示优惠额）
两个必须处理的边界：
1. 尾差：整除不了时，差额补在最后一行，否则各行相加 ≠ 订单优惠，对账天天差 0.01
2. 最后一行被补成负数：极端比例下尾差可能超过该行金额，要回退给金额最大的行
"""
if total_discount == 0:
return [Decimal("0")] * len(lines)
base = sum(l["amount"] for l in lines)
if base == 0: # 全额赠品单：无金额可分摊，全部记到首行
out = [Decimal("0")] * len(lines)
out[0] = total_discount
return out
shares, allocated = [], Decimal("0")
for l in lines[:-1]:
share = (l["amount"] * total_discount / base).quantize(CENT, ROUND_HALF_UP)
shares.append(share)
allocated += share
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
tail = total_discount - allocated # 尾差兜底
last = lines[-1]["amount"]
if shares and tail > last: # 最后一行承受不住，转给金额最大行
idx = max(range(len(lines)), key=lambda i: lines[i]["amount"])
shares[idx] += tail
shares.append(Decimal("0"))
else:
shares.append(tail)
return shares
```
**那个尾差不是理论问题。** 我们遇到过一个客户，对账表每天差 0.01，财务查了两周，最后发现是三个行按比例分摊，四舍五入后多出 1 分钱没人认领。ERP 对账是**分毫必究**的，普通业务系统里可以忽略的四舍五入，在这里会每天生成一条差异记录。
还有个容易漏的：**赠品行的 `amount` 是 0，但必须保留**。很多人清洗数据时把 0 元行删了，看起来数据更干净，实际是把这个赠品的**成本丢出了订单**——月末算单均成本时偏低，越卖越亏还找不到原因。
### 变形二：一个销售 SKU 不等于一个物料
组合装是最典型的坑。平台上一个「三件套礼盒」是一个 SKU，ERP 里是三个物料各若干件：
```python
# erp/bom.py
# 平台 SKU = 销售单元；ERP 物料 = 库存单元。两者不是一一对应
BOM: dict[str, list[tuple[str, int]]] = {
"SKU-GIFT-3IN1": [("MUG-001", 2), ("COASTER-002", 4), ("GIFTBOX-003", 1)],
"SKU-CUP-PAIR": [("MUG-001", 2)],
}
def expand(sku_id: str, qty: int) -> list[tuple[str, int]]:
"""把销售数量展开成物料需求；非组合装原样返回"""
bom = BOM.get(sku_id)
if not bom:
return [(sku_id, qty)]
return [(mat, n * qty) for mat, n in bom]
```
不展开的后果是**静默的**：卖了 100 套，只扣了 100 个杯子的库存（如果 SKU 编码恰好和主物料同名），包装盒和杯垫根本没扣。系统不报错，账面看着正常，直到盘点发现包装盒少了 100 个——**而这时已经严重缺货，还在接单**。
比这更隐蔽的是反方向：**平台侧修改了组合装成分**（运营把三件套改成了两件套），你本地的 BOM 没更新。新订单按旧 BOM 扣料，多扣一个杯子，日积月累库存对不上，看谁都像偷了货。所以 BOM 要有版本和生效时间，且**跟着商品详情一起增量同步**（第 13 篇的详情同步应该带出组合装信息）。
### 变形三：ERP 要的不是「时间」，是「账期」
平台给你 `created`（下单）和 `pay_time`（支付），ERP 要的是**业务日期**——这笔钱算哪个月的。三条规矩：
1. **用支付时间，不用创建时间。** 上月 30 号下单、本月 1 号付款的订单，按创建时间归上月，上月 GMV 虚高、本月偏低，两个月的报表都要解释。
2. **日结切点可配。** 很多 ERP 的日结在凌晨 1~2 点（财务第二天早上结前一天的账），0 点到 2 点之间的订单应该算前一个业务日。
3. **时区必须显式。** 这条第 14 篇讲过，但 ERP 里更要命：财务的月报是以北京时间的自然月为准的。
```python
# erp/calendar.py
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
CN = ZoneInfo("Asia/Shanghai")
def business_date(paid_at: datetime, cutoff_hour: int = 0) -> str:
"""业务日期：按支付时间归属账期，支持自定义日结切点"""
t = paid_at.astimezone(CN) if paid_at.tzinfo else paid_at.replace(tzinfo=CN)
if cutoff_hour and t.hour < cutoff_hour:
t -= timedelta(days=1) # 凌晨单归前一业务日
return t.strftime("%Y-%m-%d")
```
### 变形四：订单状态机不能直接搬
第 3 篇讲过订单状态机怎么设计。ERP 里还要多一层：一个平台订单在 ERP 里会**生成多张单据**（销售订单 → 出库单 → 应收单 → 发票），每张单据有自己的状态。把平台状态直接映射成 ERP 主单状态，会出现「已出库但未开票」这类中间态无处安放的问题。
**规矩是：平台状态映射到「事件」，事件驱动 ERP 单据流转。** 不要映射状态，映射事件。`WAIT_SELLER_SEND_GOODS` 不是一个状态，是「可以配货了」这个事件。
## 二、超时：ERP 说的超时，八成不是接口超时
这是最容易被技术人员误解的部分。工程师听到「超时」，第一反应是调 `timeout=` 参数；而在 ERP 场景里，真正的超时发生在三层，而且**接口层是最好解决的那层**：
| 层级 | 时限量级 | 超时的后果 | 谁来兜 |
| ------- | ----------- | ------------ | ------- |
| 接口超时 | 秒 | 页面转圈、用户重试 | 工程师 |
| 审单/发货时限 | 小时~天 | 平台扣分、罚款、流量降权 | 运营 + 系统 |
| 对账窗口 | 天（T+1 / 月结） | 账不平，财务停工 | 系统 + 财务 |
### 2.1 接口层：ERP 前台是同步阻塞的，所以不能长重试
普通后台同步可以慢慢重试（第 9 篇的预算式重试），但 ERP 不一样：**用户在页面上点「同步订单」，鼠标在那儿等着**。一次重试 30 秒，用户就以为卡死了，会连点五次——于是同一批订单被并发拉五遍。
解法不是把超时调长，而是**把长事务拆成「提交 + 查结果」**：
```python
# erp/jobs.py —— 提交-轮询模式：前台 3 秒拿到 task_id，重活交后台
import asyncio
import uuid
class JobService:
def __init__(self, store, worker):
self.store = store # 持久化任务表（对应第 2 篇的 SQLite 任务表思路）
self.worker = worker
async def submit(self, kind: str, idem_key: str, payload: dict) -> str:
"""幂等键：同一批订单连点五次，只产生一个任务"""
exist = await self.store.find_by_idem(idem_key)
if exist:
return exist.task_id
task_id = uuid.uuid4().hex
await self.store.create(task_id, kind, idem_key, payload, status="pending")
self.worker.wake() # 不 await 执行，立即返回
return task_id
async def status(self, task_id: str) -> dict:
task = await self.store.get(task_id)
return {"task_id": task_id, "status": task.status,
"done": task.done, "failed": task.failed,
"errors": task.errors[:10]} # 只回传前 10 条，前端可展开
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 后台 worker：每项任务有独立的业务截止时间（DDL），到点没做完就切下一批
async def run_batch(items, handler, ddl_seconds: int = 300):
loop = asyncio.get_running_loop()
deadline = loop.time() + ddl_seconds
skipped = []
for it in items:
if loop.time() >= deadline:
skipped.append(it) # 超预算的放回队列，下一轮继续，不算失败
continue
await handler(it)
return skipped
```
三个设计点：
1. **幂等键 = 业务主键**（比如 `sync_order:2026-10-29:shop_123`）。用户连点、两个运营同时点、定时任务和手动任务撞在一起，都只产生一个任务。
2. **接口 3 秒内一定返回**。返回 `task_id` 而不是结果，前端轮询 `status`。这比把 `timeout` 调到 60 秒体验好得多，也避免了长连接占资源。
3. **后台任务有 DDL，超了就把剩下的放回队列**，不要硬跑完——因为下一次调度很快就到，硬跑完只会让两轮任务叠在一起（这就是第 10 篇说的「任务堆积」）。
### 2.2 业务层：库存回写要两段式，不能「等接口成功」
超卖的根因在 ERP 里通常不是接口慢，而是**扣库存的时机错了**。
想一想 ERP 的出库流程：审单 → 生成出库单 → 拣货 → 出库 → 回写平台库存。如果只在「出库完成」时才扣库存，那么从审单到出库这段时间，库存是**虚高**的，平台还在卖，卖出去的单进来又排队等出库——超卖就是这么堆起来的。
正确做法是**预占 + 确认**（第 3 篇讲过 `total - reserved - safety` 的账本，这里给 ERP 的落法）：
```python
# erp/stock.py —— 预占 + 确认，两段式
async def reserve(order_id: str, sku_id: str, qty: int, ttl_s: int = 3600) -> bool:
"""审单通过就预占，不等到出库。返回 False 表示可用库存不足"""
ok = await db.execute(
"UPDATE stock SET reserved = reserved + :q "
"WHERE sku_id = :s AND total - reserved - safety >= :q",
{"q": qty, "s": sku_id},
)
if ok:
await db.execute(
"INSERT INTO reservation(order_id, sku_id, qty, expires_at) "
"VALUES (:o, :s, :q, now() + :ttl)",
{"o": order_id, "s": sku_id, "q": qty, "ttl": ttl_s},
)
return bool(ok)
async def confirm(order_id: str):
"""出库成功：预占转实扣，同步回写平台"""
await db.execute(
"UPDATE stock SET total = total - r.qty, reserved = reserved - r.qty "
"FROM reservation r WHERE r.order_id = :o AND r.sku_id = stock.sku_id",
{"o": order_id},
)
await db.execute("DELETE FROM reservation WHERE order_id = :o", {"o": order_id})
await push_stock_to_platform(order_id)
async def release_expired():
"""TTL 到期的预占自动释放——取消单、审核不通过、接口超时都靠这个兜底"""
await db.execute("DELETE FROM reservation WHERE expires_at < now() RETURNING sku_id, qty")
```
关键在 `release_expired`：**所有异常路径最终都收敛成「预占超时自动释放」**。审核卡住、出库单作废、平台接口 502，不需要每种情况都写补偿逻辑，等 TTL 到就行。这是把分布式一致性问题降级成一个定时任务。
### 2.3 对账层：日切和差异表
T+1 对账是 ERP 的硬需求，坑集中在两处：
- **日切边界**：ERP 按业务日切（前面 `business_date`），平台按自然日，两边窗口不一样，边界上一定有差异。解法是**对账窗口取两边的交集并额外多取一天**，差异在窗口内消化。
- **差异要有表，不能只有日志**。差异记录必须能落到 `recon_diff` 表里（订单号、金额差、状态差、处理结果），财务才能逐条勾对。只在日志里喊一句「对账不平」，等于没做对账。
## 三、把两类问题收成一份契约表
字段和超时，看起来是两个问题，本质是一个：**你和 ERP 之间对「这笔业务的含义」和「多久必须到达」没有书面约定**。
所以我们每个 ERP 项目都会产出一份 `contract.yaml`，既是配置，也是交接文档：
```yaml
# contract.yaml —— ERP 对接契约：含义 + 单位 + 缺失处理 + 超时语义
fields:
paid_amount:
source: { taobao: payment, jd: orderPrice, alibaba: totalAmount }
semantics: 买家实付，含运费、已扣优惠
unit: CNY # jd 原始单位为分，由适配器归一（见第 14 篇）
missing: error # 缺失即报错，不允许落 0
discount:
semantics: 订单级优惠总额
allocate: by_line_amount # 必须分摊到行，规则写进契约
sku_id:
semantics: 销售单元，可能对应多个物料
expand: bom # 出库前必须展开，见 bom.py
pay_time:
semantics: 业务日期基准（不是创建时间）
business_day_cutoff: 2 # 凌晨 2 点日结
timeouts:
sync_order:
deadline_ms: 3000 # 前台同步接口上限
on_timeout: enqueue_job # 超时不重试，转异步任务
idem_key: "sync_order:{shop_id}:{date}"
push_stock:
deadline_ms: 2000
on_timeout: retry_queue # 后台任务可重试，退避由链路层负责
consistency: reserve_confirm # 两段式，不允许直写
reconcile:
window: T+1
diff_table: recon_diff
```
这张表的价值在项目交付后：**换人接手时，不用读代码猜「这个字段到底什么意思」，表里写着。** 半年后平台改字段，也是从这张表开始改，改完跑一遍启动自检确认没有漏配。
```python
# erp/contract_check.py —— 启动自检：契约里声明的，代码里必须都实现
def check(contract: dict, adapters: dict):
problems = []
for name, spec in contract["fields"].items():
for platform in spec["source"]:
if platform not in adapters:
problems.append(f"{name}: 缺 {platform} 适配器")
elif not hasattr(adapters[platform], f"map_{name}"):
problems.append(f"{name}: {platform} 未实现映射")
if spec.get("allocate") and not spec.get("unit"):
problems.append(f"{name}: 声明了分摊但没声明单位")
if problems:
raise SystemExit("契约定稿失败:\n " + "\n ".join(problems))
```
启动就炸，比上线三天后发现某个平台少了个字段好得多。
## 四、一句话总结
ERP 对接难，不难在协议。**难在平台描述的是「交易」，ERP 记的是「账」；平台保证的是「最终一致」，ERP 需要的是「账期内闭合」。**
字段是语义翻译问题，超时是时间窗口问题。把这两件事都写进契约表，90% 的问题在发生之前就消掉了。
剩下 10% 是签名、限流、授权那些，前面几篇都写过：签名和鉴权看第 14 篇，限流看第 10 篇，链路层稳定性看第 9 篇。
---
*如果不想在这一层反复踩坑，我们提供的三平台数据接口已经把金额拆分、物料展开、业务日期这三件事做成了标准字段，直接对接 ERP 少写一半适配代码。评论区或私信聊。*未分类
鉴权、签名、字段映射：三平台对接最容易翻车的三个环节（附 python 源码）
万邦科技Lex 发表于2026-10-29 浏览76 评论0
# 鉴权、签名、字段映射：三平台对接最容易翻车的三个环节（附 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 源码）
万邦科技Lex 发表于2026-10-29 浏览70 评论0
# 商品详情、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 小时都够。**关键是分级这件事本身，不是每级的具体数字。**