一套接口打三平台，省下 80% 的对接工时（附 Python 源码）

80% 不是省在第一次对接，是省在之后每一次改动。这笔账要算清楚，否则你会以为统一接口层是过度设计。

一、三套代码的日常

先看一个很常见的项目结构：

project/
├── taobao/
│   ├── client.py       # 签名 + 请求
│   ├── product.py      # 商品：字段映射、SKU 处理
│   ├── order.py        # 订单：分页、状态转换
│   └── logistics.py
├── alibaba/
│   ├── client.py       # 又是一份签名
│   ├── product.py      # 又是一份字段映射
│   ├── order.py
│   └── logistics.py
├── jd/
│   ├── client.py
│   ├── product.py
│   ├── order.py
│   └── logistics.py
└── service/
    ├── price_sync.py    # ← 业务逻辑，里面全是 if platform == "taobao"
    ├── stock_sync.py    # ← 又一次 if platform == ...
    └── order_sync.py    # ← 第三次

问题不在文件多，在于每个业务需求都要改三处。

产品说了句「库存要精确到 SKU」，你得改三个 stock 模块。发现淘宝有个坑要修，另外两个平台得确认一遍是不是也有同样的问题。加个新字段，三个平台各写一遍映射。

代码量翻三倍，bug 面积也翻三倍——因为三份实现里总有一份写得跟另外两份不一样。

二、先算一笔诚实的账

「省 80% 工时」这种话，如果只算首次对接，是站不住的。算细一点：

首次对接

首次对接只省 28%。因为适配器还是得一个一个写，平台差异不会因为抽象就消失。

后续每一次改动

真正拉开差距的是这里。假设一个持续一年的项目：

一年下来：三套实现约 43 天，统一层约 16 天。

维护期省 63%。 加上首次的 28%，以及「bug 只修一处就不会漏」带来的隐性收益，综合到 80% 是合理的。

关键结论：统一接口层的价值随时间增长，项目越久省得越多。 如果只做一个短期的数据导出脚本，确实没必要。但如果是长期跑的业务系统，这个投入一年内就回本。

三、架构：门面 + 适配器 + 能力声明

       业务代码（只写一次）
              ↓
       CommerceAPI（门面）
       ├─ 统一重试 / 缓存 / 限流
       ├─ 能力检查（不支持就明确报错）
       └─ 结果归一化
              ↓
    ┌─────────┼─────────┐
 TaobaoAdapter  AlibabaAdapter  JdAdapter
    ↓              ↓              ↓
  各自签名、字段、分页、错误码

三个角色的职责必须分清：

• 门面：业务逻辑唯一入口，负责跨平台共性（重试、缓存、能力检查）
• 适配器：只做「平台方言 → 统一语言」的翻译，不含业务判断
• 能力声明：显式声明每个平台支持什么，不支持的操作直接抛错

第三条是最容易被忽略、也最容易埋雷的。下面单独讲。

四、代码

4.1 统一模型和能力声明

# core/models.py
from dataclasses import dataclass, field
from decimal import Decimal


@dataclass
class Product:
    platform: str
    item_id: str
    title: str
    price: Decimal
    stock: int | None = None          # None 表示该平台不提供
    images: list[str] = field(default_factory=list)
    skus: list["Sku"] = field(default_factory=list)


@dataclass
class Sku:
    sku_id: str
    props: dict[str, str]
    price: Decimal
    stock: int | None = None


@dataclass
class Order:
    platform: str
    order_id: str
    status: str                       # 统一状态，不是平台原始值
    pay_amount: Decimal
    created_at: str
    modified_at: str
    buyer: str = ""


@dataclass
class Capability:
    """
    每个平台支持什么，显式声明。
    这是统一接口里最关键的一层——三平台能力根本不对等，
    不声明清楚，业务代码就会在某个平台上拿到空数据还以为正常。
    """
    sku_level_stock: bool = False     # 库存是否精确到 SKU
    tiered_price: bool = False        # 是否有阶梯价（1688 特有）
    order_increment: bool = False     # 订单是否支持按修改时间增量拉
    logistics_subscribe: bool = False # 物流是否支持订阅推送
    batch_product_query: bool = False # 商品是否支持批量查询

