# 做 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 接入的最后一坑。
