×

🌉《奇门标准接口在二手ERP中的适配:淘宝奇门协议对接闲鱼的桥接方案》(附Python源码)

万邦科技Lex 万邦科技Lex 发表于2026-09-10 11:14:03 浏览24 评论0

抢沙发发表评论

关键信息已确认,特别是奇门3.0的四套模板(勿流发货/配送可视化/自动发货/共享库存)和必须聚石塔部署的约束。现在写正文和源码,核心是"桥接器"模式——把奇门作为统一传输层,闲鱼作为业务适配层。

🌉《奇门标准接口在二手ERP中的适配:淘宝奇门协议对接闲鱼的桥接方案》(附Python源码)

结论先拍:奇门(qimen.taobao.com)是阿里系统一的商家-服务商标准接口协议,但它的标准API模板只有四套——电子面单发货(勿流)、配送可视化、自动发货、共享库存。 闲鱼的业务能力(商品发布、订单履约、退款、消息)并不在奇门标准模板里,奇门是"传输/规范层",闲鱼是"业务层"。 所以正确的桥接姿势是:奇门做统一传输标准(鉴权/签名/报文规范),自研适配层做闲鱼业务语义翻译——把闲鱼的 alibaba.idle.* API 包装成符合奇门报文规范的标准接口,让 ERP 只认奇门一套协议。 下面是完整桥接方案 + 源码实测。

一、先把概念分清楚:奇门 ≠ 闲鱼API


奇门(Qimen)
闲鱼开放平台
定位
协议/标准层(统一报文、鉴权、签名)
业务能力层(二手交易API)
提供什么
四套标准API模板 + 聚石塔部署规范
alibaba.idle.item.publish 等具体API
谁在用
ERP/WMS/物流服务商的互通标准
闲鱼商家/服务商的业务调用
部署
必须聚石塔(内部集成应用走奇门)
官方API(聚石塔强制度视标签)
关键认知:奇门是"所有阿里系商家能力的标准化出口",闲鱼是其中一个业务能力域。直接调 alibaba.idle.* 是点对点;过奇门是把闲鱼能力"标准化封装"后暴露给 ERP,好处是 ERP 只需对接奇门一套协议,未来换平台只需改适配层。

二、奇门四套标准模板(实测边界)

官方明确的标准API能力:
模板
场景
核心
电子面单发货(勿流)
仓→快递 → ERP回传物流
deliveryorder.create / consign
配送可视化
物流轨迹回传平台
deliveryorder.track
自动发货
虚拟/卡密类自动交付
item.auto.deliver
共享库存
多仓库存实时同步
inventory.report / occupy
这四套全是"履约/库存"维度,没有"商品发布/退款"——所以闲鱼的发布、退款能力必须自研扩展接口(奇门允许自定义API,但需走标准报文规范)。

三、桥接架构:三层分离

┌────────────────────────────────────────────────────┐
│  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': ['内部集成应用必须聚石塔部署(奇门规范)', ...]}

五、五个桥接铁律

  1. 奇门是协议不是能力库:它的价值是统一报文/鉴权/部署规范,不是替你实现闲鱼业务逻辑。四套标准模板只覆盖履约+库存,发布/退款必须自研扩展。

  2. method命名要分层:标准模板用官方method(deliveryorder.*/inventory.*),闲鱼自研扩展必须用命名空间隔离alibaba.idle.*),否则和官方模板冲突=路由混乱。

  3. 报文翻译是适配层核心:奇门字段(orderCode/logisticsNo/itemCode)↔ 闲鱼字段(order_id/out_sid/item_id)的转换集中在 _translate(),业务侧无感。

  4. 聚石塔是硬约束:内部集成应用走奇门必须聚石塔部署(前篇 JushitaCheck 同理),本地能跑不代表上线合规。

  5. SessionKey贯穿全链路:奇门报文里的 session 就是前篇 TokenManageraccess_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 统一通过奇门总线调度?


群贤毕至

访客