# 一套接口打三平台，省下 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% 工时」这种话，如果只算首次对接，是站不住的。算细一点：

### 首次对接

| 项目        | 三套独立实现        | 统一接口层             |
| --------- | ------------- | ----------------- |
| 签名/请求基础设施 | 3 × 1 天 = 3 天 | 1 天（抽象基类）         |
| 商品资源适配    | 3 × 2 天 = 6 天 | 3 × 1.5 天 = 4.5 天 |
| 订单资源适配    | 3 × 2 天 = 6 天 | 3 × 1.5 天 = 4.5 天 |
| 物流资源适配    | 3 × 1 天 = 3 天 | 3 × 0.5 天 = 1.5 天 |
| 统一层设计     | 0             | 1.5 天             |
| **合计**    | **18 天**      | **13 天**          |

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

### 后续每一次改动

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

| 变更类型      | 频率      | 三套实现                     | 统一层         |
| --------- | ------- | ------------------------ | ----------- |
| 加一个业务字段   | 每月 1 次  | 3 处 × 0.5 天 = 1.5 天      | 1 处 × 0.5 天 |
| 修一个平台 bug | 每月 2 次  | 2 × 0.5 天（有时要排查是否影响其他两份） | 0.5 天       |
| 调整业务规则    | 每季度 1 次 | 3 处 × 1 天                | 1 处 × 1 天   |
| 接第四个平台    | 每年 1 次  | 5 天                      | 1.5 天       |
| 新人接手理解成本  | 每年 1 次  | 3 天                      | 1 天         |

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

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

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

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

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

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

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

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

## 四、代码

### 4.1 统一模型和能力声明

```python
# 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 适配器基类

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

from .models import Product, Order, Capability

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
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 三个适配器（只展示关键差异）

```python
# adapters/taobao.py
from decimal import Decimal

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

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
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
```

```python
# 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]
```

```python
# adapters/jd.py
from decimal import Decimal

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


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 门面

```python
# core/api.py
import asyncio

from .adapter import BaseAdapter, NotSupported
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

class CommerceAPI:
    """
    业务代码唯一入口。三平台一个门面，业务层不再出现 platform 判断。
    """

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

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

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

**没有统一接口：**

```python
# 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"),
}

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

**有统一接口：**

```python
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 行到一个方法调用。而且**加第四个平台，业务代码一行都不用改。**

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

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

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

| 能力        | 淘宝  | 1688        | 京东        |
| --------- | --- | ----------- | --------- |
| SKU 级库存   | 支持  | 不支持（按起订量阶梯） | 部分接口支持    |
| 阶梯价       | 不支持 | 支持          | 不支持       |
| 订单按修改时间增量 | 支持  | 支持          | 部分接口只支持按天 |
| 物流订阅推送    | 支持  | 需单独申请       | 支持        |
| 商品批量查询    | 不支持 | 支持          | 支持        |

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

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

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

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

```python
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 参数化，同一套契约测试跑三个适配器：

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

from core.adapter import NotSupported


@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)

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
@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`。

## 八、收益复盘

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

| 指标     | 三套实现        | 统一接口层          |
| ------ | ----------- | -------------- |
| 首次对接   | 18 天        | 13 天（省 28%）    |
| 一年维护   | 约 43 天      | 约 16 天（省 63%）  |
| 业务层代码量 | 30 行 / 每个需求 | 3 行 / 每个需求     |
| 加第四个平台 | 约 5 天       | 约 1.5 天（省 70%） |
| 新人上手   | 读三套代码       | 读一个门面 + 三个薄适配器 |

综合下来，**一年周期省 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`

# 从 3 小时到 3 分钟：三平台商品同步是怎么提速的（附 Python 源码）

> 60 倍提速不是靠某一个技巧，是四个优化叠加。每个贡献多少，本文算给你看。

## 一、一个真实的慢

某个多平台商家，3000 个商品，每天要同步一次三平台的商品数据（标题、价格、库存、状态）。

原来的脚本：串行逐个拉，每个商品调一次详情接口，拉完逐条写库。跑一次 3 小时。

运营的抱怨很直接：「我早上想看昨天的库存，脚本要到中午才跑完，看的是上午的数据。」

更糟的是，**没人说得清 3 小时都花在哪**。有人说接口慢，有人说服务器不行，有人说数据库写不动。都在猜。

## 二、第一步：先测量，别猜

性能优化的第一原则：**先测量，再动手。** 我在这上面吃过亏——曾经花一天优化数据库索引，后来 profile 一跑发现数据库只占 2% 的时间。

给脚本加了个简单的计时器，跑一次全量，结果如下：

