🧩《一套ERP管5个国内二手平台：统一适配层设计与各平台接口差异对照表》（附Python源码）

一、统一适配层的三层架构

┌─────────────────────────────────────────────────────────┐
│  业务层 (一套代码, 不感知平台)                           │
│  create_product() / sync_stock() / query_order()         │
└──────────────────────┬──────────────────────────────────┘
                       │  统一领域模型 Product/Order/Stock
                       ▼
┌─────────────────────────────────────────────────────────┐
│  适配层核心                                             │
│  ┌────────────┐  ┌──────────────┐  ┌────────────────┐ │
│  │ Protocol   │  │ Registry     │  │ Translator     │ │
│  │ (端口定义) │  │ (动态分发)   │  │ (双向翻译)     │ │
│  └────────────┘  └──────────────┘  └────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
                       │
        ┌──────┬──────┼──────┬──────┬──────┐
        ▼      ▼      ▼      ▼      ▼      ▼
     idle   douyin  jd_pop  pdd   zhuanzhuan (Adapter)

1. Protocol：用 typing.Protocol 定义统一端口（publish_product / sync_stock / query_order），只定义契约不强制继承

2. Registry：dict[str, Adapter] + register() / get()，运行时动态注册，支持同一平台多版本共存

3. Translator：每个 Adapter 内聚"平台→领域"和"领域→平台"双向翻译，业务层永远只看见领域模型

二、五平台接口差异对照表（真实坑点）

这张表就是适配层存在的全部理由——任何一个"统一字段"假设都会在第3~4列崩塌（成色在闲鱼是主字段、在京东/拼多多却是类目属性值）。统一领域模型 + 各平台独立翻译是唯一能保持业务层干净的做法。

三、Python：Protocol + Registry + 统一调用（完整源码）

# unified_adapter_layer.py
"""
一套ERP管5个国内二手平台: 统一适配层
- MarketplaceAdapter Protocol (统一端口, 不强制继承)
- AdapterRegistry (运行时动态注册/分发, 支持多版本)
- 统一领域模型: Product / Order / Stock (平台无关)
- 5个平台Adapter: 各做"平台报文 <-> 领域模型"双向翻译
- 业务层: 一套代码统一调用
复用前几篇: TokenManager / StockEngine / ConditionIntMapper(各平台成色翻译)
"""
from typing import Protocol, Dict, List, Optional, Any, runtime_checkable
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
from enum import Enum
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 统一领域模型 (平台无关) ====================
@dataclass
class Money:
    amount: float = 0.0
    currency: str = "CNY"

@dataclass
class Product:
    master_sku: str
    platform: str = ""
    title: str = ""
    price: Money = field(default_factory=Money)
    condition_int: int = 0         # 统一成色 0~100 (归一化)
    condition_text: str = ""       # 原始成色文本(保留, 回写用)
    category_id: str = ""
    stock: int = 0
    outer_id: str = ""
    attributes: Dict[str, str] = field(default_factory=dict)
    raw: Dict = field(default_factory=dict)   # 平台原始报文(按需)

@dataclass
class Order:
    order_id: str
    platform: str = ""
    status: str = ""
    total: Money = field(default_factory=Money)
    items: List[Dict] = field(default_factory=list)
    is_second_hand: bool = False
    raw: Dict = field(default_factory=dict)

@dataclass
class StockChange:
    master_sku: str
    platform: str
    qty: int = 0
    warehouse: str = "DEFAULT"
    occupy_token: str = ""
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 统一端口 (Protocol) ====================
@runtime_checkable
class MarketplaceAdapter(Protocol):
    """所有平台Adapter必须实现的统一端口"""

    @property
    def platform(self) -> str: ...

    def publish_product(self, product: Product) -> Dict: ...
    def sync_stock(self, change: StockChange) -> Dict: ...
    def query_order(self, order_id: str) -> Optional[Order]: ...
    def list_orders(self, since=None, **kwargs) -> List[Order]: ...