为什么不用 hasattr 或者 try/except 判断？因为能力应该是显式契约，不是运行时探针。业务代码在启动时就能知道「京东这个接口不支持 SKU 级库存」，从而决定降级策略，而不是等跑到一半发现返回全是 None。

4.2 适配器基类

# core/adapter.py
from abc import ABC, abstractmethod
from decimal import Decimal

from .models import Product, Order, Capability


class NotSupported(Exception):
    """平台明确不支持该操作。不要用假数据糊弄调用方。"""

    def __init__(self, platform: str, operation: str):
        super().__init__(f"[{platform}] 不支持操作：{operation}")
        self.platform = platform
        self.operation = operation


class BaseAdapter(ABC):
    platform: str = ""
    capability: Capability = Capability()

    # --- 子类必须实现 ---

    @abstractmethod
    async def get_product(self, item_id: str) -> Product:
        ...

    @abstractmethod
    async def list_orders(self, start: str, end: str, page: int = 1) -> list[Order]:
        ...

    # --- 子类按能力选择性实现 ---

    async def get_stock(self, item_id: str, sku_id: str | None = None) -> int:
        if not self.capability.sku_level_stock and sku_id:
            raise NotSupported(self.platform, "按 SKU 查询库存")
        return await self._get_stock_impl(item_id, sku_id)

    async def _get_stock_impl(self, item_id: str, sku_id: str | None) -> int:
        raise NotSupported(self.platform, "库存查询")

    # --- 状态归一：每个平台状态码不同，统一成五个 ---

    STATUS_MAP: dict[str, str] = {
   }

    def normalize_status(self, raw: str) -> str:
        return self.STATUS_MAP.get(str(raw), "UNKNOWN")

统一状态只定五个：PENDING（待付款）、PAID（待发货）、SHIPPED（已发货）、DONE（已完成）、CLOSED（已关闭）。多了没用，少了不够用。

4.3 三个适配器（只展示关键差异）

# adapters/taobao.py
from decimal import Decimal

from core.adapter import BaseAdapter
from core.models import Product, Order, Capability, Sku


class TaobaoAdapter(BaseAdapter):
    platform = "taobao"
    capability = Capability(
        sku_level_stock=True,
        order_increment=True,
        logistics_subscribe=True,
        batch_product_query=False,
    )
    STATUS_MAP = {
   
        "WAIT_BUYER_PAY": "PENDING",
        "WAIT_SELLER_SEND_GOODS": "PAID",
        "WAIT_BUYER_CONFIRM_GOODS": "SHIPPED",
        "TRADE_FINISHED": "DONE",
        "TRADE_CLOSED": "CLOSED",
    }

    def __init__(self, client):
        self.c = client

    async def get_product(self, item_id: str) -> Product:
        d = await self.c.call("taobao.item.seller.get", {
   
            "fields": "num_iid,title,price,num,quantity,pic_url,sku",
            "num_iid": item_id,
        })
        it = d["item_get_response"]["item"]

        skus = []
        for s in (it.get("skus") or {
   }).get("sku") or []:
            skus.append(Sku(
                sku_id=str(s["sku_id"]),
                props=_parse_props(s.get("properties_name", "")),
                price=Decimal(str(s.get("price", "0"))),
                stock=int(s.get("quantity", 0)),
            ))

        return Product(
            platform=self.platform,
            item_id=str(it["num_iid"]),
            title=it.get("title", ""),
            price=Decimal(str(it.get("price", "0"))),
            stock=int(it.get("quantity", 0)),
            images=[it["pic_url"]] if it.get("pic_url") else [],
            skus=skus,
        )

    async def list_orders(self, start, end, page=1) -> list[Order]:
        d = await self.c.call("taobao.trades.sold.get", {
   
            "fields": "tid,status,payment,created,modified,buyer_nick",
            "start_modified": start, "end_modified": end,
            "page_no": page, "page_size": 100,
        })
        rows = ((d.get("trades_sold_get_response") or {
   })
                .get("trades") or {
   }).get("trade") or []
        return [Order(
            platform=self.platform,
            order_id=str(t["tid"]),
            status=self.normalize_status(t.get("status", "")),
            pay_amount=Decimal(str(t.get("payment", "0"))),
            created_at=t.get("created", ""),
            modified_at=t.get("modified", ""),
            buyer=t.get("buyer_nick", ""),
        ) for t in rows]


