# 电商 ERP 对接三平台，90% 的问题都出在字段和超时上（附 python 源码）
> 我们做过十几个 ERP 对接项目，复盘下来有个很稳定的分布：**字段问题 62%，超时问题 28%，剩下的 10% 才是签名、限流、权限这些**。
>
> 这个分布有点反直觉——工程师通常以为对接难在协议和加密，实际难的地方在**语义**和**时间窗口**。
>
> 原因也简单：普通的数据同步，是把平台的数据搬到自己库里；ERP 对接，是让**平台的交易记录**和**企业的会计账**在同一个时间窗口内闭合。前者错了解析字段重跑就行，后者错了要调账。
>
> 这篇讲两件事：为什么 ERP 的字段问题特别多，以及 ERP 说的「超时」到底指什么。
先说清楚边界：单位和时区那类错误（分/元、秒/毫秒）第 14 篇讲过，这里不再重复。这篇讲的是**更靠上的语义层**——字段含义对不上，以及业务时限兜不住。
## 一、ERP 与普通同步的根本差别：交易视图 vs 凭证视图
平台接口返回的是**交易视图**：这笔交易发生了什么，买家付了多少钱，货发到哪。
ERP 需要的是**凭证视图**：这笔业务记哪个科目、算哪个成本中心、归哪个账期、冲销哪张单。
两种视图之间不是字段改名，是**结构变形**。下面四个变形，是 62% 字段问题的来源（按出现频次排序）：
| # | 变形 | 平台给的 | ERP 要的 | 错了的后果 |
| - | ----------- | ---------------- | ------------------------ | ---------------- |
| 1 | 金额拆分 | 一个 `payment`（实付） | 商品金额 + 运费 + 优惠 + 应收 + 佣金 | 毛利算不对，优惠成本全压在订单上 |
| 2 | SKU → 物料 | 一个销售 SKU | 多个物料 + 数量（组合装） | 库存只扣一部分，静默虚高 |
| 3 | 时间 → 账期 | 创建时间、支付时间 | 业务日期（可配日结切点） | 跨月订单归错月，GMV 差几个点 |
| 4 | 订单状态 → 单据状态 | 5 个交易状态 | 审单 / 配货 / 出库 / 开票多张单据 | 状态机错乱，已出库单被重复发货 |
### 变形一：金额是一个数，但账需要五个数
平台给 `payment`（买家实付），ERP 要做三件事：拆出商品金额与运费；把订单级优惠**分摊到行**（否则算不出单品行毛利）；算出应收（扣平台佣金后的金额）。
其中最容易出错、也最能体现专业度的是**优惠分摊**：
```python
# erp/money.py
from decimal import Decimal, ROUND_HALF_UP
CENT = Decimal("0.01")
def allocate_discount(total_discount: Decimal, lines: list[dict]) -> list[Decimal]:
"""
把订单级优惠按行金额比例分摊到行。
lines: [{"line_id": ..., "amount": Decimal}] —— amount 是分摊前的行金额
返回: 与 lines 等长的分摊额列表（正数表示优惠额）
两个必须处理的边界：
1. 尾差：整除不了时，差额补在最后一行，否则各行相加 ≠ 订单优惠，对账天天差 0.01
2. 最后一行被补成负数：极端比例下尾差可能超过该行金额，要回退给金额最大的行
"""
if total_discount == 0:
return [Decimal("0")] * len(lines)
base = sum(l["amount"] for l in lines)
if base == 0: # 全额赠品单：无金额可分摊，全部记到首行
out = [Decimal("0")] * len(lines)
out[0] = total_discount
return out
shares, allocated = [], Decimal("0")
for l in lines[:-1]:
share = (l["amount"] * total_discount / base).quantize(CENT, ROUND_HALF_UP)
shares.append(share)
allocated += share
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
tail = total_discount - allocated # 尾差兜底
last = lines[-1]["amount"]
if shares and tail > last: # 最后一行承受不住，转给金额最大行
idx = max(range(len(lines)), key=lambda i: lines[i]["amount"])
shares[idx] += tail
shares.append(Decimal("0"))
else:
shares.append(tail)
return shares
```
**那个尾差不是理论问题。** 我们遇到过一个客户，对账表每天差 0.01，财务查了两周，最后发现是三个行按比例分摊，四舍五入后多出 1 分钱没人认领。ERP 对账是**分毫必究**的，普通业务系统里可以忽略的四舍五入，在这里会每天生成一条差异记录。
还有个容易漏的：**赠品行的 `amount` 是 0，但必须保留**。很多人清洗数据时把 0 元行删了，看起来数据更干净，实际是把这个赠品的**成本丢出了订单**——月末算单均成本时偏低，越卖越亏还找不到原因。
### 变形二：一个销售 SKU 不等于一个物料
组合装是最典型的坑。平台上一个「三件套礼盒」是一个 SKU，ERP 里是三个物料各若干件：
```python
# erp/bom.py
# 平台 SKU = 销售单元；ERP 物料 = 库存单元。两者不是一一对应
BOM: dict[str, list[tuple[str, int]]] = {
"SKU-GIFT-3IN1": [("MUG-001", 2), ("COASTER-002", 4), ("GIFTBOX-003", 1)],
"SKU-CUP-PAIR": [("MUG-001", 2)],
}
def expand(sku_id: str, qty: int) -> list[tuple[str, int]]:
"""把销售数量展开成物料需求；非组合装原样返回"""
bom = BOM.get(sku_id)
if not bom:
return [(sku_id, qty)]
return [(mat, n * qty) for mat, n in bom]
```
不展开的后果是**静默的**：卖了 100 套，只扣了 100 个杯子的库存（如果 SKU 编码恰好和主物料同名），包装盒和杯垫根本没扣。系统不报错，账面看着正常，直到盘点发现包装盒少了 100 个——**而这时已经严重缺货，还在接单**。
比这更隐蔽的是反方向：**平台侧修改了组合装成分**（运营把三件套改成了两件套），你本地的 BOM 没更新。新订单按旧 BOM 扣料，多扣一个杯子，日积月累库存对不上，看谁都像偷了货。所以 BOM 要有版本和生效时间，且**跟着商品详情一起增量同步**（第 13 篇的详情同步应该带出组合装信息）。
### 变形三：ERP 要的不是「时间」，是「账期」
平台给你 `created`（下单）和 `pay_time`（支付），ERP 要的是**业务日期**——这笔钱算哪个月的。三条规矩：
1. **用支付时间，不用创建时间。** 上月 30 号下单、本月 1 号付款的订单，按创建时间归上月，上月 GMV 虚高、本月偏低，两个月的报表都要解释。
2. **日结切点可配。** 很多 ERP 的日结在凌晨 1~2 点（财务第二天早上结前一天的账），0 点到 2 点之间的订单应该算前一个业务日。
3. **时区必须显式。** 这条第 14 篇讲过，但 ERP 里更要命：财务的月报是以北京时间的自然月为准的。
```python
# erp/calendar.py
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
CN = ZoneInfo("Asia/Shanghai")
def business_date(paid_at: datetime, cutoff_hour: int = 0) -> str:
"""业务日期：按支付时间归属账期，支持自定义日结切点"""
t = paid_at.astimezone(CN) if paid_at.tzinfo else paid_at.replace(tzinfo=CN)
if cutoff_hour and t.hour < cutoff_hour:
t -= timedelta(days=1) # 凌晨单归前一业务日
return t.strftime("%Y-%m-%d")
```
### 变形四：订单状态机不能直接搬
第 3 篇讲过订单状态机怎么设计。ERP 里还要多一层：一个平台订单在 ERP 里会**生成多张单据**（销售订单 → 出库单 → 应收单 → 发票），每张单据有自己的状态。把平台状态直接映射成 ERP 主单状态，会出现「已出库但未开票」这类中间态无处安放的问题。
**规矩是：平台状态映射到「事件」，事件驱动 ERP 单据流转。** 不要映射状态，映射事件。`WAIT_SELLER_SEND_GOODS` 不是一个状态，是「可以配货了」这个事件。
## 二、超时：ERP 说的超时，八成不是接口超时
这是最容易被技术人员误解的部分。工程师听到「超时」，第一反应是调 `timeout=` 参数；而在 ERP 场景里，真正的超时发生在三层，而且**接口层是最好解决的那层**：
| 层级 | 时限量级 | 超时的后果 | 谁来兜 |
| ------- | ----------- | ------------ | ------- |
| 接口超时 | 秒 | 页面转圈、用户重试 | 工程师 |
| 审单/发货时限 | 小时~天 | 平台扣分、罚款、流量降权 | 运营 + 系统 |
| 对账窗口 | 天（T+1 / 月结） | 账不平，财务停工 | 系统 + 财务 |
### 2.1 接口层：ERP 前台是同步阻塞的，所以不能长重试
普通后台同步可以慢慢重试（第 9 篇的预算式重试），但 ERP 不一样：**用户在页面上点「同步订单」，鼠标在那儿等着**。一次重试 30 秒，用户就以为卡死了，会连点五次——于是同一批订单被并发拉五遍。
解法不是把超时调长，而是**把长事务拆成「提交 + 查结果」**：
```python
# erp/jobs.py —— 提交-轮询模式：前台 3 秒拿到 task_id，重活交后台
import asyncio
import uuid
class JobService:
def __init__(self, store, worker):
self.store = store # 持久化任务表（对应第 2 篇的 SQLite 任务表思路）
self.worker = worker
async def submit(self, kind: str, idem_key: str, payload: dict) -> str:
"""幂等键：同一批订单连点五次，只产生一个任务"""
exist = await self.store.find_by_idem(idem_key)
if exist:
return exist.task_id
task_id = uuid.uuid4().hex
await self.store.create(task_id, kind, idem_key, payload, status="pending")
self.worker.wake() # 不 await 执行，立即返回
return task_id
async def status(self, task_id: str) -> dict:
task = await self.store.get(task_id)
return {"task_id": task_id, "status": task.status,
"done": task.done, "failed": task.failed,
"errors": task.errors[:10]} # 只回传前 10 条，前端可展开
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 后台 worker：每项任务有独立的业务截止时间（DDL），到点没做完就切下一批
async def run_batch(items, handler, ddl_seconds: int = 300):
loop = asyncio.get_running_loop()
deadline = loop.time() + ddl_seconds
skipped = []
for it in items:
if loop.time() >= deadline:
skipped.append(it) # 超预算的放回队列，下一轮继续，不算失败
continue
await handler(it)
return skipped
```
三个设计点：
1. **幂等键 = 业务主键**（比如 `sync_order:2026-10-29:shop_123`）。用户连点、两个运营同时点、定时任务和手动任务撞在一起，都只产生一个任务。
2. **接口 3 秒内一定返回**。返回 `task_id` 而不是结果，前端轮询 `status`。这比把 `timeout` 调到 60 秒体验好得多，也避免了长连接占资源。
3. **后台任务有 DDL，超了就把剩下的放回队列**，不要硬跑完——因为下一次调度很快就到，硬跑完只会让两轮任务叠在一起（这就是第 10 篇说的「任务堆积」）。
### 2.2 业务层：库存回写要两段式，不能「等接口成功」
超卖的根因在 ERP 里通常不是接口慢，而是**扣库存的时机错了**。
想一想 ERP 的出库流程：审单 → 生成出库单 → 拣货 → 出库 → 回写平台库存。如果只在「出库完成」时才扣库存，那么从审单到出库这段时间，库存是**虚高**的，平台还在卖，卖出去的单进来又排队等出库——超卖就是这么堆起来的。
正确做法是**预占 + 确认**（第 3 篇讲过 `total - reserved - safety` 的账本，这里给 ERP 的落法）：
```python
# erp/stock.py —— 预占 + 确认，两段式
async def reserve(order_id: str, sku_id: str, qty: int, ttl_s: int = 3600) -> bool:
"""审单通过就预占，不等到出库。返回 False 表示可用库存不足"""
ok = await db.execute(
"UPDATE stock SET reserved = reserved + :q "
"WHERE sku_id = :s AND total - reserved - safety >= :q",
{"q": qty, "s": sku_id},
)
if ok:
await db.execute(
"INSERT INTO reservation(order_id, sku_id, qty, expires_at) "
"VALUES (:o, :s, :q, now() + :ttl)",
{"o": order_id, "s": sku_id, "q": qty, "ttl": ttl_s},
)
return bool(ok)
async def confirm(order_id: str):
"""出库成功：预占转实扣，同步回写平台"""
await db.execute(
"UPDATE stock SET total = total - r.qty, reserved = reserved - r.qty "
"FROM reservation r WHERE r.order_id = :o AND r.sku_id = stock.sku_id",
{"o": order_id},
)
await db.execute("DELETE FROM reservation WHERE order_id = :o", {"o": order_id})
await push_stock_to_platform(order_id)
async def release_expired():
"""TTL 到期的预占自动释放——取消单、审核不通过、接口超时都靠这个兜底"""
await db.execute("DELETE FROM reservation WHERE expires_at < now() RETURNING sku_id, qty")
```
关键在 `release_expired`：**所有异常路径最终都收敛成「预占超时自动释放」**。审核卡住、出库单作废、平台接口 502，不需要每种情况都写补偿逻辑，等 TTL 到就行。这是把分布式一致性问题降级成一个定时任务。
### 2.3 对账层：日切和差异表
T+1 对账是 ERP 的硬需求，坑集中在两处：
- **日切边界**：ERP 按业务日切（前面 `business_date`），平台按自然日，两边窗口不一样，边界上一定有差异。解法是**对账窗口取两边的交集并额外多取一天**，差异在窗口内消化。
- **差异要有表，不能只有日志**。差异记录必须能落到 `recon_diff` 表里（订单号、金额差、状态差、处理结果），财务才能逐条勾对。只在日志里喊一句「对账不平」，等于没做对账。
## 三、把两类问题收成一份契约表
字段和超时，看起来是两个问题，本质是一个：**你和 ERP 之间对「这笔业务的含义」和「多久必须到达」没有书面约定**。
所以我们每个 ERP 项目都会产出一份 `contract.yaml`，既是配置，也是交接文档：
```yaml
# contract.yaml —— ERP 对接契约：含义 + 单位 + 缺失处理 + 超时语义
fields:
paid_amount:
source: { taobao: payment, jd: orderPrice, alibaba: totalAmount }
semantics: 买家实付，含运费、已扣优惠
unit: CNY # jd 原始单位为分，由适配器归一（见第 14 篇）
missing: error # 缺失即报错，不允许落 0
discount:
semantics: 订单级优惠总额
allocate: by_line_amount # 必须分摊到行，规则写进契约
sku_id:
semantics: 销售单元，可能对应多个物料
expand: bom # 出库前必须展开，见 bom.py
pay_time:
semantics: 业务日期基准（不是创建时间）
business_day_cutoff: 2 # 凌晨 2 点日结
timeouts:
sync_order:
deadline_ms: 3000 # 前台同步接口上限
on_timeout: enqueue_job # 超时不重试，转异步任务
idem_key: "sync_order:{shop_id}:{date}"
push_stock:
deadline_ms: 2000
on_timeout: retry_queue # 后台任务可重试，退避由链路层负责
consistency: reserve_confirm # 两段式，不允许直写
reconcile:
window: T+1
diff_table: recon_diff
```
这张表的价值在项目交付后：**换人接手时，不用读代码猜「这个字段到底什么意思」，表里写着。** 半年后平台改字段，也是从这张表开始改，改完跑一遍启动自检确认没有漏配。
```python
# erp/contract_check.py —— 启动自检：契约里声明的，代码里必须都实现
def check(contract: dict, adapters: dict):
problems = []
for name, spec in contract["fields"].items():
for platform in spec["source"]:
if platform not in adapters:
problems.append(f"{name}: 缺 {platform} 适配器")
elif not hasattr(adapters[platform], f"map_{name}"):
problems.append(f"{name}: {platform} 未实现映射")
if spec.get("allocate") and not spec.get("unit"):
problems.append(f"{name}: 声明了分摊但没声明单位")
if problems:
raise SystemExit("契约定稿失败:\n " + "\n ".join(problems))
```
启动就炸，比上线三天后发现某个平台少了个字段好得多。
## 四、一句话总结
ERP 对接难，不难在协议。**难在平台描述的是「交易」，ERP 记的是「账」；平台保证的是「最终一致」，ERP 需要的是「账期内闭合」。**
字段是语义翻译问题，超时是时间窗口问题。把这两件事都写进契约表，90% 的问题在发生之前就消掉了。
剩下 10% 是签名、限流、授权那些，前面几篇都写过：签名和鉴权看第 14 篇，限流看第 10 篇，链路层稳定性看第 9 篇。
---
*如果不想在这一层反复踩坑，我们提供的三平台数据接口已经把金额拆分、物料展开、业务日期这三件事做成了标准字段，直接对接 ERP 少写一半适配代码。评论区或私信聊。*