# ==================== 适配层异常 ====================
class AdapterError(Exception): pass
class CapabilityNotSupported(AdapterError):
    """平台不具备该能力(如转转只读)"""
    def __init__(self, platform: str, feature: str):
        super().__init__(f"{platform} 不支持 {feature} (能力不对等)")
        self.platform = platform
        self.feature = feature

# ==================== 注册中心 ====================
class AdapterRegistry:
    """运行时动态注册/分发, 支持同一平台多版本"""

    def __init__(self):
        self._adapters: Dict[str, MarketplaceAdapter] = {}
        self._meta: Dict[str, Dict] = {}

    def register(self, adapter: MarketplaceAdapter, version: str = "default",
                 capabilities: Optional[List[str]] = None) -> "AdapterRegistry":
        key = f"{adapter.platform}:{version}"
        self._adapters[key] = adapter
        self._meta[key] = {
            "platform": adapter.platform,
            "version": version,
            "capabilities": capabilities or ["publish", "stock", "order"],
        }
        return self   # 链式

    def get(self, platform: str, version: str = "default") -> MarketplaceAdapter:
        key = f"{platform}:{version}"
        if key not in self._adapters:
            available = [k for k in self._adapters if k.startswith(platform + ":")]
            raise AdapterError(
                f"未注册Adapter: {key}. 该平台可用版本: {available or '无'}")
        return self._adapters[key]

    def supports(self, platform: str, feature: str) -> bool:
        """探测平台是否具备某能力"""
        for key, meta in self._meta.items():
            if key.startswith(platform + ":"):
                if feature in meta.get("capabilities", []):
                    return True
        return False

    def dispatch(self, platform: str, method: str, *args, **kwargs) -> Any:
        """统一分发: 业务层无需get()再调用"""
        adapter = self.get(platform)
        if not hasattr(adapter, method):
            raise CapabilityNotSupported(adapter.platform, method)
        return getattr(adapter, method)(*args, **kwargs)

    def all_platforms(self) -> List[str]:
        return sorted({meta["platform"] for meta in self._meta.values()})

    def capabilities_of(self, platform: str) -> Dict[str, List[str]]:
        return {key: meta["capabilities"]
                for key, meta in self._meta.items()
                if meta["platform"] == platform}

# ==================== 各平台Adapter (翻译层) ====================

# ---- 闲鱼 (TOP体系: stuff_status int) ----
class IdleAdapter:
    def __init__(self, token_manager=None):
        self._tokens = token_manager

    @property
    def platform(self) -> str: return "idle"

    def publish_product(self, product: Product) -> Dict:
        # 领域 -> 闲鱼: condition_int 直接用 (stuff_status)
        params = {
            "method": "alibaba.idle.isv.item.publish",
            "title": product.title,
            "reserve_price": f"{product.price.amount:.2f}",
            "stuff_status": product.condition_int,   # ★ int
            "outer_id": product.outer_id,
        }
        return self._invoke(params)

    def sync_stock(self, change: StockChange) -> Dict:
        return self._invoke({"method": "alibaba.idle.item.stock.update",
                             "item_id": change.master_sku, "quantity": change.qty})

    def query_order(self, order_id: str) -> Optional[Order]:
        raw = self._invoke({"method": "alibaba.idle.isv.order.query", "order_id": order_id})
        return self._to_order(raw)

    def list_orders(self, since=None, **kwargs) -> List[Order]:
        raw_list = self._invoke({"method": "alibaba.idle.isv.order.list"}) or []
        return [self._to_order(r) for r in raw_list]

    def _to_order(self, raw: Dict) -> Optional[Order]:
        if not raw: return None
        return Order(order_id=str(raw.get("order_id", "")), platform="idle",
                     status=raw.get("status", ""), raw=raw)

    def _invoke(self, params: Dict) -> Any:
        # 生产: 带SessionKey调用TOP; 这里返回模拟
        return {"_mock": True, "platform": "idle", **params}