def _parse_props(s: str) -> dict[str, str]:
    out = {
   }
    for seg in filter(None, s.split(";")):
        p = seg.split(":")
        if len(p) >= 4:
            out[p[2]] = p[3]
    return out

# adapters/alibaba.py
from decimal import Decimal

from core.adapter import BaseAdapter
from core.models import Product, Order, Capability
from core.adapter import NotSupported


class AlibabaAdapter(BaseAdapter):
    platform = "alibaba"
    capability = Capability(
        sku_level_stock=False,       # 1688 按起订量阶梯，没有 SKU 概念
        tiered_price=True,
        order_increment=True,
        logistics_subscribe=False,   # 需要单独申请
    )
    STATUS_MAP = {
   
        "waitbuyerpay": "PENDING",
        "waitsellersend": "PAID",
        "waitbuyerreceive": "SHIPPED",
        "success": "DONE",
        "cancel": "CLOSED",
    }

    def __init__(self, client):
        self.c = client

    async def get_product(self, item_id: str) -> Product:
        d = await self.c.call("com.alibaba.product.get", {
   
            "productID": item_id, "webSite": "1688",
        })
        r = d.get("result") or {
   }
        ranges = (r.get("saleInfo") or {
   }).get("priceRanges") or []
        price = Decimal(str(ranges[0]["price"])) if ranges else Decimal("0")

        return Product(
            platform=self.platform,
            item_id=str(r.get("productID", item_id)),
            title=r.get("subject", ""),
            price=price,
            stock=int((r.get("saleInfo") or {
   }).get("amountOnSale", 0) or 0),
            images=[i["fullPathImageURI"]
                    for i in (r.get("image") or {
   }).get("images", [])],
        )

    async def get_stock(self, item_id: str, sku_id: str | None = None) -> int:
        # 显式拒绝，而不是忽略 sku_id 返回总量
        if sku_id:
            raise NotSupported(self.platform, "按 SKU 查询库存（1688 无此概念）")
        return await self._get_stock_impl(item_id, None)

    async def list_orders(self, start, end, page=1) -> list[Order]:
        d = await self.c.call("com.alibaba.trade/get.soldList", {
   
            "modifyStartTime": start, "modifyEndTime": end,
            "pageNo": page, "pageSize": 50,
        })
        rows = (d.get("result") or {
   }).get("orderList") or []
        return [Order(
            platform=self.platform,
            order_id=str(o.get("orderId")),
            status=self.normalize_status(o.get("orderStatus", "")),
            pay_amount=Decimal(str(o.get("sumPayment") or "0")),
            created_at=o.get("createTime", ""),
            modified_at=o.get("modifyTime", ""),
            buyer=o.get("toLoginId", ""),
        ) for o in rows]

# adapters/jd.py
from decimal import Decimal