| 环节               | 耗时     | 占比   |
| ---------------- | ------ | ---- |
| 调接口等响应（3000 次串行） | 172 分钟 | 96%  |
| JSON 解析 + 字段映射   | 4 分钟   | 2%   |
| 逐条写数据库           | 3 分钟   | 2%   |
| 其他（重试、日志）        | 1 分钟   | 0.5% |

结论非常清晰：**96% 的时间在等网络**。

这不是「接口慢」，是**串行等 3000 次**。每个请求平均 3.4 秒，串行乘以 3000，就是 172 分钟。

这个测量直接决定了优化顺序：先解决等待（并发），再减少请求数（增量），写库那 3 分钟排最后。

测量代码很简单，就几十行，但它决定了后面所有力气往哪使：

```python
# timer.py
import time
from collections import defaultdict


class StageTimer:
    """分阶段计时，跑完输出每阶段耗时和占比"""

    def __init__(self):
        self.spent = defaultdict(float)
        self.counts = defaultdict(int)
        self._t0 = None
        self._current = None

    def start(self, stage: str):
        self._current = stage
        self._t0 = time.perf_counter()

    def stop(self):
        if self._current and self._t0 is not None:
            self.spent[self._current] += time.perf_counter() - self._t0
            self.counts[self._current] += 1
        self._current = None

    def report(self) -> list[tuple[str, float, float]]:
        total = sum(self.spent.values()) or 1.0
        rows = sorted(self.spent.items(), key=lambda x: -x[1])
        return [(name, sec, sec / total) for name, sec in rows]

    def print(self):
        print(f"\n{'环节':<24}{'次数':>8}{'耗时':>12}{'占比':>8}")
        for name, sec, pct in self.report():
            print(f"{name:<24}{self.counts[name]:>8}{sec:>10.1f}s{pct:>8.1%}")
```
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
## 三、优化一：并发——最直觉，但有天花板

96% 在等网络，那并发就是第一刀。把 3000 个串行请求改成 20 路并发：

```python
# concurrent.py
import asyncio
import time


class RateGate:
    """
    匀速限流闸门：保证全进程 QPS 不超过上限。
    比每个任务自己 sleep 更准，因为它是全局视角。
    """

    def __init__(self, qps: float):
        self.interval = 1.0 / qps
        self._next_slot = time.monotonic()
        self._lock = asyncio.Lock()

    async def wait(self):
        async with self._lock:
            now = time.monotonic()
            wait = self._next_slot - now
            if wait > 0:
                await asyncio.sleep(wait)
                self._next_slot += self.interval
            else:
                self._next_slot = now + self.interval


async def fetch_all(ids: list[str], fetch_one, concurrency: int = 20,
                    qps: float = 10.0) -> list:
    """
    并发拉取。两个旋钮：
    - concurrency：同时在飞的请求数（受限于连接和服务端）
    - qps：每秒发起多少个（受限于平台限流）
    两个都要设，只设 concurrency 会被限流打回来
    """
    gate = RateGate(qps)
    sem = asyncio.Semaphore(concurrency)
    results = [None] * len(ids)

    async def one(idx: int, item_id: str):
        async with sem:
            await gate.wait()
            results[idx] = await fetch_one(item_id)

    await asyncio.gather(*[one(i, x) for i, x in enumerate(ids)])
    return results
```

注意这段代码里有两个旋钮：`concurrency`（同时在飞的请求数）和 `qps`（每秒发起多少）。

**只设并发不设 QPS，是新手最常见的翻车点。** 20 路并发、每秒发出 20 个请求，直接撞平台限流，然后全部进入重试，反而更慢。

实测：3 小时 → 9 分钟。20 倍。

但天花板立刻出现了：平台限流 QPS 10，3000 个商品理论下限就是 300 秒 = 5 分钟。**并发再高也没用，限流是硬约束。**

所以第二刀必须换思路：不是「把请求发得更快」，而是「发更少的请求」。

## 四、优化二：少发请求——本文最值钱的洞察

重新看需求：3000 个商品每天同步。真的 3000 个都变了吗？

统计了一下：**每天真正变化的商品不到 10%**（改价、改库存、上下架）。90% 的请求在拉「没变的数据」。

那能不能先知道哪些变了，再只拉那些？

能。**商品列表接口一次能返回几十条记录的 `modified` 时间戳**——用 100 次列表请求（每次 30 条）就能拿到全部 3000 个商品的更新时间，替代 3000 次详情请求。