# ---- 抖音 (枚举字符串成色) ----
class DouyinAdapter:
    @property
    def platform(self) -> str: return "douyin"

    _CONDITION_MAP = {"99新": "99新", "95新": "95新", "9新": "9新", "8新": "8新",
                     99: "99新", 95: "95新", 9: "9新", 8: "8新"}

    def publish_product(self, product: Product) -> Dict:
        cond = self._CONDITION_MAP.get(product.condition_int,
                                       self._CONDITION_MAP.get(product.condition_text, "9新"))
        params = {
            "method": "product/addV2",
            "name": product.title,
            "price": product.price.amount,
            "quality": cond,   # ★ 枚举字符串
            "category_leaf_id": product.category_id,
        }
        return self._invoke(params)

    def sync_stock(self, change: StockChange) -> Dict:
        return self._invoke({"method": "sku/syncStock",
                             "sku_id": change.master_sku, "stock_num": change.qty})

    def query_order(self, order_id: str) -> Optional[Order]:
        raw = self._invoke({"method": "order/orderDetail", "order_id": order_id})
        return self._to_order(raw)

    def list_orders(self, since=None, **kwargs) -> List[Order]:
        raw_list = self._invoke({"method": "order/list"}) or []
        return [self._to_order(r) for r in raw_list]

    def _to_order(self, raw: Dict) -> Optional[Order]:
        if not raw: return None
        return Order(order_id=str(raw.get("order_id", "")), platform="douyin",
                     status=raw.get("status", ""), raw=raw)

    def _invoke(self, params: Dict) -> Any:
        return {"_mock": True, "platform": "douyin", **params}

# ---- 京东POP (成色=类目属性 attrId×valueId) ----
class JdPopAdapter:
    @property
    def platform(self) -> str: return "jd_pop"

    # category_id -> (成色attr_id, {成色: value_id})
    _CATE_COND = {
        "15119": (90001, {"99新": 99001, "95新": 95001, "9新": 90001, "8新": 80001}),
    }

    def publish_product(self, product: Product) -> Dict:
        attrs = self._resolve_condition(product)
        params = {
            "method": "jingdong.ware.write.add",
            "ware": {"title": product.title, "categoryId": product.category_id,
                     "outerId": product.outer_id, "props": attrs},
            "sku": {"jdPrice": product.price.amount, "stockNum": product.stock},
        }
        return self._invoke(params)

    def _resolve_condition(self, product: Product) -> List[Dict]:
        cfg = self._CATE_COND.get(product.category_id, (90001, {"95新": 95001}))
        attr_id, values = cfg
        vid = values.get(product.condition_text) or values.get("95新")
        return [{"attrId": attr_id, "attrValues": str(vid)}]

    def sync_stock(self, change: StockChange) -> Dict:
        return self._invoke({"method": "jingdong.ware.stock.sku.set",
                             "sku_id": change.master_sku, "stock_num": change.qty})

    def query_order(self, order_id: str) -> Optional[Order]:
        raw = self._invoke({"method": "jingdong.pop.order.get", "order_id": order_id})
        return self._to_order(raw)

    def list_orders(self, since=None, **kwargs) -> List[Order]:
        raw_list = self._invoke({"method": "jingdong.pop.order.searchList"}) or []
        return [self._to_order(r) for r in raw_list]

    def _to_order(self, raw: Dict) -> Optional[Order]:
        if not raw: return None
        return Order(order_id=str(raw.get("order_id", "")), platform="jd_pop",
                     status=raw.get("status", ""), raw=raw)

    def _invoke(self, params: Dict) -> Any:
        return {"_mock": True, "platform": "jd_pop", **params}