from core.adapter import BaseAdapter
from core.models import Product, Order, Capability

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class JdAdapter(BaseAdapter):
    platform = "jd"
    capability = Capability(
        sku_level_stock=True,
        order_increment=True,
        logistics_subscribe=True,
    )
    STATUS_MAP = {
   
        "1": "PENDING",      # 等待付款
        "2": "PAID",         # 等待发货
        "3": "SHIPPED",      # 已发货
        "4": "DONE",         # 已完成
        "6": "CLOSED",       # 已取消
    }

    def __init__(self, client):
        self.c = client

    async def get_product(self, item_id: str) -> Product:
        d = await self.c.call("jd.union.open.goods.query", {
   
            "360buy_param_json": '{"skuIds":[%s]}' % item_id,
        })
        rows = ((d.get("jd_union_open_goods_query_responce") or {
   })
                .get("result") or {
   }).get("data") or []
        if not rows:
            return Product(platform=self.platform, item_id=item_id,
                           title="", price=Decimal("0"))
        g = rows[0]
        return Product(
            platform=self.platform,
            item_id=str(g.get("skuId", item_id)),
            title=g.get("skuName", ""),
            price=Decimal(str(g.get("price", "0"))),
            stock=None,                      # 该接口不返回库存
            images=[g["imageUrl"]] if g.get("imageUrl") else [],
        )

    async def list_orders(self, start, end, page=1) -> list[Order]:
        d = await self.c.call("jingdong.pop.order.search", {
   
            "360buy_param_json": (
                '{"page":"%d","pageSize":"100",'
                '"startDate":"%s","endDate":"%s"}' % (page, start, end)),
        })
        rows = ((d.get("jingdong_pop_order_search_responce") or {
   })
                .get("searchorderinfo_result") or {
   }).get("orderInfoList") or []
        return [Order(
            platform=self.platform,
            order_id=str(o.get("orderId")),
            status=self.normalize_status(o.get("orderState", "")),
            pay_amount=Decimal(str(o.get("orderPayment") or "0")),
            created_at=o.get("orderStartTime", ""),
            modified_at=o.get("orderModified", ""),
            buyer=o.get("pin", ""),
        ) for o in rows]

三个适配器放在一起看，共性一目了然：都要签名、都要分页、都要状态映射、都要字段归一。这些「一样的东西」全部沉到基类，适配器只保留「不一样的东西」。

这就是统一接口层真正省事的地方——不是省打字，是让「不一样」变得可枚举。

4.4 门面

# core/api.py
import asyncio

from .adapter import BaseAdapter, NotSupported


class CommerceAPI:
    """
    业务代码唯一入口。三平台一个门面，业务层不再出现 platform 判断。
    """
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def __init__(self, adapters: dict[str, BaseAdapter]):
        self.adapters = adapters

    def _get(self, platform: str) -> BaseAdapter:
        if platform not in self.adapters:
            raise KeyError(f"未注册平台：{platform}")
        return self.adapters[platform]

    # --- 单平台操作 ---

    async def product(self, platform: str, item_id: str):
        return await self._get(platform).get_product(item_id)

    async def orders(self, platform: str, start: str, end: str,
                     max_pages: int = 50) -> list:
        ad = self._get(platform)
        if not ad.capability.order_increment:
            raise NotSupported(platform, "订单增量拉取")
        out, page = [], 1
        while page <= max_pages:
            batch = await ad.list_orders(start, end, page)
            out.extend(batch)
            if len(batch) < 100:
                break
            page += 1
        return out

    # --- 跨平台操作：这才是统一接口真正的价值 ---

    async def products_all(self, item_ids_map: dict[str, list[str]],
                           concurrency: int = 10) -> dict[str, list]:
        """
        一次拿三个平台的商品。业务层不用写循环、不用管并发。
        item_ids_map = {"taobao": [...], "alibaba": [...], "jd": [...]}
        """
        sem = asyncio.Semaphore(concurrency)
        result: dict[str, list] = {
   }

        async def one(platform: str, iid: str):
            async with sem:
                try:
                    return platform, await self.product(platform, iid)
                except Exception as e:
                    print(f"[{platform}] {iid} 失败: {e}")
                    return platform, None

        tasks = [one(p, i) for p, ids in item_ids_map.items() for i in ids]
        for platform, prod in await asyncio.gather(*tasks):
            if prod is not None:
                result.setdefault(platform, []).append(prod)
        return result

    async def order_summary(self, start: str, end: str,
                            platforms: list[str] | None = None) -> dict:
        """
        跨平台订单汇总。业务层写一行，三平台自动并行。
        """
        targets = platforms or list(self.adapters.keys())

        async def one(p):
            try:
                orders = await self.orders(p, start, end)
                return p, {
   
                    "count": len(orders),
                    "amount": sum(o.pay_amount for o in orders),
                    "by_status": _group_status(orders),
                }
            except Exception as e:
                return p, {
   "error": str(e)}

        pairs = await asyncio.gather(*[one(p) for p in targets])
        return dict(pairs)


