Указанные элементы задают логическую осн...
注册账号免费测试1688API接口1688 的 item_review 接口（官方标准方法名：alibaba.item.review.get）是阿里巴巴开放平台提供的核心接口之一，用于获取指定商品的买家评论数据，包括文本评价、评分、晒图、采购属性（如采购数量、...
还有 85 人发表了评论 加入10854人围观未分类
双 11 前必须打通的三条数据链路，晚一周就来不及（附 python 源码）
万邦科技Lex 发表于2026-11-08 浏览27 评论0
# 双 11 前必须打通的三条数据链路，晚一周就来不及（附 python 源码）
距离大促只剩几天，这个时间点最不该做的事是加功能。
但有三条链路如果现在还没打通，这几天必须打通。不是为了让系统更先进，是为了**不在这三天里出事故**。
大促出事故，从来不是因为系统不够快。是三种最朴素的情况：
- **卖错了价**——活动没生效、券没叠加、改价延迟
- **卖超了货**——多平台同时卖同一批货
- **客户付了钱没发货**——订单漏拉，等你发现时已经被投诉
这三件事在平时都是小概率。在大促它们是必然，因为触发条件全被放大了：
| 放大项 | 平时 | 大促 | 后果 |
| ------ | -------- | ---------------- | --------------- |
| 请求量 | 100 单/小时 | 5,000 单/小时 | 限流从"偶尔撞"变成"必然撞" |
| 变更频率 | 一天改 3 次价 | 一小时改 3 次 | 人工操作的错误率线性上升 |
| 单次错误代价 | 超卖赔 5 块 | 赔 50 + 平台处罚 + 口碑 | 容错空间被压到零 |
所以这篇不讲架构，只讲三件事：**每条链路的红线是什么、崩了会怎样、怎么用一百多行代码守住**。
## 一、先讲一个认知：大促期间你不是在"同步"，是在"核对"
这是全文最重要的一句话，也是大多数人搞错的地方。
日常开发的心智模型是"同步"：我拉取平台数据，写进我的库，保持一致。
大促期间这个模型会崩，因为**写者有两个人**：
```
运营 → 在平台后台改价、改库存、设活动
你 → 在系统里同步价格、同步库存
```
两个写者，必然打架。你 10:00 同步下来的价格，运营 10:01 在后台改了，你的库就是错的；反过来，你写回去的价格可能覆盖掉运营刚设的活动价。
**正确的心智模型是：平台是唯一真相，你的系统只做三件事——读、核对、告警。**
- 你**不写**价格（除了你自己有定价权的那部分）
- 你**不猜**库存（平台的数就是数，你只维护自己占用的部分）
- 你**不假设**订单拉全了（必须用单数对账验证）
这个转变很难，因为它意味着放弃"我控制一切"的安全感。但大促期间，控制欲就是事故源。
下面三条链路，全都建立在这个前提上：**每个 guard 都是"预期 vs 实际"的比对器，不是同步器。**
## 二、先把地基打了：一张告警表 + 一个落库函数
三条链路都要写告警，所以先把这部分抽出来。它只有几十行，但有两个设计决定值得说：
```python
# guard/base.py
"""三条链路共用的地基：表结构 + 告警落库。
为什么告警落库而不是只发到群里：
大促期间消息会被刷屏,没人看得过来。
落库后能按 level 过滤、按 code 聚合,
事后还能复盘"到底哪类问题先出现"。
"""
import sqlite3
import time
SCHEMA = """
create table if not exists alert(
id integer primary key autoincrement,
level text not null, -- P0 立刻处理 / P1 当天处理
code text not null, -- 机器可读,用于聚合和静默
detail text,
created_at integer,
acked integer default 0
);
create index if not exists ix_alert_time on alert(created_at);
create table if not exists price_check(
id integer primary key autoincrement,
item_id text, sku_id text,
expected text, actual text, diff text, layers text,
checked_at integer
);
create index if not exists ix_price_item on price_check(item_id, checked_at);
create table if not exists stock_level(
item_id text primary key,
available integer, reserved integer, real integer,
floor integer, updated_at integer
);
create table if not exists stock_back(
id integer primary key autoincrement,
item_id text, qty integer, source text, day text, created_at integer
);
create table if not exists order_gap(
platform text, hour_start integer,
local_count integer, remote_count integer, checked_at integer,
primary key (platform, hour_start)
);
-- 这张是你们业务库里的表，这里给个最小结构好让 order_lag 能跑
create table if not exists orders(
order_id text primary key,
platform text,
paid_at integer, -- 买家支付时间,平台给
created_at integer -- 进入我方系统的时间,我方记
);
create index if not exists ix_orders_created on orders(created_at);
"""
def open_db(path="promo.db"):
db = sqlite3.connect(path)
db.executescript(SCHEMA)
return db
def alert(db, level, code, detail):
"""落库 + 立刻 commit。
不用 ORM、不进队列,是刻意的:告警是最后一道防线,
它自身依赖的组件越少越好。大促期间任何中间件抖动
都不该让告警丢失。
"""
db.execute(
"insert into alert(level, code, detail, created_at) values(?,?,?,?)",
(level, code, detail, int(time.time())),
)
db.commit()
print(f"[{level}] {code} {detail}")
```
两个决定：
1. **`code` 字段是机器可读的**（`PRICE_DRIFT`、`ORDER_GAP`），不是给人看的中文。这样能做聚合——一晚上 300 条 `STOCK_LOW` 应该合成一条"37 个 SKU 触底"，而不是 300 条消息。
2. **告警必须落库，不能只发消息群。** 群消息在大促会被刷屏，落库的能查、能统计、能复盘。
## 三、链路一：价格——红线是偏差不超过 1 分钱
### 大促期间它为什么不一样
平时改价是低频事件，错一次改回来就行。大促不一样：
- 活动价有**生效时间点**（0 点、10 点、20 点），差一分钟就是一批订单按错价成交
- 优惠叠加有三层（标价 → 单品券 → 平台满减），任何一层没生效，到手价就是错的
- 改价窗口集中在开抢前 10 分钟，运营手忙脚乱，改错的概率最高
### 最容易崩的那个点：你核对的是标价，不是到手价
标价对得上，不代表到手价对得上。
```
标价 99.00
单品券 -10.00 ← 券没生效
平台满减 -15.00
到手价 74.00 ← 客户实际付的
```
如果只看标价，99 是对的，一切正常。但客户实际付了 84 而不是 74，客诉从这里开始；反过来如果券提前生效，客户付了 64，每单亏 10 块。
**所以价格核对必须核到"到手价"这一层**，而且要能拆开看是哪一层没生效。
### 代码
```python
# guard/price.py
"""价格核对器。注意：它不是同步器，不写平台价格。
定位：运营是唯一写者，我们是观测者。
发现偏差 → 告警 → 人工确认 → 人工修。
绝不允许程序自动"修正"平台价格——大促期间自动写价是灾难。
"""
import time
from decimal import Decimal, InvalidOperation
from base import alert
TOL = Decimal("0.01") # 1 分钱。低于这个数就是数据精度问题，不是业务问题
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def to_dec(v):
try:
return Decimal(str(v))
except (InvalidOperation, TypeError, ValueError):
return None
def check_price(db, item_id, sku_id, expected, actual, layers=None):
"""核对到手价。
expected: 我方算出的到手价（标价 - 券 - 满减）
actual: 平台返回的实际到手价
layers: {"list": 99, "coupon": -10, "promo": -15}，用于定位是哪一层错
返回 (是否通过, 偏差)
"""
e, a = to_dec(expected), to_dec(actual)
if e is None or a is None:
# 拿不到价本身就是故障：大促期间价格字段为空，通常是接口降级了
alert(db, "P0", "PRICE_MISSING",
f"{item_id}/{sku_id} 价格不可用 expected={expected} actual={actual}")
return False, None
diff = a - e
db.execute(
"insert into price_check(item_id, sku_id, expected, actual, diff, layers, checked_at) "
"values(?,?,?,?,?,?,?)",
(item_id, sku_id, str(e), str(a), str(diff),
str(layers or {}), int(time.time())),
)
if abs(diff) <= TOL:
return True, diff
# 偏差超过 1% 就是资损级，1 分钱到 1% 之间是配置级
level = "P0" if abs(diff) > e * Decimal("0.01") else "P1"
where = _locate(layers, e, a) if layers else "未知层"
alert(db, level, "PRICE_DRIFT",
f"{item_id}/{sku_id} 预期 {e} 实际 {a} 偏差 {diff} 疑似 {where}")
return False, diff
def _locate(layers, expected, actual):
"""把偏差归到具体某一层，方便运营直接去改那一项。
注意：券和满减都存负数，所以这里是加，不是减。
写成减法会把符号翻两次，明明是券没生效，却报"我方算价错"。
"""
lst = to_dec(layers.get("list"))
coupon = to_dec(layers.get("coupon")) or Decimal("0")
promo = to_dec(layers.get("promo")) or Decimal("0")
if lst is None:
return "标价缺失"
# 先确认我方算得对
if lst + coupon + promo != expected:
return "我方算价逻辑"
# 我方算对了，那就是平台少减了某一层——逐层试，命中哪个就是哪个
if actual == lst + promo:
return "券未生效"
if actual == lst + coupon:
return "满减未生效"
if actual == lst:
return "券和满减都未生效"
return "平台价格与任一层组合都不符"
```
### 大促期间的核对频率
平时 30 分钟一次就够。大促要按活动时间点加密：
| 时间窗 | 频率 | 说明 |
| ----------- | ---- | --------------- |
| 活动生效前 10 分钟 | 10 秒 | 盯住生效，这是最容易出错的窗口 |
| 活动生效后 10 分钟 | 10 秒 | 确认生效且没有错层 |
| 活动进行中 | 1 分钟 | 运营仍在改价、改券 |
| 凌晨低峰 | 5 分钟 | 省配额 |
**不要全天 10 秒一次**。大促期间配额是最贵的资源，价格核对一秒钟扫一万个 SKU，会把库存和订单链路的额度吃光——这是第 10 篇那个"全局桶"的另一种表现。
## 四、链路二：库存——红线是可售不低于安全水位
### 大促期间它为什么不一样
平时库存是"还剩多少"。大促库存是"我承诺了多少"。
一个 SKU 在大促期间同时被四种东西占用：
```
物理库存 120 件
├── 日常可售 40 ← 期就在卖的
├── 秒杀预留 50 ← 20:00 开抢，此刻不能卖
├── 预售定金 20 ← 已付定金，待付尾款
└── 安全水位 10 ← 预留的缓冲，防超卖
```
如果你只维护一个数字 `120`，那么秒杀开始的瞬间，日常订单会把这 50 件吃掉，等到 20:00 开抢，库存是 0。
**所以库存不是数量，是承诺。**要按用途分池，而不是一个总数。
### 最容易崩的那个点：多平台同时卖同一批货
这是超卖的头号原因，也是最多人踩的：
```
实际仓库：120 件
淘宝设置：100 件
京东设置：100 件
→ 两边各卖 100，一共 200，超卖 80 件
```
平台之间不通气，谁都不知道对方卖了多少。你必须在**自己这边**做池化：
```
物理池 120
├── 淘宝额度 60
├── 京东额度 50
└── 缓冲 10 ← 谁先卖完可以借，但缓冲不能被借光
```
**缓冲池是关键**。它让你在某个平台爆单时还有得卖，同时给你时间调整分配。
### 代码
```python
# guard/stock.py
"""库存守护。两个职责：
1. 水位熔断——可售低于安全线就自动下架或限购
2. 池化校验——各平台额度之和不能超过物理池
注意：这里的 available 是平台可售数，reserved 是本系统已占用但
还没回写到平台的量。真实可售 = available - reserved。
这两个数分开存，是因为回写有延迟，混在一起会重复扣。
"""
import time
from base import alert
def real_available(db, item_id, platform_available, reserved):
"""扣掉本地占用后的真实可售"""
return max(0, int(platform_available) - int(reserved))
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def check_stock(db, item_id, platform_available, reserved, floor=10, action="off"):
"""水位检查。返回建议动作：normal / limit / off
floor: 安全水位。大促建议按日均销量的 0.5 倍设，不要凭感觉。
action: 触底后的动作，off=下架，limit=限购 1 件
"""
raw = int(platform_available) - int(reserved)
avail = max(0, raw)
db.execute(
"insert or replace into stock_level"
"(item_id, available, reserved, real, floor, updated_at) values(?,?,?,?,?,?)",
(item_id, platform_available, reserved, avail, floor, int(time.time())),
)
db.commit()
# 超卖和售罄必须分开报：售罄就是没货可卖，超卖是要走赔付预案的事故
if raw < 0:
alert(db, "P0", "STOCK_NEGATIVE",
f"{item_id} 超卖 {-raw} 件（平台 {platform_available} 本地占用 {reserved}），走赔付预案")
return "off"
if avail <= 0:
alert(db, "P0", "STOCK_OUT",
f"{item_id} 可售为 0（平台 {platform_available} 本地占用 {reserved}）")
return "off"
if avail <= floor:
# 大促期间这里必须是 P0：P1 会淹没在其他告警里没人看
alert(db, "P0", "STOCK_LOW",
f"{item_id} 可售 {avail} 已到水位 {floor}，建议{action}")
return action
return "normal"
def check_pool(db, item_id, physical, quotas):
"""池化校验：各平台额度之和不能超过物理库存
quotas: {"taobao": 60, "jd": 50, "pdd": 0}
超配是超卖的前置条件，必须在开卖前拦住，不是卖的时候才管。
"""
total = sum(quotas.values())
if total > physical:
alert(db, "P0", "POOL_OVERCOMMIT",
f"{item_id} 物理 {physical} 分配 {total}（{quotas}）超配 {total - physical}")
return False
return True
def replenish(db, item_id, qty, source, daily_sales=None):
"""库存回补：取消单、退款、超时未付款都要回补
但必须设上限：刷单党会反复下单占库存再取消，
没有上限的回补等于把库存交给对手。
daily_sales 传当日销量，上限按 20% 算；
拿不到销量就退回固定上限。**不能用 qty 自己算上限**——
那等于"回补多少就允许多少"，一点约束都没有。
"""
used = db.execute(
"select coalesce(sum(qty),0) from stock_back where item_id=? and day=?",
(item_id, time.strftime("%Y%m%d")),
).fetchone()[0]
cap = max(20, int(daily_sales * 0.2)) if daily_sales else 50
if used + qty > cap:
alert(db, "P1", "REPLENISH_ABNORMAL",
f"{item_id} 今日已回补 {used}，本次 {qty} 超上限 {cap}，来源 {source}")
return False
db.execute(
"insert into stock_back(item_id, qty, source, day, created_at) values(?,?,?,?,?)",
(item_id, qty, source, time.strftime("%Y%m%d"), int(time.time())),
)
db.commit()
return True
```
那个 `replenish` 的限流容易被忽略，但它是真事：大促期间同行恶意下单占库存再取消，能把你的秒杀库存锁死。回补不设上限，等于把库存交给对手。
## 五、链路三：订单——红线是漏单为零
### 大促期间它为什么不一样
订单链路的特殊性在于：**漏单不可恢复，重复可以吸收**。
- 重复拉一单 → `ON CONFLICT` 幂等掉，无损失
- 漏拉一单 → 客户付了钱，你没发货，等超时罚款才知道
所以整条链路的设计原则是：**宁可重复十次，不可遗漏一次**（第 3 篇那条铁律，在大促是生死线）。
大促会同时放大三个坑：
**坑一：分页漂移。** 你在翻第 3 页的时候，第 1 页又插入了新订单，整个列表往后挤。第 3 页的部分单会跑到第 4 页，被你跳过。
**坑二：时间窗边界。** 同一秒多单，用 `modified` 时间戳做游标，等于把同一秒的单全丢了。
**坑三：接口超时。** 大促期间超时率从 0.1% 涨到 5%，一次超时就是整页丢。
### 解法：双游标 + 单数对账
双游标（第 3 篇讲过）解决坑一和坑二：游标存 `(时间戳, 订单号)` 二元组，重叠加窗 60 秒。
真正解决"我到底漏没漏"的，是**单数对账**：
```python
# guard/order.py
"""订单对账。核心不是"拉",是"验"。
拉订单的代码到处都有，能回答"我拉全了吗"的代码很少。
方法：用平台的订单计数接口，和自己库里的数比。
不等就说明漏了,重拉整个时间窗。
"""
import time
from base import alert
def reconcile(db, platform, hour_start, local_count, remote_count):
"""按小时对账。
为什么按小时不按天：
按天对账，漏单要等到次日凌晨才发现，那时候客户已经投诉了。
按小时对账，最晚一小时发现，还有救。
"""
db.execute(
"insert or replace into order_gap"
"(platform, hour_start, local_count, remote_count, checked_at) values(?,?,?,?,?)",
(platform, hour_start, local_count, remote_count, int(time.time())),
)
db.commit()
if local_count == remote_count:
# 注意：0 比 0 也会走到这里。如果拉单任务本身死了，
# 两边都是 0，"对账通过"毫无意义。这种情况只能靠拉单任务的
# 心跳发现，不能靠对账。第一次跑之前务必确认拉单是活的。
return False
gap = remote_count - local_count
if gap > 0:
# 漏单，P0。重拉这个小时的全部订单，不要试图"补差额"——
# 你不知道漏的是哪几单，只能整窗重拉。
alert(db, "P0", "ORDER_GAP",
f"{platform} {_fmt(hour_start)} 本地 {local_count} 平台 {remote_count} "
f"漏 {gap} 单，需整窗重拉")
return True
# 本地比平台多：重复或脏数据，P1，可以慢慢查
alert(db, "P1", "ORDER_DUP",
f"{platform} {_fmt(hour_start)} 本地 {local_count} 平台 {remote_count} 多 {-gap}")
return False
def _fmt(ts):
return time.strftime("%m-%d %H:%M", time.localtime(ts))
def order_lag(db, limit=20):
"""订单延迟：从支付到进入本系统的时间差
这个指标比"成功率"重要。大促发货时限是硬指标，
延迟 30 分钟意味着你有 30 分钟的订单还不知道要发。
"""
rows = db.execute(
"select order_id, paid_at, created_at from orders "
"order by created_at desc limit ?", (limit,),
).fetchall()
lags = [c - p for _, p, c in rows if p and c > p]
if not lags:
return None
lags.sort()
p95 = lags[max(0, int(len(lags) * 0.95) - 1)]
return {"p50": lags[len(lags) // 2], "p95": p95, "max": lags[-1]}
def check_lag(db, warn=300, crit=900):
"""延迟超阈值告警。阈值按平台的发货时限倒推：
24 小时发货 → 延迟 15 分钟是危急（crit）
48 小时发货 → 30 分钟告警就够
"""
lag = order_lag(db)
if not lag:
return
if lag["p95"] > crit:
alert(db, "P0", "ORDER_LAG", f"订单延迟 P95 {lag['p95']}s 超过 {crit}s")
elif lag["p95"] > warn:
alert(db, "P1", "ORDER_LAG", f"订单延迟 P95 {lag['p95']}s 超过 {warn}s")
```
```
### 一个容易被忽略的点：订单链路的速度比完整性更容易崩
完整性靠对账能补。**速度补不回来。**
客户 10:00 付款，你 10:40 才拉到，那 40 分钟里你不知道要发什么货。大促发货时限按支付时间算，不是按你拉到的时间算。
所以订单链路要盯两个指标，不是一个：
| 指标 | 含义 | 大促红线 |
|---|---|---|
| 漏单率 | 本地单数 / 平台单数 | 必须 100%，差一单就是 P0 |
| 延迟 P95 | 支付到入库的时间 | 24h 发货时限 → ≤ 5 分钟 |
**只看漏单率会漏掉最痛的那个问题**：一单没漏，但都晚到 40 分钟，照样发不出去。
## 六、只剩 72 小时，先做哪条
现实一点：如果现在离大促只有三天，三条全做完不现实。按这个顺序砍：
| 优先级 | 做什么 | 工作量 | 为什么先做它 |
|---|---|---|---|
| **1** | 订单单数对账 | 2 小时 | 成本最低、收益最高。漏单是唯一"不可恢复"的事故 |
| **2** | 库存水位熔断 | 3 小时 | 超卖要赔付 + 平台处罚，且影响店铺权重 |
| **3** | 价格到手价核对 | 1 天 | 工作量最大（要拆三层），但错价的损失有上限，且能事后追 |
三条里只有第 1 条是"两小时就能上线"的——一个定时任务，拉平台单数，比本地单数，不等就告警。**如果你今天只能做一件事，做这个。**
价格核对放最后不是因为不重要，是因为它最费工：要把标价、券、满减三层都拆开，还要处理不同类目的规则差异。三天做不完，做一半反而危险（核对逻辑错了会误导运营）。
## 七、T-14 到 T+0 的时间表
这张表是给还没到大促的团队用的，存下来，618 还能用。
| 节点 | 该做什么 | 判断标准 |
|---|---|---|
| **T-14** | 三条链路打通 + 单链路压测 | 每条链路都能在峰值 QPS 下跑通，不含告警 |
| **T-7** | 全链路演练（模拟真实峰值 2 倍） | 演练中出现的每个告警都要有处置动作，不能只"知道了" |
| **T-3** | **代码冻结** | 只修 P0 bug，不加功能。这条最难执行，也最该执行 |
| **T-1** | 值守排班 + 降级预案确认 | 谁在 0 点看屏幕，出事了按哪张表操作，必须写下来 |
| **T+0** | 实时监控 + 手工兜底通道 | 系统挂了要能手导单、手改价，这条通道平时就要演练 |
T-3 的代码冻结值得单独说。大促前三天上线新功能，是事故的经典成因——新代码没有经历过峰值，你不知道它在哪里会崩，而这时候已经没有时间修了。**冻结不是保守，是止损。**
## 八、降级预案：出事时按这张表做，不要临场想
大促值守最怕的不是出事，是出事了大家在群里讨论"要不要关"。
提前把这张表写下来，贴在显示器上：
| 触发条件 | 动作 | 谁来决定 |
|---|---|---|
| 订单接口错误率 > 5%，持续 5 分钟 | 停止自动拉单，切备用数据源；同时开手工导单通道 | 值班开发直接执行，事后报备 |
| 订单延迟 P95 > 15 分钟 | 降低库存/价格核对频率，把配额让给订单 | 值班开发 |
| 库存对账失败（平台值 < 本地值 20% 以上） | 全店限购 1 件，人工核查 | 运营负责人 |
| 价格偏差 > 1% 且影响 SKU > 50 个 | 暂停所有自动调价，转人工 | 运营负责人 |
| 某个平台完全不可用 | 该平台商品全部下架（不是"等它恢复"） | 值班开发 + 运营 |
最后一条最反直觉：**平台挂了，第一反应应该是下架，不是等**。
因为平台挂了但你的商品还在卖，客户下单你收不到单，等平台恢复时你已经积累了一堆超时未发货。下架是止损，等待是赌博。
## 九、三条链路的共同点
写完这三段代码，你会发现它们长得几乎一样：
```python
def guard(expected, actual, threshold):
if abs(expected - actual) > threshold:
alert(...)
```
对，核心就是一个比对加一个告警。**大促的稳定性不是靠复杂架构，是靠"每个关键假设都有代码在验证"**。
三条链路对应三个假设：
| 链路 | 被验证的假设 | 假设错了会怎样 |
|---|---|---|
| 价格 | 客户看到的价 = 我算的价 | 资损，且不可追回 |
| 库存 | 平台可售 = 真实可售 | 超卖，赔付 + 处罚 |
| 订单 | 我拉到的单 = 平台所有的单 | 漏发，投诉 + 超时罚款 |
这三个假设，平时错了没人发现（因为量小、能人工兜），大促错了就是事故。
所以大促前该做的不是"优化性能"，是**把这三个假设写进代码，让它每小时替你看一遍**。
上面四段加起来三百多行，纯标准库加 SQLite，不用起 Redis 也不用上消息队列。如果你现在离大促只剩三天，从订单对账开始——两小时能上线，是这三件事里性价比最高的。
需要完整的表结构和告警落库代码，可以找我们要，也可以直接拿这套思路套你们自己的库。大促加油。未分类
服务商接单前先看：客户最常要的 8 个数据字段（附 python 源码）
万邦科技Lex 发表于2026-11-08 浏览27 评论0
# 服务商接单前先看：客户最常要的 8 个数据字段（附 python 源码）
接单最容易亏钱的地方不是报价报低了，是**字段没问清**。
我们踩过一次。客户说"我要商品的销量"，我们说"有"。交付那天他把数据接进选品系统，两周后打电话来：排序是乱的，销量高的排不上去。他要的是**近 30 天销量**，我们给的是**累计销量**——一个卖了三年但最近没人买的老品，累计销量五万，排在新品前面。
这个锅没法甩。合同里写的是"销量"，我们确实提供了"销量"。
后来我们把这件事做成了一张表：**客户说的字段名，和数据库里的字段，是两个东西，中间必须翻译一次**。这篇就把最常见的八个字段的翻译写出来，附一段可以跑的自检代码，接单前先跑一遍。
## 一、八个字段，三类落差
先说结论：这八个字段之所以容易出事，不是因为接口难，是因为它们在三个不同的地方让客户失望。
| 落差类型 | 客户什么时候失望 | 字段 |
| -------- | --------------- | -------------- |
| **口径落差** | 拿到数了，但和以为的不是一回事 | 价格、库存、销量 |
| **形态落差** | 拿到东西了，但塞不进他的系统 | SKU 属性、图片、物流编码 |
| **合规落差** | 拿到东西了，但不能随便用 | 收件信息、商品标题 |
口径落差最危险，因为**数据看起来是对的**。五万销量确实是销量，不会报错，不会异常，只有业务逻辑错了才发现。
形态落差最费工，报价时按"给个字段"算的，实际要做一次转换层。
合规落差最容易变成法律问题，也是最容易被服务商忽略的一类。
下面逐个说，每个字段给三样：客户通常怎么说、你该追问什么、最贵的那个坑。
## 二、口径落差：数是对的，但意思不对
### 1. 价格
客户说"价格"的时候，脑子里可能是六种东西之一：
| 客户说的 | 实际对应的字段 |
| ------ | ---------------------------- |
| 页面价格 | 标价（`price`） |
| 现在买多少钱 | 促销价 / 活动价（`promotion_price`） |
| 我实际付多少 | 到手价 = 促销价 − 店铺券 − 平台券 + 运费 |
| 会员多少钱 | 会员价（部分平台单独返回，多数要自己算） |
| 批发多少钱 | 1688 阶梯价，按起订量取档 |
| 含税多少钱 | 多数平台不含税，需要自己加 |
**该追问的三句**：
1. 含不含券？含不含运费？
2. 会员价要不要单独一列，还是并进到手价？
3. 是给用户看，还是给财务算账？（前者要到手价，后者要标价+优惠明细，能对得上账）
**最贵的坑**：促销价没有时效字段。客户按促销价算了一个月的利润，活动结束了数据还在返回促销价，利润表对不上，回来找你查一周。第 13 篇讲过价格要建模成带 `promo_end` 的对象，这是同一个问题的另一面——**服务商交付时如果只给一个数字，就必须同时给这个数字的有效期**。
### 2. 库存
客户说"库存"，实际可能是：
- **总库存**：仓库里有多少件
- **可售库存**：总库存 − 已占用 − 安全库存，这才是"还能卖几件"
- **锁定库存**：下单未付款占用的
- **在途库存**：采购中，还没入库
这四个数能差出好几倍。做活动备货的客户要总库存，做防超卖的客户要可售库存，两个数给错了都是事故。
**该追问的一句**：你要的是"还剩几件"，还是"还能卖几件"？
**最贵的坑**：1688 没有 SKU 库存这个概念。它的业务模型是按起订量阶梯卖，返回的是商品的**可售状态**，不是一个数字。你用淘宝的思维去接，适配器里会写一句"忽略 sku_id，返回总量"——**调用方以为拿到了 SKU 库存，实际上是错的，而且不报错**。第 8 篇把这叫"假装都支持"，在我看来这是服务商最容易犯的错：不是不会做，是不好意思说不支持。
### 3. 销量
这个字段水最深，因为三个平台给的东西根本不是一回事：
| 平台 | 通常返回 | 说明 |
| ---- | --------------------- | ----------------- |
| 淘宝 | 累计销量（`sold_quantity`） | 商品生命周期累计，老品数字很大 |
| 京东 | 评价数为主，销量部分类目不返回 | 很多接口根本不给你销量 |
| 1688 | 成交件数 / 成交笔数 | 批发的"笔"和零售的"件"完全不同 |
**该追问的一句**：这个销量你拿去干什么？
答案决定你该给什么：
- **排序/选品** → 用**评价增速**或**近 N 天增量**，比累计销量准得多。累计销量对老品有严重偏袒。
- **备货预测** → 必须近 30 天，而且要和退货率一起给，不然备多了。
- **给客户看**（比如"本店已售 10 万件"）→ 累计销量就够，甚至平台前台展示的那个数就行。
**最贵的坑**：**答应了给"月销"，实际平台只给累计**。这种承诺在售前最容易出口，因为客户问的时候你觉得"销量不就是销量"。交付时你只能自己算：`累计(t) − 累计(t−30d)`。这需要你从接单第一天就开始存历史快照——**如果你第三个月才想起来算，前三个月的数据永远补不回来**。
## 三、形态落差：东西给了，但塞不进去
### 4. SKU 属性
客户说"我要颜色和尺码两列"，听起来像是在表格上加两列。实际拿到的是这个：
```
1627207:28320:颜色分类:黑色;1627208:34117:尺码:M
```
这是淘宝的属性串格式：`属性ID:值ID:属性名:值`，多组用分号隔开。京东是另一套结构（对象数组），1688 又是另一套（带 `skuId` 和自定义属性）。
三个平台三种给法，这就是"必须做归一化"的全部理由——不是洁癖，是客户的 ERP 只认两列。
**该追问的两句**：
1. 要"颜色/尺码"这两列，还是要完整属性集（含材质、产地、执行标准）？
2. 属性值要不要标准化？客户系统里是"黑色"，平台给的是"黑色/深黑/曜石黑"，不统一就合并不了。
**最贵的坑**：**属性值不稳定**。同一个商品，这次返回"黑色"，下次商家改了叫"经典黑"，你的 SKU 匹配逻辑就断了。做组合装、做对账的客户会遇到"同一个 SKU 变成两个"。
### 5. 图片
客户说"给我图片链接"，他以为拿到 URL 就能用。实际有三个坑：
1. **防盗链**：1688 的图片直接外链会 403，必须转存到自己的存储
2. **尺寸**：主图有 800×800、1200×1200 多个规格，详情图是长图（可能 750×8000）
3. **有效期**：部分 CDN 链接带签名参数，过期就失效
**该追问的一句**：你是要**链接**，还是要**文件**？
要链接，你得保证这个链接在他那边能打开、能长期打开；要文件，你得做转存，这是存储成本和带宽成本，报价时必须算进去。
**最贵的坑**：交付了链接，客户上线一个月后图片全挂。原因是 CDN 签名过期。这种事不报错，就是图片一个个变成灰块，客户发现的时候往往是客户投诉商品页打不开。
### 6. 物流编码
客户说"物流状态"，实际拿到的是 `WAIT_SELLER_SEND_GOODS`、`STOP_ONE_STAGE`、`GOT` 这一堆东西，三个平台三套编码，同一个"已揽收"在三家可能叫三个名字。
更要命的是**节点的粒度不一样**：有的平台给你 7 个状态，有的给你 20 个子状态（运输中、到达分拨中心、离开分拨中心……）。
**该追问的一句**：你要几个状态？
做客服系统的客户要细（"包裹现在在哪"），做订单看板的客户要粗（"发出/在途/签收"）。给粗了客服答不了，给细了客户系统没有对应字段，会全部落到"其他"。
**最贵的坑**：**只给原始编码，不给映射后的标准状态**。客户得自己写映射表，写完还得跟着平台改。这个工作量不小，而且一旦映射错了，"已签收"被当成"在途"，自动确认收货的逻辑就永远不触发。
## 四、合规落差：拿到了，但不能随便用
### 7. 收件信息
这个字段是八个里唯一会让你承担法律责任的。
客户说"我要订单的收件人姓名、电话、地址"，多数平台返回的是**脱敏值**：
```
收件人：张*
手机：138****1234
地址：浙江省杭州市西湖区****
```
要拿到明文，得走**解密接口**，而且要单独申请权限、有调用额度限制、有审计日志。这不是技术门槛，是平台在履行个人信息保护义务。
**该追问的两句**：
1. 你要明文干什么？（发货打单可以，做用户画像不行）
2. 你有没有做过去标识化处理？
**最贵的坑**：服务商把明文收件信息落库、落日志、甚至缓存到 Redis 里，客户的数据泄露了，追责链条上是**你**。第 18 篇讲数据服务的风险传导，这一条是最短的那根引线：你手里有十万条真实手机号，出事的时候没有"我只是个技术供应商"这种说法。
建议做法：明文只在内存里过一遍，直接推给客户的打单接口，**不落库、不打日志**。日志里要有 `138****1234` 这种打码，而不是全量。
### 8. 商品标题
客户说"标题照搬过来就行"，你照搬了，然后被平台驳回。
标题里的雷有四个：
1. **极限词**："最"、"第一"、"国家级"——违反广告法，平台会拦
2. **平台痕迹**："淘宝爆款"、"天猫同款"、"京东专供"——跨平台铺货必挂
3. **促销词**："【限时秒杀】"、"满 199 减 50"——价格变了标题没变，属于虚假宣传
4. **他人商标**：品牌词没授权就写上去
**该追问的一句**：这个标题你是要**原样展示**，还是要**重新发布**？
原样展示（做比价、做监控）没问题，但要注意展示合规；重新发布（铺货）必须清洗，第 5 篇那套正则就是干这个的。
**最贵的坑**：清洗时**删词不删句**。"【爆款】不锈钢保温杯 500ml 大容量"，删掉"【爆款】"就好，别把整句重写——重写会丢规格词，500ml 变成"大容量"，客户那边同款匹配全乱。
## 五、代码：字段契约 + 填充率探测
上面八个字段的问题是"说不清"，代码解决的是"说不清就没法验收"。
思路：把每个字段写成一份**契约**（客户可能怎么叫、对应哪几个实际字段、口径是什么、填充率下限多少），然后拿真实数据跑一遍，输出一份可以直接发给客户的报告。
### 5.1 字段契约
```python
# contract.py
"""字段契约：接单前和客户确认的内容，落成代码。
每条契约四件事：
aliases 客户可能怎么叫（用于把需求清单映射到契约）
candidates 我方数据里可能对应的字段路径（按优先级）
semantic 口径说明——这句要写进合同附件
min_fill 填充率下限，低于此值必须在报价前告知客户
"""
FIELDS = {
"price": {
"label": "价格",
"aliases": ["价格", "售价", "单价", "页面价", "标价"],
"candidates": ["price", "sku.price", "sku_list[0].price"],
"semantic": "商品标价，不含优惠券、不含运费；活动价见 promotion_price",
"must_ask": ["含不含券", "含不含运费", "会员价是否单列"],
"min_fill": 0.99,
},
"promo_price": {
"label": "促销价",
"aliases": ["活动价", "到手价", "现价", "最低价"],
"candidates": ["promotion_price", "sku.promotion_price"],
"semantic": "活动期内的成交价，必须与 promo_end 一起交付",
"must_ask": ["活动结束时间要不要一起给"],
"min_fill": 0.60, # 没活动的商品本来就没有，60% 是正常的
},
"stock": {
"label": "库存",
"aliases": ["库存", "剩余", "可售", "数量"],
"candidates": ["stock", "quantity", "sku.quantity"],
"semantic": "可售库存（已扣占用与安全库存），非仓库总库存",
"must_ask": ["要总库存还是可售", "1688 无 SKU 库存，是否接受商品级"],
"min_fill": 0.95,
},
"sales": {
"label": "销量",
"aliases": ["销量", "月销", "已售", "成交"],
"candidates": ["sold_quantity", "sales", "volume"],
"semantic": "累计销量，非近 30 天；近 30 天需历史快照计算",
"must_ask": ["用来排序还是备货", "能否接受累计口径"],
"min_fill": 0.80,
},
"sku_props": {
"label": "SKU 属性",
"aliases": ["规格", "颜色尺码", "SKU", "属性"],
"candidates": ["skus", "sku_list", "props"],
"semantic": "已归一为 {spec_name: spec_value} 结构；原始属性串另存 raw_props",
"must_ask": ["要两列还是要完整属性集", "属性值是否需标准化"],
"min_fill": 0.90,
},
"images": {
"label": "图片",
"aliases": ["主图", "详情图", "图片", "素材"],
"candidates": ["images", "pic_url", "item_imgs"],
"semantic": "已转存至自有存储的直链，非平台原始 CDN 地址",
"must_ask": ["要链接还是要文件", "主图还是详情长图"],
"min_fill": 0.98,
},
"logistics_status": {
"label": "物流状态",
"aliases": ["物流", "轨迹", "快递状态"],
"candidates": ["status", "trace_list[-1].status"],
"semantic": "已映射为 5 个标准状态；原始编码见 raw_status",
"must_ask": ["要几个状态粒度", "要不要节点时间"],
"min_fill": 0.85,
},
"receiver": {
"label": "收件信息",
"aliases": ["地址", "收件人", "电话", "收货信息"],
"candidates": ["receiver_name", "receiver_mobile", "receiver_address"],
"semantic": "默认脱敏值；明文需单独授权，仅透传不落库",
"must_ask": ["要明文做什么", "是否接受脱敏"],
"min_fill": 1.00,
},
"title": {
"label": "商品标题",
"aliases": ["标题", "名称", "商品名"],
"candidates": ["title", "name"],
"semantic": "原始标题，含营销词与平台痕迹；用于重新发布需清洗",
"must_ask": ["原样展示还是重新发布"],
"min_fill": 0.99,
},
}
```
（八个字段，契约里为什么有九条：`price` 和 `promo_price` 是同一个字段的两种口径，拆开写是因为**客户说"价格"的时候，你必须先问清是哪一个**。）
### 5.2 填充率探测
这段是整篇最实用的部分。多数人接单只验证"接口通不通、字段在不在"，但真正让客户失望的是**字段在，值是空的**。
```python
# probe.py
"""拿真实样本跑一遍，输出填充率、口径分布、字段可用性。
用法：
python probe.py --file sample.json # 本地样本回放
python probe.py --file sample.json --md # 输出 markdown 报告
样本格式：一个 JSON 数组，每个元素是一个商品/订单的原始返回。
"""
import argparse
import json
from collections import Counter
from decimal import Decimal, InvalidOperation
from pathlib import Path
from contract import FIELDS
MISSING = object()
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def pick(obj, path):
"""按 "sku_list[0].price" 取值，取不到返回 MISSING（区别于 None）
为什么区分 MISSING 和 None：
key 不存在 = 我方没给这个字段（或没权限）
key 存在但值是 null = 平台没给（该商品不适用）
这两种在验收时是两回事，混在一起客服会答错。
"""
cur = obj
for seg in path.split("."):
if "[" in seg:
name, idx = seg.rstrip("]").split("[")
cur = cur.get(name, MISSING) if isinstance(cur, dict) else MISSING
if cur is MISSING:
return MISSING
try:
cur = cur[int(idx)]
except (IndexError, TypeError):
return MISSING
else:
if not isinstance(cur, dict) or seg not in cur:
return MISSING
cur = cur[seg]
return cur
def filled(v):
"""空值的判定比想象中麻烦：0 是有效值，空串不是"""
if v is MISSING or v is None:
return False
if isinstance(v, str):
return v.strip() != ""
if isinstance(v, (list, dict)):
return len(v) > 0
return True
def normalize_price(v):
"""价格比较必须用 Decimal，float 会把 19.9 和 19.90 判成不同"""
try:
return Decimal(str(v)).quantize(Decimal("0.01"))
except (InvalidOperation, TypeError, ValueError):
return None
def probe(samples, field_key):
spec = FIELDS[field_key]
stats = {p: 0 for p in spec["candidates"]}
samples_with_any = 0
values = Counter()
for row in samples:
hit = False
for path in spec["candidates"]:
v = pick(row, path)
if filled(v):
stats[path] += 1
hit = True
if field_key in ("price", "promo_price"):
d = normalize_price(v)
values[str(d) if d else str(v)] += 1
elif field_key == "sku_props" and isinstance(v, list):
values[f"{len(v)}个SKU"] += 1
if hit:
samples_with_any += 1
total = len(samples) or 1
fill = samples_with_any / total
# 只有命中率接近才叫"多口径并存"，否则就是主备关系
live = {p: c for p, c in stats.items() if c > 0}
return {
"label": spec["label"],
"fill": fill,
"ok": fill >= spec["min_fill"],
"min_fill": spec["min_fill"],
"hit": live,
"distinct": len(values),
"top_values": values.most_common(3),
"semantic": spec["semantic"],
"must_ask": spec["must_ask"],
}
def stability(a, b, field_key):
"""同一批样本拉两次，看字段值变了多少。
变了不一定是坏事（价格本来就该变），但要知道是哪个字段在变：
title / sku_props 频繁变 = 商家在改，你的匹配逻辑会断
price 变 = 正常，但交付时必须带更新时间
"""
spec = FIELDS[field_key]
path = spec["candidates"][0]
changed = 0
compared = 0
for x, y in zip(a, b):
vx, vy = pick(x, path), pick(y, path)
if not filled(vx) or not filled(vy):
continue
compared += 1
if vx != vy:
changed += 1
return changed / compared if compared else 0.0
def render(results, total_samples, n_days=None):
lines = [
f"# 字段可用性报告（样本 {total_samples} 条）",
"",
"| 字段 | 填充率 | 门槛 | 判定 | 备注 |",
"|---|---|---|---|---|",
]
for r in results:
mark = "通过" if r["ok"] else "**不足**"
if not r["ok"]:
note = "需在报价前告知客户"
elif r["distinct"] > 1:
note = f"{r['distinct']} 种取值" # 口径可能不唯一，值得问一句
else:
note = ""
if r.get("drift", 0) > 0.3:
note += f"；两次抽样差异 {r['drift']:.0%}"
lines.append(
f"| {r['label']} | {r['fill']:.1%} | {r['min_fill']:.0%} | {mark} | {note} |"
)
bad = [r for r in results if not r["ok"]]
if bad:
lines += ["", "## 需要提前说明的字段", ""]
for r in bad:
lines.append(f"**{r['label']}** 填充率 {r['fill']:.1%}，低于 {r['min_fill']:.0%}。")
lines.append(f"- 口径：{r['semantic']}")
lines.append(f"- 必须确认：{'；'.join(r['must_ask'])}")
lines.append("")
lines += ["", "## 各字段口径", ""]
for r in results:
lines.append(f"- **{r['label']}**：{r['semantic']}")
return "\n".join(lines)
def _pad(text, width):
"""中文占两格。str.ljust 按字符数补空格，表格会歪，必须按显示宽度算"""
w = sum(2 if ord(c) > 127 else 1 for c in text)
return text + " " * max(0, width - w)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--file", required=True, help="样本 JSON 文件")
ap.add_argument("--compare", help="第二次抽样，用于比对字段漂移")
ap.add_argument("--md", action="store_true", help="输出 markdown 报告")
ap.add_argument("--out", default="field_report.md")
args = ap.parse_args()
samples = json.loads(Path(args.file).read_text(encoding="utf-8"))
if not isinstance(samples, list) or not samples:
raise SystemExit("样本必须是 JSON 数组，且不能为空")
results = [probe(samples, k) for k in FIELDS]
if args.compare:
later = json.loads(Path(args.compare).read_text(encoding="utf-8"))
for r, k in zip(results, FIELDS):
r["drift"] = stability(samples, later, k)
if args.md:
Path(args.out).write_text(render(results, len(samples)), encoding="utf-8")
print(f"已写出 {args.out}（{len(samples)} 条样本）")
return
print(_pad("字段", 14) + _pad("填充率", 10) + _pad("门槛", 8) + "判定")
print("-" * 46)
for r in results:
flag = "OK" if r["ok"] else "LOW"
line = (_pad(r["label"], 14) + _pad(f"{r['fill']:.1%}", 10)
+ _pad(f"{r['min_fill']:.0%}", 8) + flag)
if "drift" in r:
line += f" 漂移 {r['drift']:.0%}"
print(line)
if not r["ok"]:
print(f" └ {r['semantic']}")
if __name__ == "__main__":
main()
```
```
### 5.3 怎么用
三步，接单前跑完，只要半天：
```bash
# 1. 从客户的目标商品里抽 200 条，存成 sample.json
python dump_samples.py --shop shop_123 --limit 200 > sample.json
# 2. 跑一遍，先看哪些字段填不满
python probe.py --file sample.json
# 3. 隔一天再抽一次，看哪些字段在漂（可选，但强烈建议）
python probe.py --file sample.json --compare sample_day2.json
# 4. 生成报告，直接发给客户确认
python probe.py --file sample.json --md --out 字段确认单.md
```
输出大概是这样：
```
字段 填充率 门槛 判定
----------------------------------------
价格 100.0% 99% OK
促销价 62.5% 60% OK
库存 98.5% 95% OK
销量 41.0% 80% LOW
└ 累计销量，非近 30 天；近 30 天需历史快照计算
SKU 属性 93.0% 90% OK
图片 99.5% 98% OK
物流状态 86.0% 85% OK
收件信息 100.0% 100% OK
商品标题 100.0% 99% OK
```
那个 41% 就是这篇的价值。**如果你在报价前跑出这个数，就可以跟客户说"销量这个字段有一半商品拿不到，我们建议用评价增速替代，或者加价做历史快照"**。如果等交付了才发现，客户只会记住"你说有销量的"。
顺带一个观察：跑完这份报告，你会发现自己对数据的了解比读三天文档还多。文档告诉你字段存在，报告告诉你字段**有值**。
## 六、接单前该给客户的三张纸
代码解决的是"我知道"，三张纸解决的是"客户知道"。少一张，后面就会有一次扯皮。
**第一张：字段确认单**
就是上面代码生成的那份报告，每个字段要写清四件事：口径（一句话）、示例值（真的从数据里取的）、填充率、更新频率。**示例值必须是真的**，从样本里截一段；自己编的示例，客户拿去验证会发现对不上。
**第二张：不保证清单**
这张最容易被跳过，也最保命。写清楚哪些事不承诺：
- 平台字段变更不另行通知，但我们会在 24 小时内适配
- 部分类目的可选属性，不保证有值
- 历史数据只能从接入日开始（除非另行采购）
- 平台侧的限流、审核、政策变化导致的延迟，不算 SLA 违约
**第三张：验收标准**
不要写"以客户满意为准"，要写成可测的：
- 字段准确率 ≥ 99.5%（抽 50 条人工比对）
- 主键准确率 100%（`item_id` / `order_id` 不能错，错了就是数据串店）
- 填充率达到确认单标注的门槛
- 数据延迟 ≤ N 分钟（P95）
这三张纸加起来不到两页，但它们把"你当时说的是这个意思"这类争执，提前到签约之前解决。
## 七、最后一句
这八个字段里，没有一个技术上难。它们难在**客户用的词和系统里的字段之间，隔着一层业务理解**。
服务商真正卖的不是接口访问权，是这层翻译。翻译得好，客户觉得你懂他的业务；翻译得不好，你会一直在解释"我给你的确实是销量"。
而翻译这件事，是可以提前做、并且用代码验证的。上面那段 `probe.py` 两百行出头，但它比任何一份字段文档都更能避免交付期的争执——**因为它测的不是字段存不存在，是字段有没有值，值是什么口径**。
如果你正在评估要不要自己接这三个平台，或者手上有一批客户需求要报，可以拿这套契约模板去套一遍。跑完你会知道哪些能承诺、哪些要提前说清——这个清单本身就是报价的依据。
有具体的字段拿不准口径，也可以直接问我们，我们把三平台的对照表摊开一起看。未分类
刚起步的电商小团队，第一批该接哪几个接口（附 Python 源码）
万邦科技Lex 发表于2026-11-07 浏览43 评论0
# 刚起步的电商小团队，第一批该接哪几个接口（附 Python 源码）
> 起步团队最容易犯的错，是把「接口清单」当成任务清单，一条条往下接。
>
> 三平台文档里有上千个接口，你真正需要的第一批只有 **4 个**。剩下的不是不重要，是**现在接它们会拖慢你赚到第一笔钱的时间**。
>
> 这篇给一个明确的顺序：第一批 4 个、第二批 3 个、第三批 4 个，以及三个「现在千万别接」的。判断标准只有一条——**这个接口能不能让货卖出去、或者让钱收回来**。
>
> 代码是一个 150 行的单文件，跑完就是一条完整闭环：商品上架 → 客户下单 → 订单进库 → 扣库存 → 日终对账。
## 一、先定一个判断标准
接口的价值不是「技术上重要」，是**它对收入的贡献**。按这个标准，起步期能碰的接口只分三类：
| 类型 | 作用 | 起步期优先级 |
| ------- | --------- | ---------- |
| **换收入** | 商品能上架、能卖 | 最高，第一批 |
| **保现金** | 订单不漏、账对得上 | 最高，第一批 |
| **补体验** | 物流可查、售后顺畅 | 第二批，等有单了再说 |
| **纯优化** | 评价、竞品、素材 | 第三批，或者永远不接 |
一个反例：很多团队第一批就去接「物流轨迹查询」，理由是「客户会问快递到哪了」。但起步期你一天可能就 5 单，客户问了你手动查一下，30 秒的事。**为一天 5 次的查询，写一套订阅 + 去重 + 状态机，是典型的提前优化**（第 3 篇那套方案是对的，但不是现在该做的）。
## 二、第一批：4 个接口，按这个顺序接
| 顺序 | 接口 | 为什么在这个位置 |
| -- | ----------------------- | --------------------------------------- |
| ① | **商品详情**（含 SKU、标价、可售库存） | 没商品就没得卖。三平台的详情接口都一次返回 SKU 列表和标价，别拆成四次请求 |
| ② | **订单列表**（按更新时间增量） | 收入的证据。也是「有没有单」的唯一可靠来源 |
| ③ | **订单详情**（含收货地址、支付时间） | 发货必需。**列表接口通常不含完整地址**，这个坑第一天就会撞上 |
| ④ | **库存**（可售库存） | 防超卖。但起步期用「简单账本」撑住就行，别做实时同步 |
### 为什么「先订单、后库存」
这是这篇最反直觉的一条，值得单独说。
按技术难度排，库存比订单简单（一个数字 vs 一堆状态）。但按**收益**排，订单在前：
- **订单**接进来，你立刻能发货、能算账、能知道今天赚了多少。这是正反馈。
- **库存**接进来，你防的是「超卖」。而起步期 SKU 少、单量小，超卖的实际损失远小于「商品还没上架」的损失。
所以正确的做法是：**库存先用静态值撑住，下单即扣本地账本，宁可少卖**。具体就是三行逻辑：
```
可卖数 = 平台可售 − 本地已占 （查询时算，不写库）
下单 → 够就 reserved += n，不够就拦住这笔单
日终 → 平台库存重新拉过之后，reserved 归零重算
```
这么做会偶尔少卖（账面比实际紧），但不会超卖，而且**没有并发问题**——因为你从来不做「实时同步」这件难事。
等单量上来了（比如日单过百，或者开始做多仓），再把第 3 篇那套 `total - reserved - safety` 加上事件校准换上。那时候你已经有收入了，花两周做对它是合理的。
反过来做的团队我见过不少：花两周把库存并发扣减做对，商品还没上架，一分钱没赚。**技术正确，商业上错了。**
## 三、第二批：有第一笔订单之后
大概在第一笔订单之后的 2~4 周，接这三个：
| 接口 | 触发条件 | 不接的代价 |
| ----------- | --------------------- | ---------------------- |
| **物流轨迹** | 客户开始问「到哪了」，且每天超过 10 次 | 客服时间被吃掉，但还不致命 |
| **售后 / 退款** | 出现第一笔退款 | **账永远对不上**，你会以为是接口有问题 |
| **发货回写** | 你做代发，或对接了 ERP | 只能在平台后台手工点发货，一天几十单就撑不住 |
退款这个特别值得说：起步团队常常只拉「已付款」的订单，退款不管。结果月底一算，GMV 和实际到账差一截，财务来问，你查三天。
**退款不是异常，是正常业务**。它该在第二批，不该在「以后再说」。
## 四、第三批：月订单过千再看
| 接口 | 什么时候值得 |
| --------- | --------------------------------- |
| 活动价 / 优惠券 | 你开始做促销，且发现「到手价」算不对（第 13 篇的价格对象建模） |
| 类目属性 | 你开始批量铺货、被审核驳回（第 5 篇） |
| 评价 | 你要做选品或口碑分析（第 16 篇） |
| 素材 / 图片 | 你要做自动生成主图、多平台分发 |
这四个都是「业务推着你接」，不是「文档里有所以接」。
## 五、三个现在千万别接的
**① TMC 消息推送。**
听起来很美：订单来了实时推给你。实际要维护长连接、断线重连、消息确认、重复消费——四件事都能坏，坏了都要你查。起步期用「5 分钟轮询 + 时间窗重叠」完全够，而且失败就是这一轮没跑，下一轮补上，心智负担接近于零。
等你日单过千、或者对时效真的有要求了再换。切换成本是一次可控的重构，不是重写。
**② 批量 / 异步接口。**
批量接口的特点是：提交任务 → 拿 task_id → 轮询结果 → 处理部分失败。这套东西要状态机、要重试、要存储中间态。起步期你的商品可能就 200 个，循环调 200 次详情，10 分钟跑完，够用。
**③ 全站 / 竞品数据。**
先把自己店的货卖好。比价、竞品监控是第二阶段的事（第 16 篇），而且它需要的数据源和自营接口是两套逻辑（联盟接口 / 聚合服务），混在一起做会让你的第一版架构变形。
## 六、代码：150 行，一条闭环
起步版的原则：**小到一屏能看懂**。不要配置层、不要抽象层、不要依赖注入——这些东西等你有了第二个平台再谈。
```python
# mvp.py —— 起步版：4 个接口，1 条闭环
# 依赖：pip install httpx
import asyncio
import sqlite3
import time
import httpx
BASE = "https://api.example.com" # 换成你的网关
KEY = "your_app_key"
SECRET = "your_app_secret"
SHOP = "shop_001"
SCHEMA = """
pragma journal_mode = wal;
create table if not exists items(
item_id text primary key, title text, price real,
stock int, skus text, updated_at int);
create table if not exists orders(
tid text primary key, status text, pay_time int,
amount real, receiver text, updated_at int);
create table if not exists ledger(
item_id text primary key, reserved int default 0, last_sync int);
"""
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def call(client, path, **params):
"""一个函数搞定所有接口：拼参数、判错、返回 data"""
r = await client.get(path, params={"shop_id": SHOP, **params})
r.raise_for_status()
body = r.json()
if body.get("code") not in (0, 200, "0", None):
raise RuntimeError(f"{body.get('code')} {body.get('msg')}")
return body.get("data", body)
async def pull_items(client, db):
"""① 商品详情：一次带回 SKU、标价、可售库存。
三平台详情接口都支持，别拆成 SKU 接口 + 价格接口 + 库存接口"""
ids = [r[0] for r in db.execute("select item_id from items")]
if not ids: # 第一次：从列表接口拿 ID
lst = await call(client, "/v1/item/list", page_size=100)
ids = [it["item_id"] for it in lst.get("items", [])]
for item_id in ids:
d = await call(client, "/v1/item", item_id=item_id)
skus = d.get("skus") or []
stock = sum(int(s.get("stock") or 0) for s in skus) if skus \
else int(d.get("stock") or 0)
db.execute(
"insert into items(item_id,title,price,stock,skus,updated_at) "
"values(?,?,?,?,?,?) on conflict(item_id) do update set "
"title=excluded.title, price=excluded.price, stock=excluded.stock, "
"skus=excluded.skus, updated_at=excluded.updated_at",
(item_id, d.get("title", ""), float(d.get("price") or 0),
stock, str(len(skus)), int(time.time())))
db.commit()
return len(ids)
async def pull_orders(client, db, since: int):
"""② 订单列表：按更新时间增量，重叠加窗（第 3 篇的边界坑）"""
start = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(since - 600))
page, got = 1, 0
while True:
lst = await call(client, "/v1/orders",
updated_start=start, page=page, page_size=100)
rows = lst.get("orders", [])
if not rows:
break
for o in rows:
db.execute(
"insert into orders(tid,status,pay_time,amount,updated_at) "
"values(?,?,?,?,?) on conflict(tid) do update set "
"status=excluded.status, updated_at=excluded.updated_at",
(o["tid"], o["status"], int(o.get("pay_time") or 0),
float(o.get("payment") or 0), int(time.time())))
got += 1
page += 1
db.commit()
return got
async def fill_receiver(client, db):
"""③ 订单详情：列表接口没有地址，发货前补。
只补「待发货」的，别全量拉——这一步最容易把额度烧光"""
todo = db.execute(
"select tid from orders where receiver is null and status='WAIT_SELLER_SEND'"
).fetchall()
for (tid,) in todo:
d = await call(client, "/v1/order", tid=tid)
addr = d.get("receiver") or {}
db.execute("update orders set receiver=? where tid=?",
(f"{addr.get('name','')}|{addr.get('mobile','')}|"
f"{addr.get('address','')}", tid))
db.commit()
return len(todo)
def available(db, item_id: str) -> int:
"""④ 可卖数 = 平台可售 − 本地已占。
查询时算，不写库：`items.stock` 是平台镜像，每次拉取覆盖；
本地占用量单独记在 ledger，两者相减才是能卖的数。
不写回的好处是不存在「谁覆盖谁」的问题——第 3 篇那个
「校准取 min」的原则，是给「要把库存写回平台」的场景用的，
起步期我们不写回，所以用不上。"""
item = db.execute("select stock from items where item_id=?",
(item_id,)).fetchone()
if item is None:
return 0
row = db.execute("select reserved from ledger where item_id=?",
(item_id,)).fetchone()
return max(0, item[0] - (row[0] if row else 0))
def reserve_stock(db, item_id: str, num: int = 1) -> bool:
"""下单占库存：够就扣，不够就拦住这笔单（宁可少卖，不超卖）。
注意先判后扣——先扣再判，被拦下的单也会留下占用，账面越推越紧。"""
if available(db, item_id) < num:
return False
row = db.execute("select reserved from ledger where item_id=?",
(item_id,)).fetchone()
if row is None:
db.execute("insert into ledger(item_id,reserved,last_sync) "
"values(?,?,?)", (item_id, num, int(time.time())))
else:
db.execute("update ledger set reserved=reserved+?, last_sync=? "
"where item_id=?", (num, int(time.time()), item_id))
db.commit()
return True
def reset_reserved(db):
"""日终：平台库存已经在 pull_items 里重新拉过，本地占用量归零重算。
必须在拉取成功之后调，否则会把当天的占用抹掉——那就是超卖。"""
n = db.execute("update ledger set reserved=0 where reserved>0").rowcount
db.commit()
return n
async def main():
db = sqlite3.connect("mvp.db")
db.executescript(SCHEMA)
async with httpx.AsyncClient(base_url=BASE, timeout=10,
headers={"X-Secret": SECRET, "X-Key": KEY}) as c:
n_items = await pull_items(c, db) # 先拉最新库存
n_orders = await pull_orders(c, db, int(time.time()) - 3600)
n_addr = await fill_receiver(c, db)
n_reset = reset_reserved(db) # 拉成功之后才清零
print(f"商品 {n_items} · 订单 {n_orders} · 补地址 {n_addr} · 归零 {n_reset}")
if __name__ == "__main__":
asyncio.run(main())
```
这 150 行就是起步版的全部。它故意没有的东西，是按重要性排的：
| 没有 | 为什么现在不加 | 什么时候加 |
| --------- | ------------------- | ---------------- |
| 配置层（YAML） | 5 个常量改起来比读配置快 | 有第二个部署环境 |
| 多平台适配器 | 先接一个平台，等第二个出现重复代码 | 第二个平台接的时候（第 8 篇） |
| 并发 / 限流 | 200 个商品串行跑 4 分钟，能接受 | 商品上千，或单量上来 |
| 重试 / 熔断 | 失败就下一轮再跑，起步期能忍 | 开始有客户依赖你的数据 |
| 消息推送 | 轮询 5 分钟够用 | 日单过千 |
## 七、验收标准：一条能跑完的闭环
「接口接完了」不等于「能用了」。第一批的正确验收标准是一个动作：**把你自己的一个商品，从头到尾跑一遍**。
1. 商品信息从接口拉下来，落到你的库，页面能展示
2. 用另一个账号在你店里下一单（1 元商品，自己买）
3. 轮询把订单拉进来，地址补全，状态是「待发货」
4. 本地库存被扣掉 1，页面显示的可售数减 1
5. 你在平台后台点发货，回写状态（第二批才有，现在手工）
这五步跑通，第一批就算完成。**注意第 2 步必须真下单**——只用接口返回的模拟数据，你永远不知道字段口径对不对。1 元的成本，换来的是「整条链路真的通」，这是起步期最划算的一笔投入。
跑完你会发现至少两个问题：某个字段为空（通常是权限没开）、某个时间格式不对（时区）。这两个坑每个团队都会踩，早踩比上线后踩好。
## 八、起步期三个最常见的错误
**① 三个平台一起接。**
看起来平行推进效率高，实际是把「三个平台各自的怪癖」同时砸在你头上。第 2 篇那个坑表里的东西——1688 没有 SKU 库存、京东返回体多套一层、淘宝错误藏在 body 里——同时遇到三个，你会怀疑人生。
正确顺序：**一个平台跑通闭环 → 再复制到第二个**。第二个平台的成本大约是第一个的 1/3，因为业务逻辑已经验证过了。
**② 一开始就做增量和高并发。**
起步期商品 200 个、订单一天 20 个，全量拉一遍 4 分钟。这个阶段做增量调度、做并发优化，是把第 7 篇那套 60 倍提速用在 4 分钟上——**优化一个你根本感知不到的问题**。
**③ 一开始就自建。**
第 12 篇算过账：自建爬虫三年 37 万，官方 API 17 万。起步团队还有一笔更贵的隐藏成本：**你在写同步代码的那两周，没有在卖货**。
起步期最该花钱买的就是数据接口，因为它把你从「修管道」里解放出来，去做只有你能做的事——选品、内容、投放。这笔账在收入为零的时候最容易被算错。
## 九、时间预算
按这篇的顺序做，第一周（约 20 小时）应该能到：
| 天 | 内容 | 小时 |
| ----- | ---------------- | -- |
| D1 | 申请、拿 key、跑通商品详情 | 4 |
| D2 | 商品落库 + 页面展示 | 4 |
| D3-D4 | 订单增量 + 地址补全 + 幂等 | 8 |
| D5 | 库存账本 + 日终校准 | 4 |
对比「全都接」的版本：三平台 + 物流 + 售后 + 推送，两个月起步，而且大概率在第二个月放弃。
**起步团队真正的资源不是技术能力，是「还没被复杂度拖住」的那段时间。**
---
这一篇放在系列里的位置比较特殊：前面二十一篇讲的是「怎么把系统做对」，这篇讲「什么时候该做多少」。如果你现在只有一个店、一天十几单，上面 150 行代码就够了——第 3 篇的库存账本、第 7 篇的并发、第 9 篇的五道关，都等你需要的时候再看，那时候你会看得更懂。未分类
代运营公司怎么管 20 个店铺的数据？我们是这样做的（附 Python 源码）
万邦科技Lex 发表于2026-11-07 浏览45 评论0
# 代运营公司怎么管 20 个店铺的数据？我们是这样做的（附 Python 源码）
> 代运营的数据问题和卖家不一样。
>
> 卖家的问题只有一个店：取到数据、别出错、别断。代运营的问题是一堆店：**授权会过期、数据不能串、客户要各自的报表、一个店出问题不能影响其他十九个**。
>
> 我们踩过的坑基本都在这四件事上，跟接口本身关系不大。20 个店铺、47 个有效授权（有的店只在两个平台开店）、三个平台三套授权体系——这套东西现在是五个文件的代码在跑，人工每天 10 分钟看一张汇总表。
>
> 这篇把授权、隔离、汇报三块拆开讲，代码可以直接拿去改。
## 一、先把「店铺」当成一级实体
大部分多店铺系统出问题，根子在建模：**把店铺当成了配置，而不是资源**。
配置是没有生命周期的——写进去就不管了。但一个店铺授权是有生命周期的：客户签进来 → 授权 → 用着 → 快到期 → 客户换人/改密码 → 授权失效 → 客户走了 → 撤权。**每一步都有状态，每一步都需要代码处理**。
正确的建模是两层：
```
店铺（shop）
└── 授权（shop_id + platform） ← 唯一的凭证单元
```
20 家店 × 3 平台，不是 20 条记录，是 47 个授权（有 7 家店只在两个平台开店）。凭证表的主键必须是 `(shop_id, platform)`，不能是 `shop_id`——**否则一家店在三个平台的 token 只能存一个，后写的覆盖先写的，你会以为授权好了，其实只好了最后一个平台**。这个 bug 我们见过不止一次。
授权有五个状态，别用布尔值糊过去：
| 状态 | 含义 | 系统该做什么 |
| ---------- | ------------- | ------------ |
| `pending` | 已建店，授权还没完成 | 催授权，不进调度队列 |
| `active` | 正常 | 正常调度 |
| `expiring` | 剩余 < 48 小时 | 提前预警，任务照跑 |
| `expired` | 已过期 | 停止调度，标记客户待处理 |
| `revoked` | 客户主动解绑 / 平台收回 | 停止调度，通知商务 |
`expiring` 这个状态是最值钱的一个。**代运营每月有 3~5 个授权到期**，等过期了再处理，意味着那家店的日报当天就断了，客户第二天早上会来问。提前 48 小时预警，实际处理时间是两天，从容得多。
## 二、token 存储：加密与否，是责任问题
47 个店铺的 refresh token 躺在数据库里，这就是 47 张店铺的钥匙。数据库被拖、备份文件进了 git、开发同学的笔记本被偷——任何一个环节泄露，客户的全店数据就没了。
**这不是「应该加密」，是「必须加密」**，而且要在写代码的第一天做，后面补的成本高得多（要迁移存量、要处理轮换）。
```python
# ops/vault.py
import base64
import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
class Vault:
"""token 加解密：AES-GCM，密文入库，密钥永远不进数据库。
keys 是 {key_id: 32字节密钥}，current 是当前用来加密的 key_id。
存 key_id 的目的只有一个：轮换。换密钥时老密文还能用老 key 解开，
解密成功的记录顺手用新 key 重新加密，慢慢滚过去。
"""
def __init__(self, keys: dict[str, bytes], current: str):
self.keys = keys
self.current = current
for k in keys.values():
if len(k) != 32:
raise ValueError("AES-256 需要 32 字节密钥")
def seal(self, plaintext: str) -> tuple[str, str, str]:
"""返回 (密文, nonce, key_id)，三列一起入库"""
nonce = os.urandom(12)
ct = AESGCM(self.keys[self.current]).encrypt(
nonce, plaintext.encode("utf-8"), None)
return (base64.b64encode(ct).decode(),
base64.b64encode(nonce).decode(),
self.current)
def open(self, ct_b64: str, nonce_b64: str, key_id: str) -> str:
return AESGCM(self.keys[key_id]).decrypt(
base64.b64decode(nonce_b64),
base64.b64decode(ct_b64),
None).decode("utf-8")
```
密钥从哪来？三个档，按你的规模选：
| 方式 | 适合 | 注意 |
| -------------- | ---------- | --------------------------- |
| 环境变量 | 小团队、单机 | 进程能读到，宿主机被拿就没了；至少别写进代码和 git |
| 云 KMS / 密钥管理服务 | 十家店以上、多机部署 | 每次解密一次网络调用，要配本地短缓存（5 分钟） |
| 硬件加密机 | 有等保要求 | 代运营基本用不上，成本不划算 |
我们用的是云 KMS + 进程内 5 分钟缓存。缓存的是**解密后的 token**，不是密钥——这个区别很重要，密钥永不落地。
**一个容易漏的地方**：`refresh_token` 也要加密，而且它比 `access_token` 更重要。`access_token` 两小时就过期了，`refresh_token` 能换出无数个新的。只加密前者不加密后者，等于没加密。
## 三、并发刷新撞车：多店铺场景的第一次翻车
单个店铺的时候你不会遇到这个问题。20 个店铺、多个任务并发跑的时候，它一定会来。
场景长这样：某个店的 token 在 10:00:00 过期。10:00:01，订单同步任务和商品同步任务同时发现过期，同时发起刷新。平台的处理是**后刷作废先刷**（京东和 1688 都有这个行为），于是：
- 任务 A 拿到的 token 已经作废了 → 401
- 任务 B 拿到的 token 有效 → 继续跑
- 更糟的情况：两个都拿到「刷新成功」，但只有一个真正有效，另一个在下一次调用时才暴露
结果就是「同一个店，一半任务成功一半失败，重试之后好了」——这种故障最难查，因为它不稳定复现。
解法是**单飞（single-flight）**：同一个 `(shop_id, platform)` 在同一时刻只能有一个刷新在跑，其他协程等它的结果。
```python
# ops/refresher.py
import time
# 抢锁靠这一行 update：只有拿到了刷新权的那个协程，rowcount 才是 1
CLAIM = """
update grants
set locked_until = :until
where shop_id = :shop_id and platform = :platform
and (locked_until is null or locked_until < :now)
"""
def ensure_token(db, vault: Vault, shop_id: str, platform: str,
refresh_fn, ahead: int = 300) -> str:
"""返回可用的 access_token。
ahead：提前多少秒视为「需要刷新」。别卡着过期点刷，
网络一来一回加上平台处理，卡点必翻车。
"""
for _ in range(3): # 最多等三轮，够覆盖一次正常刷新
row = db.execute(
"select access_token, nonce, key_id, refresh_token, "
"refresh_nonce, refresh_key_id, expire_at, state "
"from grants where shop_id=? and platform=?",
(shop_id, platform)).fetchone()
if row is None:
raise LookupError(f"未找到授权 {shop_id}/{platform}")
(ct, nonce, key_id, rct, rnonce, rkey_id, expire_at, state) = row
if state in ("revoked", "expired"):
raise PermissionError(f"授权不可用 {shop_id}/{platform} state={state}")
if expire_at > time.time() + ahead:
return vault.open(ct, nonce, key_id)
now = int(time.time())
cur = db.execute(CLAIM, {"until": now + 30, "shop_id": shop_id,
"platform": platform, "now": now})
db.commit()
if cur.rowcount == 0:
# 别人正在刷，等一下再读它刷好的结果
time.sleep(0.3)
continue
try:
plain_rt = vault.open(rct, rnonce, rkey_id)
new_token, expires_in, new_rt = refresh_fn(plain_rt)
n_ct, n_nonce, n_kid = vault.seal(new_token)
r_ct, r_nonce, r_kid = vault.seal(new_rt or plain_rt)
db.execute(
"update grants set access_token=?, nonce=?, key_id=?, "
"refresh_token=?, refresh_nonce=?, refresh_key_id=?, "
"expire_at=?, state='active', locked_until=null, "
"updated_at=? where shop_id=? and platform=?",
(n_ct, n_nonce, n_kid, r_ct, r_nonce, r_kid,
int(time.time()) + expires_in, now, shop_id, platform))
db.commit()
return new_token
except Exception as e:
# 刷新失败：不清旧 token，只标记状态。旧的可能还能用一会儿，
# 清掉就是立刻断服——代运营断一家店，客户当天就知道。
db.execute(
"update grants set locked_until=null, last_error=?, "
"state=case when expire_at < ? then 'expired' else state end "
"where shop_id=? and platform=?",
(repr(e)[:300], int(time.time()), shop_id, platform))
db.commit()
raise
finally:
# 释放锁。用条件更新而不是直接置 null：
# 万一本进程卡了很久、锁已被别人重新抢到，别把别人的锁解掉
db.execute(
"update grants set locked_until=null "
"where shop_id=? and platform=? and locked_until <= ?",
(shop_id, platform, now + 30))
db.commit()
raise TimeoutError(f"等待刷新超时 {shop_id}/{platform}")
```
三个细节值得说：
1. **抢锁用的是 `update ... where locked_until is null or < now`**，没有引入 Redis。行级条件更新本身就是原子的，`rowcount == 1` 就是拿到锁。多店铺场景下这比引入一套分布式锁简单，也少一个故障点——除非你多机部署，那就换 Redis 或 PG 的 `select ... for update`。
2. **`locked_until` 是超时时间，不是布尔锁**。进程被 kill、机器重启，锁不会永久卡住——30 秒后自动可抢。用布尔锁的人一定会遇到「一次 OOM 之后这个店再也刷不了 token」。
3. **刷新失败不清旧 token**。这条和第 20 篇单人版一致，但理由不同：单人版是「没空管」，代运营是「断一家店客户当天就知道」。让旧 token 多活一会儿，比立刻 401 好。
## 四、隔离：一个店挂掉，不能带走其他十九个
这是多店铺和单店铺最本质的区别。20 家店的调度如果写成这样：
```python
for shop in shops:
await sync(shop) # 一家超时 60 秒，20 家就是 20 分钟
```
那么任何一家店出问题——授权过期、类目权限被收回、平台侧限流——整轮任务就停在那里。更隐蔽的是额度：**20 家店共用一个令牌桶，一家店做大促把额度烧光，其他 19 家的日报全部延迟**。
三件事必须按店铺拆开：
```python
# ops/fanout.py
import asyncio
class ShopWorker:
"""每个店铺一个独立上下文：独立限流桶、独立退避、独立错误计数。
这就是第 9 篇「舱壁隔离」在代运营场景的具体形态。"""
def __init__(self, shop_id: str, gate, backoff_s: float = 0):
self.shop_id = shop_id
self.gate = gate # 按 (shop, platform, api) 分桶
self.backoff_s = backoff_s
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def run_round(workers: list[ShopWorker], task_fn, timeout: float = 300):
"""一轮调度：所有店铺并发，互不阻塞，单店异常只影响自己"""
async def one(w: ShopWorker):
try:
# 单店整体超时：防止一家店拖住整轮
return w.shop_id, await asyncio.wait_for(
task_fn(w), timeout=timeout)
except asyncio.TimeoutError:
return w.shop_id, {"ok": False, "reason": "shop_timeout"}
except PermissionError as e:
return w.shop_id, {"ok": False, "reason": f"auth:{e}"}
except Exception as e:
return w.shop_id, {"ok": False, "reason": repr(e)[:160]}
results = await asyncio.gather(*(one(w) for w in workers))
ok = sum(1 for _, r in results if r.get("ok"))
return {"ok": ok, "fail": len(results) - ok, "detail": dict(results)}
```
关键就一句：`gather(..., return_exceptions=True)` 的效果——**用 `wait_for` 包住单店 + 逐个捕获**，这样一家店超时，其他十九家的结果照常返回。别用 `gather` 的默认行为，它会让第一个异常直接冒出，剩下的协程被取消，等于全挂。
额度也要按店分：
```python
# 每个店铺每天一个额度配额，用完就停，不占别人的
gate = Gate(per_key_quota={
"shop:taobao:item_list": 800,
"shop:jd:item_list": 500,
})
```
这样一家店做活动把额度打满，只是它自己停，其他店照跑。**这个设计在 618、双 11 那种日子价值极高**——那一天你一定会遇到某家店的数据量翻十倍。
## 五、数据隔离：靠 review 记住加 `where` 一定会漏
技术隔离是内部问题，数据隔离是**对客户的责任问题**。你的系统里同时有 20 家店的数据，客户 A 登录进来，绝不能看到客户 B 的任何一行。
两种做法，差别很大：
**做法一（错）**：业务代码里手写 `where shop_id = ?`。写的时候都记得，三个月后新加一个查询，忘了。这类事故在代运营行业不是罕见事件。
**做法二（对）**：在数据访问层强制注入，业务层拿不到「不带 shop 的入口」。
```python
# ops/tenant.py
class TenantRepo:
"""所有查询强制带 shop_id。业务代码只有这一个入口，
想查全库得绕过这个类——那就要 review 了，能拦住。"""
def __init__(self, db, shop_id: str):
if not shop_id:
raise ValueError("shop_id 不能为空") # 空值 = 全库查询，直接炸
self.db, self.shop_id = db, shop_id
def items(self, platform: str, limit: int = 500):
return self.db.execute(
"select item_id, payload, updated_at from items "
"where shop_id=? and platform=? order by updated_at desc limit ?",
(self.shop_id, platform, limit)).fetchall()
def orders(self, platform: str, start: int, end: int):
return self.db.execute(
"select * from orders where shop_id=? and platform=? "
"and created_at between ? and ?",
(self.shop_id, platform, start, end)).fetchall()
# 需要跨店聚合的（内部报表），单独一个类，命名上区分开
```
`shop_id` 为空就抛异常那两行是关键：**空值等于「查全库」**，历史上无数越权事故就是这么来的（参数没传、前端选错、定时任务忘了带）。让它炸，比让它查全库安全。
再配一条测试，比任何 review 都可靠：
```python
def test_no_cross_shop_leak(db):
a = TenantRepo(db, "shop_a").items("taobao")
b_ids = {r[0] for r in TenantRepo(db, "shop_b").items("taobao")}
assert not ({r[0] for r in a} & b_ids), "A 的查询结果里出现了 B 的商品"
```
这条测试跑在 CI 里，任何人写出串店查询都会被拦下。
## 六、店铺生命周期：进来和走的时候最容易乱
20 家店的代运营，每个月都有新客户进来、老客户离开。这两件事没有做成流程，就会留下垃圾和风险。
**进来（onboarding）**，25 分钟能搞定，但要有清单：
| 步骤 | 容易漏的 |
| --------- | ---------------------------------------- |
| 建店记录、绑定平台 | 一个客户在三个平台开店，要建 3 个授权，别只建一个 |
| 商家授权 | **必须商家本人扫码**，不能用员工账号代授——员工离职那天，20 家店一起失效 |
| 核对权限包 | 商品、订单、物流、售后四个包分别确认，缺哪个当场说 |
| 拉一次样本 | 抽 5 个商品 + 3 个订单，跟商家后台对一遍，确认字段和口径 |
| 配日报 | 收件人、时间、关注字段；先发一次给客户确认格式 |
| 记入到期日历 | 授权到期日 + 48 小时预警，写进系统而不是某人的日历 |
那次「员工用老板账号授权」值得单独说。代运营公司里常见做法是运营用客户的主账号扫码授权，图省事。后果是：客户改密码、老板换手机、员工离职交接不清，任何一种都会让授权失效，而且**失效的时候你不会收到通知**，只会发现数据不更新了。正确做法是让商家用自己的账号授权给你的服务商应用，然后给你开子账号。
**离开（offboarding）**，比进来更重要，因为涉及数据责任：
1. **立即 revoke**：调平台的解绑接口，别只是把状态改成 `revoked`——平台侧的授权还在，只是你不用，等于钥匙还在门上。
2. **数据保留期写进合同**：常见是 30 天，到期物理删除。别口头说「我们会删」。
3. **密钥销毁**：这家店的 token 密文连同 `key_id` 一起删。留着解不开也没意义，但审计时会问。
4. **归档报表**：客户可能要历史数据，导出成文件交给他，然后删库。
第 1 条我们踩过：客户走了三个月，我们只是标了 `revoked`，平台侧授权还在。客户的 IT 后来做安全审计，发现我们的应用还在他的授权列表里，直接投诉到了平台。**这件事的后果不是技术问题，是信任问题。**
## 七、20 家店的日报：一张表看完，异常行标记
最后是每天真正要看的东西。47 个授权，不可能一个个点进去看，所以汇总成一张表，一屏装下：
```python
# ops/dashboard.py
import time
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def daily_board(db) -> list[dict]:
"""一行 = 一个授权。异常行带 reason，前端标红。"""
rows = db.execute("""
select g.shop_id, g.platform, g.state,
(g.expire_at - strftime('%s','now')) / 86400.0 as days_left,
(select count(*) from items i
where i.shop_id=g.shop_id and i.platform=g.platform
and i.updated_at > strftime('%s','now') - 86400) as changed,
(select max(updated_at) from items i
where i.shop_id=g.shop_id and i.platform=g.platform) as last_ok
from grants g
order by g.state desc, days_left asc
""").fetchall()
out = []
for shop_id, platform, state, days_left, changed, last_ok in rows:
reason = ""
if state in ("expired", "revoked"):
reason = "授权不可用"
elif days_left is not None and days_left < 2:
reason = f"授权 {days_left:.1f} 天后到期"
elif last_ok is None or (time.time() - last_ok) > 86400 * 2:
reason = "超过 2 天没有新数据"
elif changed == 0:
reason = "昨日无变更" # 不一定是错，但值得看一眼
out.append({"shop": shop_id, "platform": platform, "state": state,
"days_left": round(days_left, 1) if days_left else None,
"changed": changed, "reason": reason})
return out
```
这张表每天 9:00 推到运营群，20 行，异常行带原因。运营的实际工作量从「逐店巡检 2 小时」变成「看 10 分钟，处理 1~3 个异常」。
要注意 `changed == 0` 这一条：**它不是错误，但它是信号**。一家正常营业的店，一天商品和订单一定有变化。连续两天为 0，八成是拉取失败但没报错（第 14 篇讲过的空值问题）。这条规则帮我们抓到过两次静默故障。
## 八、六个坑，按踩到的频率排
| 坑 | 症状 | 解法 |
| ------------------ | ------------------------- | --------------------------- |
| 员工用客户主账号授权 | 员工离职 / 客户改密码，一批店同时失效，且无通知 | 商家本人授权 + 子账号操作 |
| 凭证表主键用 `shop_id` | 一家店三个平台只有一个 token 生效 | 主键必须是 `(shop_id, platform)` |
| 只加密 access_token | 数据库泄露 = 永久授权泄露 | refresh_token 一起加密 |
| 全局令牌桶 | 一家店大促把额度吃光，其他店全延迟 | 按店铺分桶 |
| 手写 `where shop_id` | 某次查询漏了，串店 | 数据层强制注入 + CI 测试 |
| 离开时只标状态不解绑 | 客户审计发现授权还在 | 调平台解绑接口 |
## 九、这套东西的规模效应
47 个授权，每天跑一轮增量：
- 全量拉取：约 18,800 次/天（47 × 平均 400 商品）
- 改增量后：约 2,300 次/天（只有变动的商品和订单）
人工部分：巡检从 2 小时/天降到 10 分钟/天，月底对账从半天降到 1 小时。真正省下的不是时间，是**「哪家店可能出问题」这件事不再依赖某个人记得**。
---
多店铺管理这件事，技术上没有什么新鲜东西——加密、单飞、隔离、强制注入，都是老办法。它难的地方在于：**每一个细节都对应一个真实客户的损失**。串店是信任问题，断服是续费问题，授权过期是关系问题。
所以这套代码里我写得最仔细的，不是并发和性能，是那几行「失败时不清旧 token」「shop_id 为空就抛异常」「离开时必须真的解绑」。未分类
做 ERP 的看过来：三平台的商品订单接口，不用再一家家啃文档（附 Python 源码）
万邦科技Lex 发表于2026-11-03 浏览94 评论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 浏览101 评论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 浏览119 评论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 浏览99 评论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 浏览152 评论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 浏览143 评论0
这问题比“用哪个接口”更靠前一层。先给结论：**比价系统的数据源不该只选一个，而应按用途分三层**。只选一个，必然在覆盖和精度之间二选一。
## 五类数据源的真实差异