# ---- 拼多多 (standard_id 属性) ----
class PddAdapter:
    @property
    def platform(self) -> str: return "pdd"

    def publish_product(self, product: Product) -> Dict:
        sid = self._resolve_standard(product)
        params = {
            "method": "pdd.goods.add",
            "cat_id": product.category_id,
            "goods_name": product.title,
            "goods_properties": [{"template_id": 1, "value": str(sid)}],
        }
        return self._invoke(params)

    def _resolve_standard(self, product: Product) -> int:
        # 1001全新, 1002/1003/1004非全新
        return 1001 if product.condition_int >= 100 else {99: 1002, 95: 1002, 9: 1003, 8: 1004}.get(
            product.condition_int, 1002)

    def sync_stock(self, change: StockChange) -> Dict:
        return self._invoke({"method": "pdd.goods.sku.stock.update",
                             "sku_id": change.master_sku, "stock": change.qty})

    def query_order(self, order_id: str) -> Optional[Order]:
        raw = self._invoke({"method": "pdd.order.info.get", "order_sn": order_id})
        return self._to_order(raw)

    def list_orders(self, since=None, **kwargs) -> List[Order]:
        raw_list = self._invoke({"method": "pdd.order.list.get"}) or []
        return [self._to_order(r) for r in raw_list]

    def _to_order(self, raw: Dict) -> Optional[Order]:
        if not raw: return None
        return Order(order_id=str(raw.get("order_sn", "")), platform="pdd",
                     status=raw.get("status", ""), raw=raw)

    def _invoke(self, params: Dict) -> Any:
        return {"_mock": True, "platform": "pdd", **params}

# ---- 转转 (MCP只读, 写能力显式抛异常) ----
class ZhuanzhuanAdapter:
    @property
    def platform(self) -> str: return "zhuanzhuan"

    def publish_product(self, product: Product) -> Dict:
        # ★ 转转MCP只读, 写能力明确不支持
        raise CapabilityNotSupported("zhuanzhuan", "publish_product")

    def sync_stock(self, change: StockChange) -> Dict:
        raise CapabilityNotSupported("zhuanzhuan", "sync_stock")

    def query_order(self, order_id: str) -> Optional[Order]:
        raise CapabilityNotSupported("zhuanzhuan", "query_order")

    def list_orders(self, since=None, **kwargs) -> List[Order]:
        # 转转订单走App私有, 明确降级
        raise CapabilityNotSupported("zhuanzhuan", "list_orders")

    # ---- 转转独有: 只读能力 (估值/行情) ----
    def estimate_recycle(self, description: str) -> Dict:
        return {"platform": "zhuanzhuan", "tool": "recycle_valuation",
                "input": description, "estimated_price": 1800}

    def market_price(self, model: str) -> Dict:
        return {"platform": "zhuanzhuan", "tool": "market_price", "model": model}

# ==================== 业务层: 一套代码统一调用 ====================
class MarketplaceService:
    """业务层: 完全不感知具体平台, 只依赖 Protocol"""

    def __init__(self, registry: AdapterRegistry):
        self.registry = registry

    def publish(self, platform: str, product: Product) -> Dict:
        adapter = self.registry.get(platform)
        return adapter.publish_product(product)

    def sync_stock(self, platform: str, change: StockChange) -> Dict:
        adapter = self.registry.get(platform)
        return adapter.sync_stock(change)

    def get_order(self, platform: str, order_id: str) -> Optional[Order]:
        adapter = self.registry.get(platform)
        return adapter.query_order(order_id)

    def publish_all(self, product: Product, platforms: List[str]) -> Dict[str, Dict]:
        """一键发布到多个平台 (批量)"""
        results = {}
        for pf in platforms:
            try:
                results[pf] = self.publish(pf, product)
            except CapabilityNotSupported as e:
                results[pf] = {"error": str(e)}
        return results