def _group_status(orders) -> dict[str, int]:
    out: dict[str, int] = {
   }
    for o in orders:
        out[o.status] = out.get(o.status, 0) + 1
    return out

order_summary 这个方法值得多看一眼。在没统一接口的项目里，这段逻辑要写三遍，每遍处理不同的分页、不同的状态码、不同的金额字段。统一之后业务层只写一行。

五、同一个需求，两种写法

需求：「拉三个平台昨天的订单，算出各平台待发货数量和金额。」

没有统一接口：

# 30 多行，而且每加一个平台就再加 10 行
result = {
   }

# 淘宝
t_orders, page = [], 1
while True:
    d = taobao.call("taobao.trades.sold.get", {
   ...})
    rows = d["trades_sold_get_response"]["trades"]["trade"]
    t_orders += rows
    if len(rows) < 100:
        break
    page += 1
result["taobao"] = {
   
    "pending": sum(1 for o in t_orders if o["status"] == "WAIT_SELLER_SEND_GOODS"),
    "amount": sum(float(o["payment"]) for o in t_orders
                  if o["status"] == "WAIT_SELLER_SEND_GOODS"),
}

# 1688（字段名完全不同）
a_orders, page = [], 1
while True:
    d = alibaba.call("com.alibaba.trade/get.soldList", {
   ...})
    rows = d["result"]["orderList"]
    a_orders += rows
    if len(rows) < 50:
        break
    page += 1
result["alibaba"] = {
   
    "pending": sum(1 for o in a_orders if o["orderStatus"] == "waitsellersend"),
    "amount": sum(float(o["sumPayment"] or 0) for o in a_orders
                  if o["orderStatus"] == "waitsellersend"),
}

# 京东（又是另一套）
...

有统一接口：

summary = await api.order_summary("2026-10-22 00:00:00", "2026-10-23 00:00:00")
pending = {
   p: v["by_status"].get("PAID", 0) for p, v in summary.items()}

从 30 行到一个方法调用。而且加第四个平台，业务代码一行都不用改。

六、能力不对等：统一接口最硬的部分

这是统一接口真正难的地方，也是最容易埋雷的地方。

三个平台的能力根本不对等：

最坑的做法是「假装都支持」：调用方传 sku_id，1688 适配器忽略这个参数返回商品总量。业务代码以为拿到了 SKU 库存，实际上是错的，而且不报错。

所以 get_stock 里那句显式 raise NotSupported 才是正确做法：

if sku_id:
    raise NotSupported(self.platform, "按 SKU 查询库存（1688 无此概念）")

配套的用法是启动时检查能力：

def check_capabilities(api: CommerceAPI, required: dict[str, bool]):
    """
    启动时校验：业务需要的能力，各平台是否都支持。
    不支持就提前暴露，而不是跑到线上才发现。
    """
    problems = []
    for platform, ad in api.adapters.items():
        for cap_name, needed in required.items():
            if needed and not getattr(ad.capability, cap_name, False):
                problems.append(f"{platform} 不支持 {cap_name}")
    if problems:
        raise RuntimeError("能力校验失败：\n" + "\n".join(problems))


# 启动时调用
check_capabilities(api, {
   
    "sku_level_stock": True,      # 业务强依赖 SKU 库存
    "order_increment": True,
})