```python
# incremental.py
import json
from datetime import datetime

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def list_all_modified(client, shop_id: str) -> dict[str, str]:
    """
    分页拉全量商品列表，只取 item_id -> modified 映射。
    3000 个商品 = 100 次轻量请求，代替 3000 次重量请求
    """
    mapping = {}
    page = 1
    while True:
        rows = client.list_products(shop_id, page=page, page_size=30)
        if not rows:
            break
        for r in rows:
            mapping[str(r["item_id"])] = r["modified"]
        if len(rows) < 30:
            break
        page += 1
    return mapping


def pick_changed(current: dict[str, str],
                 last_sync: dict[str, str] | None) -> list[str]:
    """
    和上次同步记录对比，只返回有变化的 item_id
    新商品 + 修改过的商品 + 已删除的商品（单独处理）
    """
    last = last_sync or {}
    changed = [iid for iid, m in current.items() if last.get(iid) != m]
    removed = [iid for iid in last if iid not in current]
    if removed:
        print(f"[sync] {len(removed)} 个商品已下架，需清理")
    return changed


def save_sync_state(state_path: str, current: dict[str, str]):
    with open(state_path, "w", encoding="utf-8") as f:
        json.dump(current, f, ensure_ascii=False)


def load_sync_state(state_path: str) -> dict[str, str] | None:
    try:
        with open(state_path, encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        return None
```

账算一下：

- 原来：3000 次详情请求（每次约 3.4 秒等效负载）
- 现在：100 次列表请求 + 约 300 次详情请求 = 400 次请求

**请求数降到原来的 13%。** 这比任何并发技巧的收益都大。

这个思路的本质：**用便宜的请求替代昂贵的请求。** 列表接口一次返回 30 条的元数据，单价是详情接口的三十分之一。先便宜地筛一遍，再对筛出来的做贵的操作。

## 五、优化三 + 四：写库和连接

到这一步，网络等待只剩 40 秒左右，剩下的小头也该收一收。

**逐条写库 → 批量写。** SQLite 用 `executemany`，MySQL 用多值 INSERT：

```python
# bulk_write.py
import sqlite3


def bulk_upsert(conn: sqlite3.Connection, rows: list[tuple]):
    """
    批量 upsert。3000 条逐条写 = 3000 次事务；
    executemany 一批 = 1 次事务，差两个数量级
    """
    conn.executemany("""
        INSERT INTO products(platform, item_id, title, price, stock, synced_at)
        VALUES(?,?,?,?,?,datetime('now'))
        ON CONFLICT(platform, item_id) DO UPDATE SET
            title=excluded.title, price=excluded.price,
            stock=excluded.stock, synced_at=excluded.synced_at
    """, rows)
    conn.commit()
```

**每次新建连接 → 复用连接。** 这个最隐蔽：每次请求新建 TCP + TLS 握手要 200~~400ms，3000 次就是 10~~20 分钟。`httpx.AsyncClient` 复用连接后，握手只发生一次。

```python
# 复用连接：整个进程共享一个 client
# 错误：每次 fetch_one 里 with httpx.AsyncClient() as c: ...
# 正确：模块级或应用级创建一次
import httpx

client = httpx.AsyncClient(
    timeout=10.0,
    limits=httpx.Limits(max_connections=20, max_keepalive_connections=20),
)
```

顺带把限流做成自适应（AIMD），限流报错就减半 QPS，平稳运行则缓慢恢复：

```python
# adaptive.py
import random


class AdaptiveRate:
    """
    自适应限速（AIMD）：
    - 撞限流：QPS 减半（Multiplicative Decrease）
    - 连续 50 次成功：QPS +10%（Additive Increase）
    比写死一个保守值快得多，也比写死激进值安全
    """

    def __init__(self, qps: float = 5.0, min_qps: float = 0.5, max_qps: float = 20.0):
        self.qps = qps
        self.min_qps = min_qps
        self.max_qps = max_qps
        self._ok_streak = 0
        self._next_slot = 0.0

    def on_success(self):
        self._ok_streak += 1
        if self._ok_streak >= 50:
            self.qps = min(self.max_qps, self.qps * 1.1)
            self._ok_streak = 0

    def on_rate_limited(self):
        self.qps = max(self.min_qps, self.qps / 2)
        self._ok_streak = 0

    @property
    def interval(self) -> float:
        return 1.0 / self.qps

    async def wait(self):
        """带抖动的等待，避免多任务同刻发车"""
        import asyncio, time
        now = time.monotonic()
        wait = max(0.0, self._next_slot - now)
        if wait:
            await asyncio.sleep(wait)
        self._next_slot = max(now, self._next_slot) + self.interval * random.uniform(0.8, 1.2)
```

## 六、结果：60 倍是怎么来的

四个优化叠加，每一步的贡献都有数：

