《1688 × 闲鱼 × Mercari 的统一采购适配层：ExternalOrderNo / OfferId / ListingId 的三角映射 + 供应商备供切换状态机》（附Python源码）
一、先把这个“三角”说清楚（很多人一开始就把 ID 混了）
平台 | 商品侧 ID | 订单侧 ID | 外部单号字段 | 本质 |
|---|---|---|---|---|
1688 | offerId / item_id（商品唯一标识） | purchase_order_id | outerOrderNo / externalOrderNo（下游单号回传） | 供货侧：买别人的货 |
闲鱼 | item_id（卖家商品ID） / isvItemId（服务商商品ID） | biz_order_id | outerOrderId（下单页透传） | 销售侧：卖自己的闲置 |
Mercari | listing_id / item_id（如 m14214569688） | order_id（平台内交易单） | 通常自己系统生成 erp_order_id | 销售侧：C2C 转售 |
核心认知：OfferId = 货源坐标，ListingId = 销售渠道上的商品坐标，biz_order_id / purchase_order_id = 交易坐标，ExternalOrderNo = 你 ERP 里的“主键锚点”。统一采购适配层干的事，就是把“卖出去的一单”反向定位到“1688 上该找哪个 offer / sku / 供应商下单”。
二、统一映射模型（Unified Procurement Mapping）
# procurement_mapping.py
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class SalesChannel(str, Enum):
XIANYU = "xianyu"
MERCARI = "mercari"
class SupplyChannel(str, Enum):
ALI1688 = "1688"
@dataclass(frozen=True)
class SourceListingRef:
"""销售渠道上的商品坐标"""
channel: SalesChannel
listing_id: str # 闲鱼 item_id / Mercari mxxxx
sku_id: Optional[str] = None
@dataclass(frozen=True)
class SupplyOfferRef:
"""供货渠道上的商品坐标"""
channel: SupplyChannel = SupplyChannel.ALI1688
offer_id: str = "" # 1688 offerId
sku_id: str = "" # 1688 skuId / spec_id
supplier_id: str = ""
@dataclass
class ProcurementBinding:
"""
核心三角映射:
erp_order_id ──1──▶ SourceListingRef (在哪个平台卖了)
└──▶ SupplyOfferRef (去哪里采购)
└──▶ external_order_no (1688 分销单回传用)
"""
erp_order_id: str
source: SourceListingRef
supply: SupplyOfferRef
external_order_no: str = "" # 1688 侧 outerOrderNo
purchase_order_id: Optional[str] = None
active_supplier_rank: int = 0 # 当前用第几个备供
version: int = 1 # 映射变更版本，防并发覆盖
created_at: float = field(default_factory=lambda: __import__("time").time())映射注册表（带幂等）
class ProcurementMappingRegistry:
def __init__(self):
# erp_order_id -> ProcurementBinding
self.by_order: dict[str, ProcurementBinding] = {}
# (channel, listing_id, sku_id) -> erp_order_id
self.by_listing: dict[tuple, str] = {}
def bind(self, binding: ProcurementBinding):
# 幂等：已存在且 version 不旧于新版本则跳过
old = self.by_order.get(binding.erp_order_id)
if old and old.version >= binding.version:
return old
self.by_order[binding.erp_order_id] = binding
self.by_listing[(binding.source.channel.value,
binding.source.listing_id,
binding.source.sku_id)] = binding.erp_order_id
return binding
def resolve_by_listing(self, channel: str, listing_id: str, sku_id: str = None) -> Optional[ProcurementBinding]:
oid = self.by_listing.get((channel, listing_id, sku_id))
return self.by_order.get(oid) if oid else None
def attach_purchase_result(self, erp_order_id: str,
purchase_order_id: str, external_order_no: str):
b = self.by_order.get(erp_order_id)
if not b:
raise KeyError(f"未绑定采购关系: {erp_order_id}")
self.by_order[erp_order_id] = ProcurementBinding(
**{**b.__dict__,
"purchase_order_id": purchase_order_id,
"external_order_no": external_order_no,
"version": b.version + 1}
)三、供应商备供切换状态机（主供售罄/断货/物流失败才切）
状态定义
class SupplierState(str, Enum): IDLE = "idle" # 未使用 ACTIVE = "active" # 当前主供 PRECHECK_FAIL = "precheck_fail" # 预览/库存不行 ORDER_FAIL = "order_fail" # 下单失败 SHIP_FAILED = "ship_failed" # 发货/物流回传失败 FALLBACK = "fallback" # 已切到备供 DISABLED = "disabled" # 人工禁用
备供表
@dataclass class SupplierCandidate: supplier_id: str offer_id: str sku_id: str priority: int # 0 = 主供 state: SupplierState = SupplierState.IDLE last_error: str = "" cost: float = 0.0 freight: float = 0.0 available: bool = True
状态机
class BackupSupplierFSM:
"""
切换触发条件:
1. createOrder.preview 返回库存不足 / 报价异常
2. trade.fenxiaoOrder.create 返回 SUB_ORDER_NOT_ENOUGH
3. 物流回传静默失败（shipped 但无 tracking_no）
4. 供应商被人工禁用
"""
def __init__(self, candidates: list[SupplierCandidate]):
self.candidates = sorted(candidates, key=lambda c: c.priority)
def current(self) -> SupplierCandidate:
for c in self.candidates:
if c.state == SupplierState.ACTIVE:
return c
# 自动选第一个可用
for c in self.candidates:
if c.state in (SupplierState.IDLE,) and c.available:
c.state = SupplierState.ACTIVE
return c
raise RuntimeError("无可用供应商")
def on_precheck_fail(self, c: SupplierCandidate, reason: str):
c.state = SupplierState.PRECHECK_FAIL
c.last_error = reason
# 切下一个
self._promote_next(c)
def on_order_fail(self, c: SupplierCandidate, reason: str):
c.state = SupplierState.ORDER_FAIL
c.last_error = reason
self._promote_next(c)
def on_ship_failed(self, c: SupplierCandidate, reason: str):
# 物流失败不一定换货，但记录并告警
c.state = SupplierState.SHIP_FAILED
c.last_error = reason
def _promote_next(self, failed: SupplierCandidate):
idx = self.candidates.index(failed)
for nxt in self.candidates[idx + 1:]:
if nxt.available and nxt.state in (SupplierState.IDLE,):
nxt.state = SupplierState.ACTIVE
failed.state = SupplierState.FALLBACK
return
# 没得切了
failed.state = SupplierState.DISABLED
raise RuntimeError("主备供全部不可用，需人工介入")
def snapshot(self):
return [
{"supplier": c.supplier_id, "priority": c.priority,
"state": c.state.value, "cost": c.cost + c.freight}
for c in self.candidates
]四、统一采购编排器（闲鱼/Mercari 卖出 → 1688 采购）
# procurement_orchestrator.py
import time
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class ProcurementOrchestrator:
def __init__(self, registry, ali1688_client, qps_limiter):
self.registry = registry
self.ali = ali1688_client
self.limiter = qps_limiter
def fulfill(self, erp_order_id: str, source: SourceListingRef,
buyer_receiver: dict, channel: str) -> dict:
"""
1. 解析销售单 -> 找到 1688 供货绑定
2. 主供 preview
3. 失败切备供
4. 下单 + 回写 external_order_no
"""
# ---- 步骤1: 商品映射（闲鱼 item_id / Mercari listing_id -> 1688 offer）
binding = self.registry.resolve_by_listing(
channel, source.listing_id, source.sku_id)
if not binding:
# 没绑定过：进入选品池人工/规则绑定（不自动瞎买）
return {"ok": False, "code": "NO_SUPPLY_BINDING",
"msg": "销售单未绑定供货 offer，禁止盲采"}
fsm = binding._fsm # 实际实现里 Binding 持有一个 FSM 或在外面管
supplier = fsm.current()
# ---- 步骤2: 下单前预览（防超卖/错价）
pre = self.ali.preview(binding.erp_order_id, PurchaseReq(
offer_id=supplier.offer_id,
sku_id=supplier.sku_id,
qty=buyer_receiver.get("qty", 1),
receiver=buyer_receiver,
outer_order_no=erp_order_id,
channel=channel,
))
if not pre["ok"]:
fsm.on_precheck_fail(supplier, pre.get("code", "preview_fail"))
supplier = fsm.current()
# 再试一次主备供
pre = self.ali.preview(binding.erp_order_id, PurchaseReq(
offer_id=supplier.offer_id,
sku_id=supplier.sku_id,
qty=buyer_receiver.get("qty", 1),
receiver=buyer_receiver,
outer_order_no=erp_order_id,
channel=channel,
))
if not pre["ok"]:
return {"ok": False, "code": "ALL_SUPPLIERS_PRECHECK_FAIL",
"snapshot": fsm.snapshot()}
# ---- 步骤3: 创建 1688 分销/代发单
self.limiter.acquire("trade.fenxiaoOrder.create")
order = self.ali.create_purchase_order(binding.erp_order_id, PurchaseReq(
offer_id=supplier.offer_id,
sku_id=supplier.sku_id,
qty=buyer_receiver.get("qty", 1),
receiver=buyer_receiver,
outer_order_no=erp_order_id,
channel=channel,
))
if not order["ok"]:
fsm.on_order_fail(supplier, order.get("code"))
return {"ok": False, "code": "ORDER_FAIL",
"snapshot": fsm.snapshot()}
# ---- 步骤4: 回写三角映射
self.registry.attach_purchase_result(
erp_order_id=binding.erp_order_id,
purchase_order_id=order["purchase_order_id"],
external_order_no=erp_order_id, # 1688 outerOrderNo = 自己 ERP 单号
)
return {
"ok": True,
"erp_order_id": binding.erp_order_id,
"purchase_order_id": order["purchase_order_id"],
"supplier": supplier.supplier_id,
"external_order_no": erp_order_id,
}五、和前几篇的拼接
《1688 API 全链路》：本文把“选品 → preview → fenxiaoOrder.create → 物流重推”收进编排器
《闲鱼 order.ship》：闲鱼侧
biz_order_id发货；本文是“闲鱼卖出后去 1688 买”，方向反过来《两套接口边界》：1688 商品采集 ≠ 1688 代发下单；
offerId进“选品池”，绑定后才能进“采购池”《六大坑》：这里新增第 7 坑——“销售单和采购单的 ID 体系不一致，导致退款/物流追丢”
《中台调度》：
Ali1688Adapter只懂供货，XianyuAdapter/MercariAdapter只懂销售，ProcurementOrchestrator 是中间那根轴
六、生产落地铁律
销售单不直接调 1688：必须先有
SourceListing → SupplyOffer绑定ExternalOrderNo 用 ERP 主键：别用平台单号当外键，平台单号会变
preview 必跑：1688 库存是快照，不 preview 必超卖
备供切换只因失败：别为“便宜 5 毛”动态换供，物流/售后会炸
purchase_order_id 必须落库：后续退款/拦截/截单都靠它
Mercari 无官方代发：1688 采购后只能自己/货代发日本，别以为 Mercari 有“一键代发”
闲鱼
outerOrderId是透传字段：用来把支付单和 ERP 单对齐，不是订单号本体
七、一句话收口
多平台二手 ERP 的采购层，不是“能调 1688 接口”，而是把 ListingId（在哪卖）→ OfferId（去哪买）→ ExternalOrderNo（我系统里是谁） 三根线焊死，再在供应商挂掉时，用状态机把单子无声无息切到下一根供货线——买家无感，财务能对，售后能追。