# ==================== 演示 ====================
if __name__ == "__main__":
    # ---- 注册 (支持链式 + 多版本) ----
    registry = AdapterRegistry()
    registry.register(IdleAdapter(), "default", ["publish", "stock", "order"]) \
            .register(DouyinAdapter(), "default", ["publish", "stock", "order"]) \
            .register(JdPopAdapter(), "default", ["publish", "stock", "order"]) \
            .register(PddAdapter(), "default", ["publish", "stock", "order"]) \
            .register(ZhuanzhuanAdapter(), "default",
                      ["estimate_recycle", "market_price"])  # 只读能力

    # 同一平台多版本共存示例
    class IdleAdapterV2(IdleAdapter):
        @property
        def platform(self) -> str: return "idle"
    registry.register(IdleAdapterV2(), "v2", ["publish", "stock", "order"])

    service = MarketplaceService(registry)

    print("=== 注册中心 ===")
    print(f"  平台: {registry.all_platforms()}")
    for pf in registry.all_platforms():
        print(f"    {pf}: {registry.capabilities_of(pf)}")

    print("\n=== 统一领域模型 ===")
    product = Product(
        master_sku="MSKU-IP13-128-BLK-95", title="二手iPhone13 128G 黑色 95新",
        price=Money(3299.0), condition_int=95, condition_text="95新",
        category_id="15119", stock=10, outer_id="MSKU-IP13-128-BLK-95")

    print(f"  condition_int={product.condition_int} (统一, 各平台自己翻译)")

    print("\n=== 业务层: 一套代码发5个平台 ===")
    results = service.publish_all(product, ["idle", "douyin", "jd_pop", "pdd", "zhuanzhuan"])
    for pf, r in results.items():
        print(f"  [{pf}] {r}")

    print("\n=== 关键: 成色在各平台的字段差异 (翻译实证) ===")
    print("  idle   → stuff_status =", product.condition_int, "(int)")
    print("  douyin → quality     =", DouyinAdapter().publish_product(product)["quality"], "(枚举)")
    jd_props = JdPopAdapter().publish_product(product)["ware"]["props"]
    print(f"  jd_pop → props(attrId×valueId) = {jd_props}")
    pdd_props = PddAdapter().publish_product(product)["goods_properties"]
    print(f"  pdd    → standard_id = {pdd_props}")

    print("\n=== 转转: 写操作显式降级 (能力不对等) ===")
    try:
        service.publish("zhuanzhuan", product)
    except CapabilityNotSupported as e:
        print(f"  ⚠️ 捕获: {e}")
    # 但转转独有只读能力可用
    zz = registry.get("zhuanzhuan")
    print(f"  ✅ 估值: {zz.estimate_recycle('iPhone13 128G')}")
    print(f"  ✅ 行情: {zz.market_price('iPhone 13')}")

    print("\n=== 动态分发 (dispatch 无需先get) ===")
    stock_change = StockChange(master_sku="MSKU-IP13", platform="idle", qty=5)
    r = registry.dispatch("idle", "sync_stock", stock_change)
    print(f"  idle sync_stock: {r}")

    print("\n=== 查询订单 (统一返回 Order) ===")
    for pf in ["idle", "douyin", "jd_pop", "pdd"]:
        order = service.get_order(pf, "ORDER_001")
        print(f"  [{pf}] order_id={order.order_id if order else None} platform={pf}")

    print("\n=== 版本切换演示 ===")
    print(f"  idle:default = {type(registry.get('idle', 'default')).__name__}")
    print(f"  idle:v2      = {type(registry.get('idle', 'v2')).__name__}")

    print("\n=== Protocol运行时校验 ===")
    print(f"  IdleAdapter符合Protocol: {isinstance(IdleAdapter(), MarketplaceAdapter)}")
    print(f"  ZhuanzhuanAdapter符合: {isinstance(ZhuanzhuanAdapter(), MarketplaceAdapter)}")

=== 注册中心 ===
  平台: ['douyin', 'idle', 'jd_pop', 'pdd', 'zhuanzhuan']
    idle: ['idle:default', 'idle:v2']
    zhuanzhuan: ['zhuanzhuan:default']

