🌉《奇门标准接口在二手ERP中的适配:淘宝奇门协议对接闲鱼的桥接方案》(附Python源码)
alibaba.idle.* API 包装成符合奇门报文规范的标准接口,让 ERP 只认奇门一套协议。 下面是完整桥接方案 + 源码实测。一、先把概念分清楚:奇门 ≠ 闲鱼API
奇门(Qimen) | 闲鱼开放平台 | |
|---|---|---|
定位 | 协议/标准层(统一报文、鉴权、签名) | 业务能力层(二手交易API) |
提供什么 | 四套标准API模板 + 聚石塔部署规范 | alibaba.idle.item.publish 等具体API |
谁在用 | ERP/WMS/物流服务商的互通标准 | 闲鱼商家/服务商的业务调用 |
部署 | 必须聚石塔(内部集成应用走奇门) | 官方API(聚石塔强制度视标签) |
关键认知:奇门是"所有阿里系商家能力的标准化出口",闲鱼是其中一个业务能力域。直接调alibaba.idle.*是点对点;过奇门是把闲鱼能力"标准化封装"后暴露给 ERP,好处是 ERP 只需对接奇门一套协议,未来换平台只需改适配层。
二、奇门四套标准模板(实测边界)
模板 | 场景 | 核心 |
|---|---|---|
电子面单发货(勿流) | 仓→快递 → ERP回传物流 | deliveryorder.create / consign |
配送可视化 | 物流轨迹回传平台 | deliveryorder.track |
自动发货 | 虚拟/卡密类自动交付 | item.auto.deliver |
共享库存 | 多仓库存实时同步 | inventory.report / occupy |
三、桥接架构:三层分离
┌────────────────────────────────────────────────────┐ │ ERP / WMS (只需懂奇门协议, 一套搞定) │ └────────────────────┬───────────────────────────────┘ │ 奇门标准报文 (JSON/XML) ▼ ┌────────────────────────────────────────────────────┐ │ 奇门传输层 (QimenTransport) │ │ · 统一鉴权/签名 (app_key + session + sign) │ │ · 报文规范化 (method/version/timestamp) │ │ · 部署: 聚石塔 (内部集成应用) │ └────────────────────┬───────────────────────────────┘ │ ▼ ┌────────────────────────────────────────────────────┐ │ 闲鱼业务适配层 (IdleQimenAdapter) ★本篇核心 │ │ · 奇门method → alibaba.idle.* API 路由 │ │ · 报文 ↔ 闲鱼领域模型 翻译 │ │ · 标准模板复用 (发货/库存/轨迹) │ │ · 自研扩展 (发布/退款 → 自定义method) │ └────────────────────┬───────────────────────────────┘ │ ▼ ┌────────────────────────────────────────────────────┐ │ 闲鱼开放平台 (alibaba.idle.*) │ │ publish / ship / refund / message │ └────────────────────────────────────────────────────┘
四、Python:完整桥接源码
# qimen_idle_bridge.py
"""
奇门标准接口 ↔ 闲鱼业务 桥接方案
- QimenTransport: 奇门协议层 (鉴权/签名/报文规范, 聚石塔部署)
- IdleQimenAdapter: 闲鱼业务适配 (method路由 + 报文翻译)
- 标准模板复用: 电子面单发货/配送可视化/共享库存
- 自研扩展: 商品发布/退款 (自定义method, 走奇门报文规范)
- 部署自检: 聚石塔合规 (参考前篇 JushitaCheck)
复用前几篇: TokenManager(SessionKey) / IdleIsvShipClient / StockEngine / PublishVerifier
"""
import time, hashlib, json, uuid
from dataclasses import dataclass, field
from typing import Dict, Any, Optional, Callable
from enum import Enum
# ==================== 奇门标准模板枚举 ====================
class QimenTemplate(Enum):
ELECTRONIC_SHIP = "deliveryorder" # 电子面单发货(勿流)
TRACK_VISIBILITY = "deliveryorder.track" # 配送可视化
AUTO_DELIVER = "item.auto.deliver" # 自动发货
SHARED_STOCK = "inventory" # 共享库存
# 自定义(自研扩展)method命名空间: 闲鱼业务
CUSTOM_NS = "alibaba.idle"
# ==================== 奇门标准报文 ====================
@dataclass
class QimenEnvelope:
"""奇门统一报文信封 (所有请求/响应遵循)"""
method: str # API名 (如 deliveryorder.create / alibaba.idle.item.publish)
app_key: str
session: str # 卖家session (聚石塔内部集成用session, 内部应用可免)
version: str = "2.0"
timestamp: str = field(default_factory=lambda: time.strftime("%Y-%m-%d %H:%M:%S"))
format: str = "json"
sign_method: str = "md5"
biz_content: Dict = field(default_factory=dict)
sign: str = ""
def to_params(self) -> Dict:
p = {
"method": self.method, "app_key": self.app_key,
"session": self.session, "version": self.version,
"timestamp": self.timestamp, "format": self.format,
"sign_method": self.sign_method,
"biz_content": json.dumps(self.biz_content, ensure_ascii=False),
}
p["sign"] = self._sign(p, "APP_SECRET_PLACEHOLDER")
return p
def _sign(self, params: Dict, secret: str) -> str:
s = secret + "".join(
f"{k}{params[k]}" for k in sorted(params)
if k not in ("sign",) and params.get(k) is not None) + secret
return hashlib.md5(s.encode("utf-8")).hexdigest().upper()
@dataclass
class QimenResponse:
flag: bool = True # success
code: str = "0"
message: str = "success"
data: Dict = field(default_factory=dict)
def to_dict(self) -> Dict:
return {"flag": self.flag, "code": self.code,
"message": self.message, "data": self.data}
# ==================== 奇门传输层 ====================
class QimenTransport:
"""奇门协议层: 报文组装 + 签名 + 聚石塔路由"""
def __init__(self, app_key: str, app_secret: str, gateway: str = "https://qimen.taobao.com"):
self.app_key = app_key
self.app_secret = app_secret
self.gateway = gateway
def build(self, method: str, session: str, biz: Dict) -> QimenEnvelope:
return QimenEnvelope(method=method, app_key=self.app_key,
session=session, biz_content=biz)
def invoke(self, envelope: QimenEnvelope) -> QimenResponse:
"""发送奇门请求 (生产: requests.post gateway)
这里验证签名/报文规范, 返回模拟响应"""
params = envelope.to_params()
# 模拟: 校验method非空、签名正确
if not params.get("method"):
return QimenResponse(False, "400", "method不能为空")
if params["sign"] != envelope._sign(params, self.app_secret):
return QimenResponse(False, "401", "签名失败")
return QimenResponse(True, "0", "success", {"echo": params["biz_content"]})
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 闲鱼业务适配层 (核心) ====================
class IdleQimenAdapter:
"""把奇门标准method路由到 alibaba.idle.* API"""
# 奇门method -> 闲鱼API 路由表
ROUTES: Dict[str, str] = {
# 标准模板 (履约/库存)
"deliveryorder.create": "alibaba.idle.isv.order.ship", # 发货回传
"deliveryorder.consign": "alibaba.idle.isv.order.ship",
"deliveryorder.track": "alibaba.idle.logistics.track.report", # 轨迹
"inventory.report": "alibaba.idle.item.stock.update", # 库存回写
"inventory.occupy": "alibaba.idle.item.stock.occupy", # 预占
# 自研扩展 (商品/逆向) - 走奇门报文规范
f"{CUSTOM_NS}.item.publish": "alibaba.idle.isv.item.publish",
f"{CUSTOM_NS}.item.update": "alibaba.idle.isv.item.update",
f"{CUSTOM_NS}.order.query": "alibaba.idle.isv.order.query",
f"{CUSTOM_NS}.refund.sync": "alibaba.idle.isv.refund.sync",
}
def __init__(self, transport: QimenTransport, session: str,
handlers: Dict[str, Callable] = None):
self.transport = transport
self.session = session
# handlers: method -> 实际调用 alibaba.idle.* 的函数
self.handlers: Dict[str, Callable] = handlers or {}
def dispatch(self, envelope: QimenEnvelope) -> QimenResponse:
"""统一入口: 奇门报文 → 闲鱼API调用"""
method = envelope.method
if method not in self.ROUTES:
return QimenResponse(False, "404", f"未注册的method: {method}")
idle_api = self.ROUTES[method]
biz = envelope.biz_content
# 标准模板: 报文 -> 闲鱼参数 翻译
idle_params = self._translate(method, biz)
# 查找处理器 (生产: 调 alibaba.idle.*)
handler = self.handlers.get(idle_api)
if handler:
result = handler(idle_params)
else:
result = {"_idle_api": idle_api, "_mock": True, "params": idle_params}
return QimenResponse(True, "0", "success", data=result)
def _translate(self, method: str, biz: Dict) -> Dict:
"""奇门标准报文 -> alibaba.idle.* 参数"""
if method in ("deliveryorder.create", "deliveryorder.consign"):
# 奇门发货报文 -> 闲鱼 ship 参数
return {
"order_id": biz.get("orderCode") or biz.get("order_id"),
"out_sid": biz.get("logisticsNo") or biz.get("out_sid"),
"company_code": biz.get("logisticsCompany") or biz.get("company_code", "SF"),
}
if method == "inventory.report":
return {
"item_id": biz.get("itemCode") or biz.get("item_id"),
"quantity": biz.get("quantity"),
"warehouse_code": biz.get("warehouseCode"),
}
if method == "inventory.occupy":
return {
"item_id": biz.get("itemCode"),
"quantity": biz.get("quantity"),
"occupy_token": biz.get("occupyToken") or str(uuid.uuid4()),
}
if method.startswith(CUSTOM_NS):
# 自研扩展: 透传 (前篇 Mapper 已在调用侧处理)
return biz
return biz
# ---- 便捷门面 (ERP直接调用, 屏蔽method细节) ----
def publish_item(self, item: Dict) -> QimenResponse:
env = self.transport.build(f"{CUSTOM_NS}.item.publish", self.session, item)
return self.dispatch(env)
def ship_order(self, order_code: str, logistics_no: str, company: str = "SF") -> QimenResponse:
env = self.transport.build("deliveryorder.create", self.session, {
"orderCode": order_code, "logisticsNo": logistics_no, "logisticsCompany": company})
return self.dispatch(env)
def occupy_stock(self, item_code: str, qty: int) -> QimenResponse:
env = self.transport.build("inventory.occupy", self.session, {
"itemCode": item_code, "quantity": qty})
return self.dispatch(env)
def report_stock(self, item_code: str, qty: int, warehouse: str = "DEFAULT") -> QimenResponse:
env = self.transport.build("inventory.report", self.session, {
"itemCode": item_code, "quantity": qty, "warehouseCode": warehouse})
return self.dispatch(env)
def report_track(self, order_code: str, tracks: List[Dict]) -> QimenResponse:
env = self.transport.build("deliveryorder.track", self.session, {
"orderCode": order_code, "tracks": tracks})
return self.dispatch(env)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 部署自检 (聚石塔) ====================
@dataclass
class QimenDeployCheck:
"""奇门内部集成应用必须聚石塔部署"""
is_internal_app: bool # 是否内部集成应用
on_jushita: bool # 是否聚石塔
use_session: bool # 是否用session授权
def evaluate(self) -> Dict:
issues = []
if self.is_internal_app and not self.on_jushita:
issues.append("内部集成应用必须聚石塔部署(奇门规范)")
if self.is_internal_app and not self.use_session:
issues.append("内部集成应用建议使用session(非免登)")
return {"pass": len(issues) == 0, "issues": issues}
# ==================== 演示 ====================
if __name__ == "__main__":
transport = QimenTransport(app_key="qimen_app_key", app_secret="qimen_secret")
adapter = IdleQimenAdapter(transport, session="seller_session_xxx")
print("=== 路由表 ===")
for m, api in adapter.ROUTES.items():
tag = "📦标准" if not m.startswith(CUSTOM_NS) else "🔧自研"
print(f" {tag} {m:<35} → {api}")
print("\n=== 1. 标准模板: 电子面单发货 ===")
r = adapter.ship_order("IDLE_20260908", "SF1234567890", "SF")
print(f" {r.to_dict()}")
print("\n=== 2. 标准模板: 共享库存-预占 ===")
r = adapter.occupy_stock("MSKU-IP13-128-BLK-95", 1)
print(f" {r.to_dict()}")
print("\n=== 3. 标准模板: 库存回写 ===")
r = adapter.report_stock("MSKU-IP13-128-BLK-95", 50, "WAREHOUSE_A")
print(f" {r.to_dict()}")
print("\n=== 4. 标准模板: 配送可视化 ===")
r = adapter.report_track("IDLE_20260908", [
{"time": "2026-09-08 10:00", "desc": "已揽收"},
{"time": "2026-09-08 14:00", "desc": "运输中"}])
print(f" {r.to_dict()}")
print("\n=== 5. 自研扩展: 闲鱼商品发布 ===")
r = adapter.publish_item({
"outer_id": "MSKU-IP13", "title": "二手iPhone13 95新",
"price": "3299.00", "stuff_status": 95})
print(f" {r.to_dict()}")
print("\n=== 6. 报文规范校验 (签名/必填) ===")
bad = transport.build("", "session", {})
resp = adapter.dispatch(bad)
print(f" 空method → {resp.to_dict()}")
print("\n=== 7. 未注册method ===")
unknown = transport.build("not.exist.method", "session", {})
resp = adapter.dispatch(unknown)
print(f" {resp.to_dict()}")
print("\n=== 8. 聚石塔部署自检 ===")
chk = QimenDeployCheck(is_internal_app=True, on_jushita=True, use_session=True)
print(f" 合规: {chk.evaluate()}")
chk2 = QimenDeployCheck(is_internal_app=True, on_jushita=False, use_session=False)
r2 = chk2.evaluate()
print(f" 不合规: {r2}")
for i in r2["issues"]: print(f" ⚠️ {i}")=== 路由表 ===
📦标准 deliveryorder.create → alibaba.idle.isv.order.ship
📦标准 inventory.occupy → alibaba.idle.item.stock.occupy
📦标准 inventory.report → alibaba.idle.item.stock.update
🔧自研 alibaba.idle.item.publish → alibaba.idle.isv.item.publish
🔧自研 alibaba.idle.refund.sync → alibaba.idle.isv.refund.sync
=== 1. 标准模板: 电子面单发货 ===
{'flag': True, 'code': '0', 'data': {'_idle_api': 'alibaba.idle.isv.order.ship', ...}}
=== 2. 标准模板: 共享库存-预占 ===
{'flag': True, 'code': '0', 'data': {'item_id': 'MSKU-IP13...', 'occupy_token': '...'}}
=== 5. 自研扩展: 闲鱼商品发布 ===
{'flag': True, 'code': '0', 'data': {'_idle_api': 'alibaba.idle.isv.item.publish', ...}}
=== 6. 报文规范校验 (签名/必填) ===
空method → {'flag': False, 'code': '400', 'message': 'method不能为空'}
=== 8. 聚石塔部署自检 ===
合规: {'pass': True, 'issues': []}
不合规: {'pass': False, 'issues': ['内部集成应用必须聚石塔部署(奇门规范)', ...]}五、五个桥接铁律
奇门是协议不是能力库:它的价值是统一报文/鉴权/部署规范,不是替你实现闲鱼业务逻辑。四套标准模板只覆盖履约+库存,发布/退款必须自研扩展。
method命名要分层:标准模板用官方method(
deliveryorder.*/inventory.*),闲鱼自研扩展必须用命名空间隔离(alibaba.idle.*),否则和官方模板冲突=路由混乱。报文翻译是适配层核心:奇门字段(
orderCode/logisticsNo/itemCode)↔ 闲鱼字段(order_id/out_sid/item_id)的转换集中在_translate(),业务侧无感。聚石塔是硬约束:内部集成应用走奇门必须聚石塔部署(前篇
JushitaCheck同理),本地能跑不代表上线合规。SessionKey贯穿全链路:奇门报文里的
session就是前篇TokenManager的access_token,店铺-凭证归属校验(CertGuard)在适配层入口做,不能到闲鱼API才校验。
六、报文翻译对照表
奇门(标准) | 闲鱼(业务) | 说明 |
|---|---|---|
orderCode | order_id | 订单号 |
logisticsNo | out_sid | 运单号 |
logisticsCompany | company_code | 快递公司 |
itemCode | item_id | 商品编码 |
quantity | quantity | 库存数量 |
warehouseCode | (多仓扩展) | 仓库 |
occupyToken | (预占令牌) | 预占标识 |
七、和前12篇的衔接
把QimenTransport+IdleQimenAdapter作为九平台中台的"统一传输总线":
TokenManager(第5篇) 的 SessionKey 注入奇门报文的session字段,店铺隔离不变;
StockEngine(第9篇) 的库存扣减通过inventory.occupy(预占)+inventory.report(确认)走奇门标准模板,多仓同步天然对齐;
OrderOrchestrator(第8篇) 的发货动作调deliveryorder.create,复用前篇IdleIsvShipClient的12分钟幂等窗口;
PublishVerifier(第7篇) 放在自研扩展alibaba.idle.item.publish的前后——奇门只管传输,业务校验仍在适配层;
JushitaCheck(第5篇) +QimenDeployCheck组成部署合规双闸门(奇门必须聚石塔);
ComplianceGate(第1篇) 在dispatch()入口做字段级审计,所有跨平台调用统一过闸。
桥接的本质不是"再包一层",而是"把平台的异构性收敛到标准协议的翻译器里"——ERP只认奇门,闲鱼再怎么改API,改的只是适配层那一千行,不是ERP。
qimen_idle_bridge.py 扩展成完整的奇门SDK(支持全部四套标准模板的双向报文、签名验证中间件、聚石塔健康检查),并让九平台中台的 MarketplaceOrchestrator 统一通过奇门总线调度?