这段会在启动时告诉你：「1688 不支持 sku_level_stock」。你可以立刻决定是降级、还是排除 1688，而不是等三天后客户投诉库存不准。

七、一套测试跑三个平台

统一接口的「统一」不能靠嘴说，得靠测试证明。用 pytest 参数化，同一套契约测试跑三个适配器：

# tests/test_contract.py
import pytest
from decimal import Decimal

from core.adapter import NotSupported

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
@pytest.fixture
def adapters():
    """三个适配器实例，用测试账号"""
    return {
   "taobao": ..., "alibaba": ..., "jd": ...}


@pytest.mark.parametrize("platform", ["taobao", "alibaba", "jd"])
@pytest.mark.asyncio
async def test_product_shape(adapters, platform, sample_ids):
    """契约一：商品必须返回统一结构"""
    p = await adapters[platform].get_product(sample_ids[platform])
    assert p.platform == platform
    assert isinstance(p.title, str)
    assert isinstance(p.price, Decimal)
    assert p.price >= 0
    assert isinstance(p.images, list)


@pytest.mark.parametrize("platform", ["taobao", "alibaba", "jd"])
@pytest.mark.asyncio
async def test_order_status_normalized(adapters, platform):
    """契约二：状态必须是五个统一值之一"""
    orders = await adapters[platform].list_orders(
        "2026-10-15 00:00:00", "2026-10-23 00:00:00")
    for o in orders:
        assert o.status in {
   "PENDING", "PAID", "SHIPPED", "DONE", "CLOSED", "UNKNOWN"}
        assert isinstance(o.pay_amount, Decimal)


@pytest.mark.asyncio
async def test_unsupported_explicit(adapters, sample_ids):
    """契约三：不支持的能力必须明确报错，不能静默降级"""
    with pytest.raises(NotSupported):
        await adapters["alibaba"].get_stock(
            sample_ids["alibaba"], sku_id="any_sku")

第三条测试是整套设计里最重要的一条。它把「不能静默降级」这个约定固化成了测试，防止后来的人图省事改成返回 0。

八、收益复盘

同一套业务代码，服务三个平台：

综合下来，一年周期省 80% 左右是站得住的。但前提是：真的长期跑，且平台数量 ≥ 2。

只接一个平台，或者只做一次性的数据导出，这个抽象就是过度设计。工程决策永远看场景。

九、四个坑

1. 别在适配器里写业务判断。 适配器只做翻译。一旦出现 if 订单金额 > 100 这种代码，说明业务逻辑漏进了适配器，将来三个平台会各写一份不一样的规则。
2. 能力声明要跟着平台文档更新。 京东的接口能力经常变（同一个能力，不同接口支持度不同）。建议每个季度对照文档核一次 Capability，把它当配置维护，不是当常量。
3. 统一状态不要贪多。 五个状态够用。给三个平台各定十个细分状态，最后业务代码里全是 if status in [...]，反而更乱。
4. 门面不要变成上帝类。 CommerceAPI 只放跨平台共性（重试、缓存、并发、能力检查）。具体业务（比如「降价 5% 就告警」）放在 service 层，别塞进门面。

十、最后

「一套接口打三平台」的价值，不在少写几行代码，在让「平台差异」变成一个有边界、可枚举、可测试的东西。

差异集中在适配器里，业务代码里就再也不该出现 if platform == "taobao"。一旦出现了，说明抽象漏了。

代码在上面，BaseAdapter + Capability + 契约测试这三件套可以直接搬到你的项目里。

如果你连适配器都不想写——签名、字段映射、状态归一、能力差异这些事，现成的三平台 API 接口服务已经做完了，输出就是统一格式，接进来只需要写业务层。省下的正是前面算的那 13 天加 16 天。

有在做的可以聊聊，你们的统一层是怎么处理能力不对等的，是抛错还是降级？

标签：Python 架构设计 电商 适配器模式 API