| 阶段                 | 耗时         | 相比上一步    | 手段                              |
| ------------------ | ---------- | -------- | ------------------------------- |
| 原始                 | 180 分钟     | —        | 串行 + 全量 + 逐条写                   |
| + 并发 20 + QPS 限流   | 9 分钟       | **20×**  | asyncio 并发，全局 QPS 闸门            |
| + 列表预筛增量           | 约 1.2 分钟   | **7.5×** | 100 次列表替代 2820 次详情              |
| + 批量写 + 连接复用 + 自适应 | **约 1 分钟** | 1.3×     | executemany + keep-alive + AIMD |

两个观察值得记住：

1. **最大的收益来自「少做事」（7.5×）而不是「做得快」（20× 有硬顶）。** 并发受限于平台限流，少发请求不受任何限制。
2. **最后一个优化只贡献 1.3×，但它是「从 1 分钟到 40 秒」和「偶尔翻车」的区别。** 连接复用和自适应限速不影响平均值，影响的是稳定性。

## 七、完整主流程

把四个优化串起来：

```python
# sync_main.py
import asyncio
import sqlite3

from timer import StageTimer
from concurrent import fetch_all
from incremental import (list_all_modified, pick_changed,
                         load_sync_state, save_sync_state)
from bulk_write import bulk_upsert
from adaptive import AdaptiveRate

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def run_sync(clients: dict, shop_id: str,
                   state_path: str = "sync_state.json"):
    t = StageTimer()

    # 1. 增量预筛：100 次轻量请求拿全量 modified
    t.start("列表预筛(3平台)")
    current = {}
    for platform, client in clients.items():
        current[platform] = await list_all_modified(client, shop_id)
    t.stop()

    # 2. 对比上次状态，只保留有变化的
    last = load_sync_state(state_path)
    t.start("增量对比")
    changed_map = {p: pick_changed(m, last and last.get(p))
                   for p, m in current.items()}
    t.stop()

    # 3. 并发拉详情（自适应限速）
    t.start("详情并发拉取")
    rows = []
    for platform, client in clients.items():
        ids = changed_map[platform]
        if not ids:
            continue
        rate = AdaptiveRate(qps=5.0)

        async def fetch_one(iid):
            await rate.wait()
            try:
                r = await client.get_product_detail(iid)
                rate.on_success()
                return (platform, iid, r["title"], float(r["price"]),
                        r["stock"])
            except Exception as e:
                if "limit" in str(e).lower():
                    rate.on_rate_limited()
                raise

        results = await fetch_all(ids, fetch_one,
                                  concurrency=20, qps=10.0)
        rows.extend(r for r in results if r)
    t.stop()

    # 4. 批量写库
    t.start("批量写库")
    conn = sqlite3.connect("products.db")
    bulk_upsert(conn, rows)
    conn.close()
    t.stop()

    # 5. 保存同步状态，供下次增量对比
    save_sync_state(state_path, current)
    t.print()
    return len(rows)


if __name__ == "__main__":
    from client import PlatformClient

    clients = {
        "taobao": PlatformClient("taobao", "key", "secret"),
        "alibaba": PlatformClient("alibaba", "key", "secret"),
        "jd": PlatformClient("jd", "key", "secret"),
    }
    asyncio.run(run_sync(clients, "your_shop_id"))
```

## 八、四个坑

1. **并发数和 QPS 是两个旋钮，必须分开调。** 只调并发会被限流打回来进入重试风暴，比串行还慢。
2. **增量依赖 `modified` 字段，但不是所有平台都可靠。** 有的平台批量改价不更新 `modified`。建议每周跑一次全量对账，增量为主、全量兜底——和价格同步那篇的思路一致。


1. **`fetch_one` 里闭包引用循环变量是经典 bug。** 上面代码里 `for platform, client in clients.items()` 里面定义 `fetch_one`，每次循环重新定义，没有引用问题；但如果把函数定义提到循环外再绑定变量，就会全部引用最后一次的 `client`。测试时三个平台的数据混在一起，多半是这个问题。
2. **首次运行没有同步状态，会走全量。** 这是正确行为（第一次本来就该全量），但要提醒使用者：第一次跑慢是正常的，从第二次开始才是 3 分钟。

## 九、最后

回头看这 60 倍：

- 并发（20×）：把等待重叠起来，但受限于平台限流
- 增量（7.5×）：**用便宜的请求替代昂贵的请求**，收益最大
- 批量写 + 连接复用（1.3×）：平均值贡献小，稳定性贡献大

顺序也重要：**先测量再动手**。如果先花一天优化数据库（只占 2%），一天就白费了。

代码在上面，`RateGate`、`AdaptiveRate`、增量预筛这三个组件是通用的，可以拆到任何同步场景里用。

有在做的可以聊聊，你们的同步脚本跑多久，profile 过吗？

---

**标签**：`Python` `性能优化` `电商` `异步编程` `数据同步`