=== 业务层: 一套代码发5个平台 ===
  [idle]   {'_mock': True, 'platform': 'idle', 'method': 'alibaba.idle.isv.item.publish', 'stuff_status': 95}
  [douyin] {'_mock': True, 'platform': 'douyin', 'method': 'product/addV2', 'quality': '95新'}
  [jd_pop] {'_mock': True, 'platform': 'jd_pop', ..., 'props': [{'attrId': 90001, 'attrValues': '95001'}]}
  [pdd]    {'_mock': True, 'platform': 'pdd', 'goods_properties': [{'template_id': 1, 'value': '1002'}]}
  [zhuanzhuan] {'error': 'zhuanzhuan 不支持 publish_product (能力不对等)'}

=== 关键: 成色在各平台的字段差异 ===
  idle   → stuff_status = 95 (int)
  douyin → quality     = 95新 (枚举)
  jd_pop → props(attrId×valueId) = [{'attrId': 90001, 'attrValues': '95001'}]
  pdd    → standard_id = [{'template_id': 1, 'value': '1002'}]

=== 转转: 写操作显式降级 ===
  ⚠️ 捕获: zhuanzhuan 不支持 publish_product (能力不对等)
  ✅ 估值: {'platform': 'zhuanzhuan', 'tool': 'recycle_valuation', ...}

四、六个工程设计铁律

1. Protocol 优于 ABC：typing.Protocol 是结构化类型（鸭子类型），Adapter 不必强制继承基类，新增平台零侵入，运行时 isinstance 仍可校验——比 ABC 更贴合"逐步接入5个平台"的现实。

2. Registry 必须支持多版本：platform:version 复合键让 idle:v1 / idle:v2 共存，灰度切流、A/B测试、老店铺兼容都在注册层解决，业务代码不变。

3. Translator 内聚在 Adapter：成色、库存、订单的"平台↔领域"翻译全部封装在对应 Adapter 内，领域模型保持纯净——这是避免"加平台=改全局"的关键。

4. 能力不对等必须显式建模：转转的 CapabilityNotSupported 异常让业务层知道什么时候该降级（前篇 CapabilityProbe 思路），静默失败才是真坑。

5. 统一模型 ≠ 字段拉平：领域模型只保留5平台的交集语义（master_sku/title/price/condition/stock），平台独有字段（转转估值、京东质检）走 raw 或专属方法，不污染统一接口。

6. 业务层只依赖 Protocol：MarketplaceService 看不到任何具体 Adapter 类名，新增第6个平台只需新增一个文件 + 一行 register，这是分层架构的可量化收益。

五、和前13篇的衔接

把 unified_adapter_layer.py 作为整个"国内二手电商ERP"系列的总装配层：

• Product/Order/StockChange 领域模型 是各篇统一实体的收敛点——前篇的 SecondHandItem（京东）、Product（转转）、StockChange 全部对齐到这三个类；

• 成色翻译集中在各 Adapter：ConditionIntMapper（抖音/转转）、JdConditionResolver（京东）、StandardIdResolver（拼多多）的逻辑原样迁入，业务层只传 condition_int；

• CapabilityNotSupported 承接前篇 CapabilityProbe 的探测结果——转转只读、小红书轻量接入（第17篇评分：转转1.23/小红书3.05）在注册时直接声明 capabilities，运行时 supports() 可查；

• Registry 是奇门桥接（第18篇）的天然底座：QimenTransport 可作为"虚拟平台"注册为 qimen:default，ERP 调 registry.get("qimen").publish_product() 即通过奇门总线调度，统一适配层 + 统一传输层合一；

• 限流/合规/库存引擎挂装饰器：在 Registry 的 dispatch() 上加 RateLimiter（第2篇）、ComplianceGate（第1篇）、StockEngine（第9篇）中间件，业务调用自动过闸；

• 新增平台 = 一个文件：第16篇的小红书图片合规、第12篇的抖音直播库存，都可封装成独立 Adapter 即插即用。
  统一适配层的价值不是"少写代码"，而是"把5套平台的异构性封进5个翻译器，让业务层永远只说一种语言"——这正是从'能对接'到'能规模化'的分水岭。

unified_adapter_layer.py