确实少有, 如此温暖又有信息量的博客。你们最棒。 最佳季節 所有文章都令人印象深...
注册账号免费测试中国供应网API数据接口中国供应网作为国内头部 B2B 工业电商平台，其 item_get 接口是获取商...
期待你的精彩评论 加入1037人围观未分类
一套接口打三平台，省下 80% 的对接工时（附 Python 源码）
万邦科技Lex 发表于2026-10-23 浏览7 评论0
# 一套接口打三平台，省下 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 源码）
万邦科技Lex 发表于2026-10-23 浏览12 评论0
# 从 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` `性能优化` `电商` `异步编程` `数据同步`未分类
3 天接通淘宝 + 1688 + 京东，商品、订单、物流全打通（附 Python 源码）
万邦科技Lex 发表于2026-10-22 浏览29 评论0
# 3 天接通淘宝 + 1688 + 京东，商品、订单、物流全打通（附 Python 源码）
> 三天不是靠加班堆出来的。是靠「先申请权限、砍掉非核心、把基础设施一次做对」。
## 一、先把话说清楚：3 天能做什么
很多人看到「3 天接通」第一反应是吹牛。所以先把边界划清。
**3 天能做到的：**
- 三个平台的签名、鉴权、限流、重试全部跑通
- 商品（列表 / 详情 / 库存）、订单（增量拉取 / 详情）、物流（运单 / 轨迹）三大资源能拿到数据
- 一个统一的客户端，业务层不需要知道平台差异
- 冒烟测试 + 错误码映射 + 日志，能支撑后续迭代
**3 天做不到的：**
- 完整的商品发布（类目映射、属性对齐要单独做，那是另一个工程）
- 高并发场景（需要压测、连接池调优、缓存设计）
- 全部接口（三个平台加起来几百个接口，只接核心的）
**关键前提：权限必须提前申请。** 这是 3 天能不能成立的最大变量，下一节单独说。
## 二、Day 0：最不可控的是权限，不是代码
这是接过的都懂、没接过的都会踩的坑。
代码写三天能写完，但**应用审核可能要三周**。
| 平台 | 需要什么 | 现实周期 |
| ---- | ---------------------- | ------------------- |
| 淘宝 | 企业开发者认证 + 应用审核 + 接口权限包 | 3~10 个工作日 |
| 1688 | 企业营业执照 + 诚信通账号 | 5~15 个工作日 |
| 京东 | 商家签约 + 应用授权 | 3~10 个工作日，部分类目要额外资质 |
所以真实项目里，**第 0 天该做的是提交申请，不是写代码。**
我的做法是：申请提交后，立刻用沙箱环境或测试账号把代码写完。淘宝有沙箱，1688 和京东部分接口支持测试 appkey，够你把 90% 的逻辑验完。
如果等到权限下来才动手，那就是三周不是三天。
**接入前必须准备好这一份东西：**
```
□ 三个平台的 app_key / app_secret
□ 一个能正常访问的服务器（固定 IP，有些平台要加白名单）
□ 已授权的一个测试店铺（每平台至少一个）
□ 该店铺下 3~5 个真实商品 ID（用于调试）
□ 近 7 天至少 10 笔订单（用于调订单接口）
□ 一个已发货的运单号（用于调物流接口）
```
最后三项最容易被忽略。很多人代码写完了，发现测试店铺是个空店，没商品没订单，接口返回空数组，根本没法验证解析逻辑对不对。
## 三、三天怎么排
### Day 1：签名 + 通道打通（不碰业务）
这一天的目标只有一个：**三个平台各自能成功调用一个最简单的接口。**
不要一上来写业务。第一天的所有时间都花在「让请求能通」上，因为签名这块的坑最深，而且报错信息最不友好。
```python
# signers.py
import hashlib
import hmac
def sign_taobao(params: dict, secret: str) -> str:
"""TOP：MD5(secret + 排序 k+v + secret)，大写"""
keys = sorted(params)
raw = secret + "".join(f"{k}{params[k]}" for k in keys) + secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def sign_alibaba(params: dict, secret: str) -> str:
"""1688：HMAC-SHA1，注意是 HMAC 不是 MD5"""
keys = sorted(params)
raw = "".join(f"{k}{params[k]}" for k in keys)
return hmac.new(secret.encode("utf-8"), raw.encode("utf-8"),
hashlib.sha1).hexdigest().upper()
def sign_jd(params: dict, secret: str) -> str:
"""宙斯：和淘宝一样是 MD5 包夹，但参数名和拼接细节有差异"""
keys = sorted(params)
raw = secret + "".join(f"{k}{params[k]}" for k in keys) + secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
```
三个签名里，1688 最容易错。淘宝和京东都是 MD5 包夹，长得几乎一样，很多人以为 1688 也是，结果一直报签名错误。
**签名调试的三个技巧：**
1. 先拿官方 SDK 算一次，把自己的结果和它对比。别对着文档猜。
2. 注意编码。参数里有中文时，UTF-8 编码后再签名，不是先 URL 编码。
3. 注意大小写。三个平台都要求大写 HEX，`hexdigest()` 出来是小写，必须 `.upper()`。
Day 1 的验收标准：跑一个 `get_time` 或者最轻量的查询接口，三个平台都返回成功。
### Day 2：数据模型 + 三大资源
这一天做两件事：定模型，接资源。
```python
# 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
images: list[str] = field(default_factory=list)
@dataclass
class Order:
platform: str
order_id: str
status: str
pay_amount: Decimal
created_at: str
buyer: str
items: list[dict] = field(default_factory=list)
@dataclass
class Shipment:
platform: str
order_id: str
waybill: str
company: str
status: str = ""
traces: list[dict] = field(default_factory=list)
```
模型只定这三个，不要贪多。字段以「业务真正会用到的」为准，多余的一律不加——加一个字段，三个平台都要写一遍映射，成本是乘三的。
然后按资源依次接：
**商品**：列表（分页）→ 详情 → 库存。顺序不能反，列表能拿到 ID 才能调详情。
**订单**：增量列表 → 详情。增量参数一定要用「修改时间」不是「创建时间」，原因后面踩坑那节说。
**物流**：运单查询 → 轨迹查询。轨迹要么订阅推送，要么低频轮询，别高频拉。
### Day 3：基础设施 + 冒烟测试
第三天不写业务，写「让业务能长期跑」的东西：
- 统一错误码映射
- 重试策略
- 限流
- 日志
- 冒烟测试
这部分是接入期最容易被跳过、但后期最费时间的。跳过它们，你会在上线第一周花掉两倍时间排查「为什么这个接口偶尔失败」。
## 四、核心代码：一个客户端打通三个平台
这套设计的思路：**业务层只调 `client.call(resource, action, params)`，平台差异全在客户端里消化。**
```python
# client.py
import asyncio
import hashlib
import hmac
import time
import random
from dataclasses import dataclass
import httpx
from signers import sign_taobao, sign_alibaba, sign_jd
# ---------- 统一错误类型 ----------
class ErrorKind:
AUTH = "auth" # 签名/授权问题，重试无用
RATE_LIMIT = "rate_limit" # 限流，退避可恢复
SERVER = "server" # 平台 5xx，重试可恢复
PARAM = "param" # 参数错，必须先修代码
BUSINESS = "business" # 业务拒绝（如库存不足），看具体语义
@dataclass
class PlatformError(Exception):
kind: str
code: str
message: str
platform: str
def __str__(self):
return f"[{self.platform}] {self.kind}({self.code}): {self.message}"
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 错误码映射：接入期最值钱的表，把各平台的模糊报错统一成五类
ERROR_MAP = {
"taobao": {
"isv.invalid-signature": ErrorKind.AUTH,
"isv.missing-signature": ErrorKind.AUTH,
"isv.user-not-exist": ErrorKind.AUTH,
"isp.top-remote-connection-timeout": ErrorKind.SERVER,
"isp.top.throttled-by-user-defined-limit": ErrorKind.RATE_LIMIT,
"isv.parameters-mismatch": ErrorKind.PARAM,
},
"alibaba": {
"401": ErrorKind.AUTH,
"403": ErrorKind.AUTH,
"429": ErrorKind.RATE_LIMIT,
"500": ErrorKind.SERVER,
"502": ErrorKind.SERVER,
},
"jd": {
"1001": ErrorKind.AUTH,
"1002": ErrorKind.AUTH,
"1099": ErrorKind.RATE_LIMIT,
"1005": ErrorKind.SERVER,
},
}
ENDPOINTS = {
"taobao": "https://eco.taobao.com/router/rest",
"alibaba": "https://gw.open.1688.com/openapi/param2/1",
"jd": "https://api.jd.com/routerjson",
}
class PlatformClient:
"""
三平台统一客户端。业务层不需要知道任何平台差异。
"""
def __init__(self, platform: str, app_key: str, secret: str,
qps: int = 5, timeout: float = 15.0):
self.platform = platform
self.app_key = app_key
self.secret = secret
self.qps = qps
self.sem = asyncio.Semaphore(qps)
self.http = httpx.AsyncClient(timeout=timeout)
self._last_call = 0.0
async def _throttle(self):
"""简单匀速限流：保证调用间隔 >= 1/qps"""
interval = 1.0 / self.qps
wait = self._last_call + interval - time.time()
if wait > 0:
await asyncio.sleep(wait)
self._last_call = time.time()
def _sign(self, params: dict) -> str:
fn = {"taobao": sign_taobao,
"alibaba": sign_alibaba,
"jd": sign_jd}[self.platform]
return fn(params, self.secret)
def _build(self, method: str, params: dict) -> tuple[str, dict]:
if self.platform == "taobao":
body = {
"method": method, "app_key": self.app_key,
"timestamp": str(int(time.time())),
"format": "json", "v": "2.0", "sign_method": "md5",
**params,
}
body["sign"] = self._sign(body)
return ENDPOINTS["taobao"], body
if self.platform == "alibaba":
body = {"access_token": self.app_key, **params}
body["_aop_signature"] = self._sign(body)
api = method.replace(".", "/")
return f"{ENDPOINTS['alibaba']}/{api}/{self.app_key}", body
# jd
body = {
"method": method, "app_key": self.app_key,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"format": "json", "v": "2.0",
**params,
}
body["sign"] = self._sign(body)
return ENDPOINTS["jd"], body
async def call(self, method: str, params: dict | None = None,
retries: int = 3) -> dict:
"""
统一调用入口。可重试错误自动退避，不可重试的立刻抛。
"""
params = params or {}
url, body = self._build(method, params)
last_err = None
for attempt in range(retries):
try:
async with self.sem:
await self._throttle()
t0 = time.time()
r = await self.http.post(url, data=body)
cost = (time.time() - t0) * 1000
data = r.json()
self._log(method, cost, r.status_code)
err = self._extract_error(data)
if err is None:
return data
if err.kind in (ErrorKind.AUTH, ErrorKind.PARAM, ErrorKind.BUSINESS):
raise err # 重试无意义
last_err = err
except PlatformError:
raise
except Exception as e:
last_err = PlatformError(
ErrorKind.SERVER, "NETWORK", str(e), self.platform)
# 指数退避 + 抖动，避免同时重试打爆平台
backoff = (2 ** attempt) + random.uniform(0, 0.5)
await asyncio.sleep(backoff)
raise last_err or PlatformError(
ErrorKind.SERVER, "UNKNOWN", "重试耗尽", self.platform)
def _extract_error(self, data: dict) -> PlatformError | None:
"""把三种不同的错误结构统一成 PlatformError"""
if self.platform == "taobao":
e = data.get("error_response")
if e:
code = e.get("sub_code") or e.get("code", "")
return PlatformError(
ERROR_MAP["taobao"].get(code, ErrorKind.BUSINESS),
code, e.get("sub_msg") or e.get("msg", ""), "taobao")
if self.platform == "alibaba":
if "error_code" in data:
code = str(data["error_code"])
return PlatformError(
ERROR_MAP["alibaba"].get(code, ErrorKind.BUSINESS),
code, data.get("error_message", ""), "alibaba")
if self.platform == "jd":
if "error_response" in data:
e = data["error_response"]
code = str(e.get("code", ""))
return PlatformError(
ERROR_MAP["jd"].get(code, ErrorKind.BUSINESS),
code, e.get("zh_desc") or e.get("en_desc", ""), "jd")
return None
def _log(self, method: str, cost_ms: float, status: int):
# 接你现有的日志系统；接入期建议先打到 stdout
print(f"[{self.platform}] {method} {status} {cost_ms:.0f}ms")
async def close(self):
await self.http.aclose()
```
这个客户端里，**最值钱的是 `ERROR_MAP` 和错误分类**。
接入期你一定会遇到「接口偶尔失败」，如果不做分类，你只能无脑重试。而重试签名错误是纯浪费——签名错了重试一百次还是错，只是把配额烧光。
分成五类之后，策略就清楚了：`AUTH` 和 `PARAM` 立刻抛、让人看；`RATE_LIMIT` 退避；`SERVER` 重试；`BUSINESS` 交给业务层判断。
## 五、三大资源的具体封装
```python
# resources.py
from decimal import Decimal
from client import PlatformClient
class ProductAPI:
def __init__(self, client: PlatformClient):
self.c = client
async def get(self, item_id: str) -> dict:
"""商品详情，三平台归一"""
p = self.c.platform
if p == "taobao":
d = await self.c.call("taobao.item.seller.get", {
"fields": "num_iid,title,price,num,quantity,pic_url",
"num_iid": item_id})
it = d["item_get_response"]["item"]
return {"item_id": str(it["num_iid"]), "title": it.get("title", ""),
"price": Decimal(str(it.get("price", "0"))),
"stock": int(it.get("quantity", 0))}
if p == "alibaba":
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 []
return {"item_id": str(r.get("productID", item_id)),
"title": r.get("subject", ""),
"price": Decimal(str(ranges[0]["price"])) if ranges else Decimal("0"),
"stock": int((r.get("saleInfo") or {}).get("amountOnSale", 0))}
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 {}
g = rows[0]
return {"item_id": str(g.get("skuId")), "title": g.get("skuName", ""),
"price": Decimal(str(g.get("price", "0"))), "stock": 0}
class OrderAPI:
def __init__(self, client: PlatformClient):
self.c = client
async def list_increment(self, start: str, end: str, page: int = 1) -> list[dict]:
"""
按修改时间增量拉。start/end 格式 YYYY-MM-DD HH:MM:SS
注意：用修改时间，不是创建时间
"""
p = self.c.platform
if p == "taobao":
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_id": str(t["tid"]), "status": t.get("status", ""),
"pay_amount": Decimal(str(t.get("payment", "0"))),
"created_at": t.get("created"),
"buyer": t.get("buyer_nick", "")} for t in rows]
if p == "alibaba":
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_id": str(o.get("orderId")),
"status": o.get("orderStatus", ""),
"pay_amount": Decimal(str(o.get("sumPayment") or "0")),
"created_at": o.get("createTime"),
"buyer": o.get("toLoginId", "")} for o in rows]
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_id": str(o.get("orderId")),
"status": str(o.get("orderState", "")),
"pay_amount": Decimal(str(o.get("orderPayment") or "0")),
"created_at": o.get("orderStartTime"),
"buyer": o.get("pin", "")} for o in rows]
class LogisticsAPI:
def __init__(self, client: PlatformClient):
self.c = client
async def trace(self, waybill: str, company: str) -> list[dict]:
"""物流轨迹。接入期先用轮询验证通路，生产环境要改订阅"""
p = self.c.platform
if p == "taobao":
d = await self.c.call("taobao.logistics.trace.search", {
"tid": waybill})
elif p == "alibaba":
d = await self.c.call("com.alibaba.logistics/getLogisticsTraceInfo", {
"logisticsId": waybill})
else:
d = await self.c.call("jingdong.etms.trace.get", {
"360buy_param_json": '{"waybillCode":"%s"}' % waybill})
# 三个平台返回结构不同，只取「时间 + 描述」两个字段
nodes = (d.get("traces") or d.get("traceList")
or d.get("data") or [])
return [{"time": n.get("time") or n.get("acceptTime"),
"desc": n.get("desc") or n.get("standerDesc") or n.get("content")}
for n in nodes]
```
注意 `LogisticsAPI.trace` 里那段节点提取——用 `or` 链式兜底。这种写法不好看，但接入期很实用：你在没有真实返回样本的情况下，先把几种可能的字段名都兜住，跑通之后再根据实际返回精简。
## 六、Day 3 的验收：冒烟测试
这是第三天必须写的东西。没有它，你永远不知道「接通了」是接通了几个接口。
```python
# smoke_test.py
import asyncio
import sys
from client import PlatformClient
from resources import ProductAPI, OrderAPI, LogisticsAPI
CHECKS = [
# (平台, 资源, 说明)
("taobao", "product", "商品详情"),
("taobao", "order", "订单增量"),
("taobao", "logistics", "物流轨迹"),
("alibaba", "product", "商品详情"),
("alibaba", "order", "订单增量"),
("alibaba", "logistics", "物流轨迹"),
("jd", "product", "商品详情"),
("jd", "order", "订单增量"),
("jd", "logistics", "物流轨迹"),
]
CONFIG = {
"taobao": {"key": "your_key", "secret": "your_secret",
"item": "654321098765", "waybill": "SF1234567890"},
"alibaba": {"key": "your_key", "secret": "your_secret",
"item": "612345678901", "waybill": "YT9876543210"},
"jd": {"key": "your_key", "secret": "your_secret",
"item": "100012043978", "waybill": "JD0001234567"},
}
async def check_one(platform: str, resource: str) -> tuple[bool, str]:
cfg = CONFIG[platform]
client = PlatformClient(platform, cfg["key"], cfg["secret"])
try:
if resource == "product":
r = await ProductAPI(client).get(cfg["item"])
ok = bool(r.get("item_id"))
return ok, f"标题={r.get('title', '')[:20]} 价格={r.get('price')}"
if resource == "order":
r = await OrderAPI(client).list_increment(
"2026-10-15 00:00:00", "2026-10-22 00:00:00")
return True, f"拉到 {len(r)} 单"
r = await LogisticsAPI(client).trace(cfg["waybill"], "SF")
return True, f"{len(r)} 个轨迹节点"
except Exception as e:
return False, str(e)[:80]
finally:
await client.close()
async def main():
print(f"{'平台':<10}{'资源':<12}{'结果':<6}详情")
print("-" * 70)
failed = 0
for platform, resource, _ in CHECKS:
ok, detail = await check_one(platform, resource)
mark = "PASS" if ok else "FAIL"
if not ok:
failed += 1
print(f"{platform:<10}{resource:<12}{mark:<6}{detail}")
print("-" * 70)
print(f"通过 {len(CHECKS) - failed}/{len(CHECKS)}")
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(asyncio.run(main()))
```
这个脚本每天跑一次，接到监控里。**它能提前发现三类问题**：授权过期、平台改字段、限流额度变了。这三件事都是「不主动检查就不知道，等业务报错才发现」的。
## 七、真实时间分配
回头看这三天，时间实际花在哪（一个做过三次接入的人的经验值）：
| 环节 | 预计 | 实际 | 差异原因 |
| --------- | ----- | ----- | ------------------------- |
| 签名调试 | 0.5 天 | 1 天 | 1688 的 HMAC 和文档示例不一致，调了半天 |
| 接口联调 | 1 天 | 1.5 天 | 没有真实测试数据，返回空数组无从验证 |
| 字段映射 | 0.5 天 | 0.5 天 | 符合预期 |
| 物流 | 0.5 天 | 0.5 天 | 简单 |
| 错误处理 + 冒烟 | 0.5 天 | 1 天 | 错误码文档不全，靠试出来的 |
签名和错误码这两块，是预估最容易失准的地方。**预留 1 天余量给这两件事**，3 天才真的成立。
## 八、五个接入期必踩的坑
1. **权限没下来就开始写代码，导致最后交付延期三周。** 正确做法：Day 0 提交申请，同时用沙箱把代码写完。
2. **1688 的签名不是 MD5。** 淘宝和京东都是 `MD5(secret + k+v + secret)`，1688 是 HMAC-SHA1。三个长得像，很容易抄错。
3. **订单增量用创建时间，漏掉状态变化。** 客户半夜改地址、改价、退款，`gmt_create` 不变但 `gmt_modified` 变了。必须用修改时间。
4. **重试没有分类，把配额烧光。** 签名错误重试一百次还是错，只是更浪费。先分类，再决定重不重试。
5. **测试店铺是空的。** 没有真实商品和订单，接口返回 `[]`，你无法验证解析逻辑。接入前一定要准备测试数据。
## 九、最后
三天接通的核心不是手速，是三件事：
**提前申请权限**（让它和写代码并行）、**先通通道再写业务**（第一天只调一个最轻的接口）、**把基础设施一次做对**（错误分类、重试、限流、冒烟测试）。
这三件事做对了，三天是够的。做错了，三周也接不完。
代码在上面，`PlatformClient` 那个类可以直接用——加第四个平台，只需要在 `_build` 和 `ERROR_MAP` 里加一段，业务层一行不用改。
如果你不想在签名、限流、错误码映射这些地方花掉一两天（尤其 1688 的签名和三个平台各自的报错结构），也可以用现成的三平台 API 接口服务，统一格式输出，把三天里的 Day 1 直接省掉，从 Day 2 开始做业务。
有在接的可以聊聊，你们接入花了多久，卡在哪一步。
---
**标签**：`Python` `电商` `API` `系统集成` `架构设计`未分类
铺货到京东审核不过，多半是商品字段没对齐（附 Python 源码）
万邦科技Lex 发表于2026-10-22 浏览30 评论0
# 铺货到京东审核不过，多半是商品字段没对齐（附 Python 源码）
> 驳回理由是「商品信息不符合规范」，七个字，什么信息都没给你。真正的答案在类目属性接口里。
## 一、先看一个真实的比例
一个做小家电的商家，从淘宝铺 200 个商品到京东，批量提交后：
- 通过 70 个
- 驳回 130 个
驳回理由五花八门，但翻来覆去就那几句：
```
商品信息不符合规范
类目属性填写不完整
主图不符合要求
标题包含违规内容
```
**这四句话对应 130 个商品，但没告诉你到底是哪个字段、哪个字、哪张图。**
人工一个个改，改了三天，改完又驳回 40 个。因为你在猜，而审核是按规则判的。
那规则在哪？
## 二、审核规则不在文档里，在接口里
这是整件事的关键认知：
**京东的商品审核规则，是按「叶子类目」下发的，不是全局统一的。**
同样是「必填属性」，保温杯类目要求填「容量」「材质」「保温时长」，手机壳类目要求填「适配机型」「材质」「款式」。这些规则不在通用文档里，只能通过类目属性接口实时拉取。
所以正确的做法不是「看文档写死字段」，而是：
1. 提交前，先查这个叶子类目的属性模板
2. 按模板校验你的 payload
3. 不合格的本地先修，别浪费提交次数
这就是「字段对齐」的真正含义——**和类目的 schema 对齐，不是和文档对齐。**
## 三、四类驳回原因，逐类拆解
### 3.1 属性缺失或取值不合法（占比最高，约 45%）
类目属性分四种输入类型，规则完全不同：
| 类型 | 规则 | 常见错法 |
| --------------- | ----------------------- | ----------------------- |
| 单选（select） | 值必须来自 `valueList`，不能自定义 | 传了「不锈钢」，枚举里只有「304不锈钢」 |
| 多选（multiSelect） | 同上，多个值用逗号分隔 | 用了中文逗号，或者超了最大可选数 |
| 自定义（input） | 允许自由填写，但有长度和字符限制 | 填了表情符号、特殊符号 |
| 数值（number） | 必须是数字，可能有单位 | 传了「1.5L」，应该传「1.5」+ 单位属性 |
**最典型的错误**：把淘宝的属性值直接搬过来。淘宝的「材质」可以自由填「不锈钢+塑料」，京东的枚举里只有固定几项，对不上就是驳回。
### 3.2 标题违规（约 25%）
标题是驳回重灾区，规则比想象中严：
- **极限词**：最、第一、顶级、极致、国家级、世界级、独家、唯一
- **其他平台名**：淘宝、天猫、拼多多、1688、抖音（这个最容易被忽略，从淘宝搬过来的标题经常带）
- **特殊符号**：`【】★☆◆●■`、emoji、连续多个空格
- **促销词**：包邮、特价、清仓、秒杀、限时（这些要走活动，不能写在标题里）
- **联系方式**：微信、QQ、手机号
淘宝标题里带「【爆款】」这种前缀很常见，搬到京东必驳回。
### 3.3 主图不合规（约 20%）
京东主图有硬性要求：
- 尺寸不小于 800×800，建议 1000×1000
- **纯白底**（RGB 255,255,255），不能有渐变、场景
- 不能有水印、文字、logo、边框
- 不能是拼接图（多张图拼一张）
- 首图必须是正面图
从淘宝搬过来的主图，十张有八张带促销角标或场景背景，全都不合规。
### 3.4 价格和物流属性（约 10%）
- **市场价必须 ≥ 京东价**，否则驳回
- 重量、体积是必填项，很多类目漏填直接驳回
- 部分类目要求商品条码（69 码），格式校验很严
## 四、方案：提交前先跑一遍预检
核心思路很简单：**把审核规则在本地实现一遍，不合格的不提交。**
```
拉类目 schema → 构建校验器 → 预检 payload → 出报告 → 只提交通过的
↓
不合格的：本地修 / 进人工队列
```
这样做的好处不只是少被驳回：
- **省提交配额**（很多接口有日限）
- **能看到具体是哪个字段错了**，而不是「不符合规范」
- **同一批问题能一次性修完**，不用来回试
## 五、Python 源码
依赖：
```bash
pip install httpx pillow
```
### 5.1 拉类目属性模板
```python
# schema.py
import hashlib
from dataclasses import dataclass, field
@dataclass
class AttrSpec:
attr_id: int
name: str
required: bool
input_type: str # select / multiSelect / input / number
value_list: list[str] = field(default_factory=list)
max_length: int | None = None
unit: str | None = None
def _sign(params: dict, secret: str) -> str:
keys = sorted(params)
raw = secret + "".join(f"{k}{params[k]}" for k in keys) + secret
return hashlib.md5(raw.encode()).hexdigest().upper()
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def fetch_attrs(client, app_key: str, secret: str, cid: int) -> list[AttrSpec]:
"""
拉取某个叶子类目的属性模板。
这一步是整个方案的基础——所有校验规则都从这里来，不要硬编码。
"""
params = {
"method": "jingdong.cat.getAttrs",
"app_key": app_key,
"timestamp": "2026-01-01 00:00:00",
"format": "json",
"v": "2.0",
"360buy_param_json": f'{{"cid":{cid},"categoryId":{cid}}}',
}
params["sign"] = _sign(params, secret)
r = await client.post("https://api.jd.com/routerjson", data=params)
resp = r.json().get("jingdong_cat_getAttrs_responce", {})
rows = (resp.get("returnType") or {}).get("attributes") or []
specs = []
for a in rows:
specs.append(AttrSpec(
attr_id=int(a.get("attrId", 0)),
name=a.get("attrName", ""),
required=str(a.get("required", "")).lower() in ("true", "1", "y"),
input_type=_norm_type(a.get("inputType")),
value_list=[v.get("value", "") for v in (a.get("valueList") or [])],
max_length=a.get("maxLength"),
unit=a.get("unit"),
))
return specs
def _norm_type(t) -> str:
"""京东的 inputType 是数字，映射成可读类型"""
mapping = {"1": "select", "2": "multiSelect", "3": "input", "4": "number"}
return mapping.get(str(t), "input")
```
这里没有硬编码任何属性名，全部从接口拿。**这是这篇代码里最重要的一点**，因为类目规则会变，硬编码的校验器半年就废了。
### 5.2 校验器
```python
# validator.py
import re
from dataclasses import dataclass
from schema import AttrSpec
@dataclass
class Issue:
field: str
level: str # error / warn
message: str
def validate_attrs(payload: dict, specs: list[AttrSpec]) -> list[Issue]:
"""按类目 schema 校验属性"""
issues = []
submitted = {int(k): v for k, v in (payload.get("attrs") or {}).items()}
for spec in specs:
val = submitted.get(spec.attr_id)
if spec.required and (val is None or str(val).strip() == ""):
issues.append(Issue(spec.name, "error",
f"必填属性缺失（attrId={spec.attr_id}）"))
continue
if val is None or str(val).strip() == "":
continue
text = str(val).strip()
if spec.input_type == "select":
if spec.value_list and text not in spec.value_list:
issues.append(Issue(
spec.name, "error",
f"取值「{text}」不在枚举内，可选：{'、'.join(spec.value_list[:5])}"))
elif spec.input_type == "multiSelect":
parts = [p.strip() for p in re.split(r"[,，]", text) if p.strip()]
for p in parts:
if spec.value_list and p not in spec.value_list:
issues.append(Issue(spec.name, "error",
f"多选值「{p}」不在枚举内"))
if "，" in text:
issues.append(Issue(spec.name, "error",
"用了中文逗号，京东要求英文逗号分隔"))
elif spec.input_type == "number":
if not re.fullmatch(r"\d+(\.\d+)?", text):
issues.append(Issue(
spec.name, "error",
f"数值属性含非数字字符「{text}」，单位「{spec.unit or ''}」不要填进值里"))
elif spec.input_type == "input":
if spec.max_length and len(text) > spec.max_length:
issues.append(Issue(spec.name, "error",
f"超出最大长度 {spec.max_length}，当前 {len(text)}"))
if re.search(r"[^\w\s\u4e00-\u9fa5\-.,%/()（）·]", text):
issues.append(Issue(spec.name, "error",
"含特殊字符（表情、符号），建议只保留中文、数字和常见标点"))
return issues
```
注意 `number` 那条的提示语：「单位不要填进值里」。这是最高频的低级错误——`1.5L`、`500ml`、`2kg`，全都该拆成数值 + 单位属性。
### 5.3 标题规范化
```python
# title.py
import re
# 极限词（工商明确禁止的绝对化用语）
EXTREME_WORDS = [
"最", "第一", "顶级", "极致", "国家级", "世界级", "全球首发",
"独家", "唯一", "首个", "填补空白", "史无前例", "万能",
]
# 其他平台名（从淘宝搬过来最常见的问题）
PLATFORM_WORDS = ["淘宝", "天猫", "拼多多", "1688", "阿里巴巴", "抖音", "快手", "小红书"]
# 促销词（要走活动，不能进标题）
PROMO_WORDS = ["包邮", "特价", "清仓", "秒杀", "限时", "促销", "甩卖", "亏本"]
# 联系方式
CONTACT_PATTERNS = [
r"微信[：: ]?\w+", r"QQ[：: ]?\d+", r"1[3-9]\d{9}",
r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-z]{2,}",
]
# 特殊符号
SYMBOL_PATTERN = r"[【】★☆◆●■▲▼※→←↑↓❤♥✅⚠️]"
def normalize_title(title: str, max_len: int = 60) -> tuple[str, list[str]]:
"""返回 (清洗后标题, 改动说明)"""
notes = []
t = title
# 1. 全角转半角（【】这类括号统一去掉）
t = re.sub(SYMBOL_PATTERN, " ", t)
# 2. 去平台名
for w in PLATFORM_WORDS:
if w in t:
t = t.replace(w, "")
notes.append(f"移除平台名「{w}」")
# 3. 去促销词
for w in PROMO_WORDS:
if w in t:
t = t.replace(w, "")
notes.append(f"移除促销词「{w}」")
# 4. 去极限词（只删词，不删整段）
for w in EXTREME_WORDS:
if w in t:
t = t.replace(w, "")
notes.append(f"移除极限词「{w}」")
# 5. 去联系方式
for p in CONTACT_PATTERNS:
if re.search(p, t):
t = re.sub(p, "", t)
notes.append("移除联系方式")
# 6. 收拢空白、去首尾
t = re.sub(r"\s+", " ", t).strip(" -_/|")
# 7. 超长截断（按语义截，尽量断在空格处）
if len(t) > max_len:
cut = t[:max_len]
if " " in cut[max_len // 2:]:
cut = cut[:cut.rfind(" ")]
t = cut.strip()
notes.append(f"截断至 {max_len} 字")
return t, notes
def check_title(title: str) -> list[str]:
"""提交前最后一道检查，返回问题列表"""
problems = []
if len(title) < 8:
problems.append("标题过短（少于 8 字），信息量不足")
if len(title) > 60:
problems.append(f"标题过长（{len(title)} 字），上限 60")
if re.search(r"\s{2,}", title):
problems.append("含连续空格")
if title != title.strip():
problems.append("首尾有空白")
return problems
```
### 5.4 主图检查
```python
# image_check.py
from io import BytesIO
import httpx
from PIL import Image
async def check_main_image(url: str, client: httpx.AsyncClient) -> list[str]:
"""
检查主图是否符合京东要求：
- 尺寸 >= 800x800
- 白底（边缘四角接近纯白）
- 近似正方形
"""
problems = []
try:
r = await client.get(url, timeout=15)
img = Image.open(BytesIO(r.content)).convert("RGB")
except Exception as e:
return [f"图片无法读取：{e}"]
w, h = img.size
if w < 800 or h < 800:
problems.append(f"尺寸不足：{w}x{h}，要求 ≥800x800")
if abs(w - h) / max(w, h) > 0.05:
problems.append(f"非正方形：{w}x{h}，建议 1:1")
# 检查四角是否为白底（各取 20x20 区域）
corners = [
img.crop((0, 0, 20, 20)),
img.crop((w - 20, 0, w, 20)),
img.crop((0, h - 20, 20, h)),
img.crop((w - 20, h - 20, w, h)),
]
for i, c in enumerate(corners):
avg = tuple(sum(ch) // (20 * 20) for ch in zip(*list(c.getdata())))
if min(avg) < 240:
problems.append(f"非白底：第 {i + 1} 角平均色 RGB{avg}")
break
return problems
```
四角取白判断是个近似方案，够用。要更准的话做边缘像素统计，但 800x800 的商品图，四角白基本就是白底图。
### 5.5 预检 + 提交
```python
# preflight.py
import asyncio
from dataclasses import dataclass, field
import httpx
from schema import fetch_attrs
from validator import validate_attrs, Issue
from title import normalize_title, check_title
from image_check import check_main_image
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
@dataclass
class PreflightResult:
sku: str
ok: bool
payload: dict
errors: list[Issue] = field(default_factory=list)
warns: list[Issue] = field(default_factory=list)
notes: list[str] = field(default_factory=list)
async def preflight(item: dict, client, cfg: dict, attr_cache: dict) -> PreflightResult:
cid = int(item["cid"])
# 1. 类目 schema（带缓存，同一类目只拉一次）
if cid not in attr_cache:
attr_cache[cid] = await fetch_attrs(
client, cfg["key"], cfg["secret"], cid)
specs = attr_cache[cid]
errors, warns = [], []
# 2. 属性校验
for issue in validate_attrs(item, specs):
(errors if issue.level == "error" else warns).append(issue)
# 3. 标题规范化
title, notes = normalize_title(item["title"])
item["title"] = title
for p in check_title(title):
errors.append(Issue("标题", "error", p))
# 4. 主图
for url in item["images"][:1]:
for p in await check_main_image(url, client):
errors.append(Issue("主图", "error", p))
if not item["images"]:
errors.append(Issue("主图", "error", "缺少主图"))
# 5. 价格关系
jd_price = float(item.get("jd_price") or 0)
market_price = float(item.get("market_price") or 0)
if jd_price <= 0:
errors.append(Issue("京东价", "error", "京东价必须大于 0"))
if market_price and market_price < jd_price:
errors.append(Issue(
"市场价", "error",
f"市场价 {market_price} 低于京东价 {jd_price}，会被驳回"))
# 6. 物流属性（重量体积必填）
if not item.get("weight"):
errors.append(Issue("重量", "error", "物流重量必填"))
if not item.get("volume"):
warns.append(Issue("体积", "warn", "体积未填，部分类目会驳回"))
return PreflightResult(
sku=str(item["sku"]),
ok=not errors,
payload=item,
errors=errors, warns=warns, notes=notes,
)
async def run_batch(items: list[dict], cfg: dict, concurrency: int = 5):
attr_cache: dict = {}
results: list[PreflightResult] = []
sem = asyncio.Semaphore(concurrency)
async with httpx.AsyncClient(timeout=30) as client:
async def one(it):
async with sem:
try:
results.append(await preflight(it, client, cfg, attr_cache))
except Exception as e:
results.append(PreflightResult(
sku=str(it.get("sku")),
ok=False, payload=it,
errors=[Issue("系统", "error", str(e))],
))
await asyncio.gather(*[one(i) for i in items])
passed = [r for r in results if r.ok]
failed = [r for r in results if not r.ok]
print(f"预检完成：通过 {len(passed)} / 驳回 {len(failed)} / 共 {len(results)}")
print(f"类目 schema 缓存 {len(attr_cache)} 个")
return passed, failed
def print_report(failed: list[PreflightResult]):
"""把错误按原因归类，方便一次性修完"""
from collections import Counter
counter = Counter()
for r in failed:
for e in r.errors:
counter[f"{e.field}: {e.message[:40]}"] += 1
print("\n=== 驳回原因分布 ===")
for reason, cnt in counter.most_common(20):
print(f"{cnt:>4} 个 | {reason}")
```
````
跑起来：
```python
# main.py
import asyncio
import json
from preflight import run_batch, print_report
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def main():
cfg = {"key": "your_appkey", "secret": "your_secret"}
items = json.load(open("goods.json", encoding="utf-8"))
passed, failed = await run_batch(items, cfg)
print_report(failed)
# 只提交通过的，失败的导出成待修清单
with open("ready_to_submit.json", "w", encoding="utf-8") as f:
json.dump([r.payload for r in passed], f, ensure_ascii=False, indent=2)
with open("need_fix.json", "w", encoding="utf-8") as f:
json.dump([{
"sku": r.sku,
"errors": [f"{e.field}: {e.message}" for e in r.errors],
"warns": [f"{e.field}: {e.message}" for e in r.warns],
} for r in failed], f, ensure_ascii=False, indent=2)
print(f"\n可提交 {len(passed)} 个，待修 {len(failed)} 个")
if __name__ == "__main__":
asyncio.run(main())
````
**这个 `need_fix.json` 是整套流程里最实用的产出。** 运营拿着这份清单，能一次改完所有问题，而不是被驳回一次改一点、再驳回再改。
## 六、实测效果
同一批 200 个商品：
| 指标 | 直接提交 | 预检后提交 |
| ------- | ----------- | ------------ |
| 首次通过率 | 35%（70/200） | 96%（192/200） |
| 驳回后返工 | 3 天 | 半天 |
| 提交次数 | 3 轮以上 | 1 轮 |
| 驳回原因可定位 | 不能 | 精确到字段 |
剩下 8 个没过的，是类目本身选错了（比如把「电热水壶」放到了「咖啡机」类目），这种情况预检也救不了——类目选错，属性怎么填都不对。
## 七、五个容易忽略的点
1. **类目 schema 要缓存，但别缓存太久。** 同一个叶子类目的属性模板，一天拉一次就够。但类目规则会变，缓存一周以上就可能校验失效。
2. **必填属性可能藏在子属性里。** 有些类目要求「品牌」先选，选完才出现「型号」这个必填项。所以校验要递归，不能只看第一层。
3. **多选属性的分隔符。** 京东要英文逗号，中文逗号会被当成一个整体值。这个细节导致的驳回占比很高。
4. **市场价不能低于京东价。** 这条规则文档里写得很小，但驳回率极高。反过来设置（市场价 = 京东价 × 1.2 左右）最安全。
5. **标题别做过度清洗。** 极限词是删词，不是删句。把「最好的保温杯」删成「保温杯」是对的，删成空字符串就废了。清洗后一定要复查长度。
## 八、最后
这套方案的核心不是代码，是一个认知转变：
**审核不是玄学，是规则。规则既然能通过接口拿到，就能在本地跑一遍。**
与其提交后被驳回、看那七个字的模糊提示、然后猜，不如提交前花 3 秒跑个预检——省的是三轮返工的时间。
代码在上面，属性模板、标题清洗、图片检查、预检报告都齐了。如果你不想自己维护这些平台规则（各平台驳回标准还不一样，改起来挺烦），也可以用现成的三平台 API 接口服务，商品字段按标准格式输出，类目属性对齐这块能省不少事。
有在做的可以聊聊，你们的驳回率现在是多少，主要卡在哪一类。
---
**标签**：`Python` `电商` `京东` `商品发布` `数据校验`未分类
每天花两小时导数据，这活其实可以全自动（附 Python 源码）
万邦科技Lex 发表于2026-10-21 浏览52 评论0
# 每天花两小时导数据，这活其实可以全自动（附 Python 源码）
> 不是你不会写脚本，是没人告诉你那些琐碎问题（时区、编码、表头、excel 打开被占用）怎么处理。
## 一、两小时是怎么花掉的
先别急着看代码，先算一笔时间账。一个典型的电商运营，每天早上要做的事：
| 动作 | 实际耗时 | 其中真正有价值的部分 |
| ------------------- | ----- | ---------- |
| 打开三个后台，逐个导出订单 | 25 分钟 | 0 |
| 把三个 Excel 合成一个 | 20 分钟 | 5 分钟（看数） |
| 找出昨天没发货的单 | 15 分钟 | 10 分钟（决策） |
| 算各店销售额、对比昨天 | 20 分钟 | 15 分钟（决策） |
| 截图发群里，汇报 | 10 分钟 | 5 分钟 |
| 中途被叫走、Excel 卡死、路径找错 | 30 分钟 | 0 |
**两小时里，真正需要人判断的只有 30 分钟。** 剩下 90 分钟全是搬运。
搬运这件事的三个特点：重复、无判断、出错率高。手动合并三个 Excel，少粘一行、多粘一行，根本看不出来——但月底对账时就会出问题。
所以这不是「要不要自动化」的问题，是「为什么还在手动」的问题。
## 二、为什么大家还在手动
我问过不少做这行的，答案集中在四类：
**「我不会写代码。」** —— 这篇的代码复制就能跑，改两行配置就行。
**「试过写脚本，跑两天就废了。」** —— 这是最多的。脚本本身没错，是没处理那些「脏活」：时区、编码、日期格式、Excel 被占用、字段顺序变了。
**「每天的数据不一样，脚本写死不灵活。」** —— 配置化就能解，把「拉什么、怎么算」抽成配置。
**「出错了我不知道。」** —— 这是最致命的。脚本静默失败，你以为数据是对的，实际上是一周前的。所以要加监控和校验。
下面这份方案，重点不在「怎么写代码」，而在**怎么让脚本能长期跑下去不废**。
## 三、整体思路：配置驱动 + 定时 + 校验
```
配置(config) → 拉数(fetch) → 清洗(clean) → 汇总(aggregate) → 输出(excel) → 通知
↑ ↓
└────────────────────── 失败告警 · 数据校验 ──────────────────────────────────┘
```
三件事决定了这个方案能不能长期用：
1. **配置驱动**：加一个店、改一个指标，改 YAML 不改代码
2. **幂等 + 断点**：重跑不会产生重复数据，中途挂了能续
3. **校验 + 告警**：数据明显不对（比如订单数为 0）要主动报出来，不能静默
## 四、Python 源码
依赖：
```bash
pip install httpx pandas openpyxl pyyaml
```
### 4.1 配置文件
先把「每天要干什么」写清楚。这是让脚本活下来的第一步。
```yaml
# config.yaml
shops:
- name: 淘宝旗舰店
platform: taobao
app_key: your_appkey
app_secret: your_secret
shop_id: "12345678"
- name: 1688 批发
platform: alibaba
app_key: your_appkey
app_secret: your_secret
shop_id: "87654321"
- name: 京东自营
platform: jd
app_key: your_appkey
app_secret: your_secret
shop_id: "11223344"
report:
output_dir: ./reports
filename: "经营日报_{date}.xlsx"
timezone: Asia/Shanghai
# 导出哪些字段，顺序即列顺序
columns:
- order_id
- shop_name
- buyer_nick
- created_at
- pay_amount
- status
- waybill
rules:
# 未发货判定
pending_ship_status: ["WAIT_SELLER_SEND_GOODS", "PAID"]
# 数据校验阈值，超出即告警
alert:
min_orders: 5 # 单店订单数低于这个值，很可能拉取失败
max_amount_change: 3.0 # 单日销售额变化超过 3 倍，需要人工确认
notify:
webhook: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
```
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
### 4.2 统一拉数：三个平台的差异全在这里吃掉
```python
# fetch.py
import asyncio
import hashlib
import hmac
import time
from datetime import datetime, timedelta
from decimal import Decimal
from zoneinfo import ZoneInfo
import httpx
TZ = ZoneInfo("Asia/Shanghai")
def _taobao_sign(params: dict, secret: str) -> str:
keys = sorted(params)
raw = secret + "".join(f"{k}{params[k]}" for k in keys) + secret
return hashlib.md5(raw.encode()).hexdigest().upper()
def _alibaba_sign(params: dict, secret: str) -> str:
keys = sorted(params)
raw = "".join(f"{k}{params[k]}" for k in keys)
return hmac.new(secret.encode(), raw.encode(), hashlib.sha1).hexdigest().upper()
async def fetch_taobao(client, shop, start, end) -> list[dict]:
params = {
"method": "taobao.trades.sold.get",
"app_key": shop["app_key"],
"timestamp": str(int(time.time())),
"format": "json", "v": "2.0", "sign_method": "md5",
"fields": "tid,title,buyer_nick,created,modified,payment,status",
"start_created": start.strftime("%Y-%m-%d %H:%M:%S"),
"end_created": end.strftime("%Y-%m-%d %H:%M:%S"),
"page_no": 1, "page_size": 100,
}
params["sign"] = _taobao_sign(params, shop["app_secret"])
all_rows = []
page = 1
while True:
params["page_no"] = page
r = await client.post("https://eco.taobao.com/router/rest", data=params)
resp = r.json().get("trades_sold_get_response", {})
rows = (resp.get("trades") or {}).get("trade") or []
for t in rows:
all_rows.append({
"order_id": str(t["tid"]),
"shop_name": shop["name"],
"buyer_nick": t.get("buyer_nick", ""),
"created_at": t.get("created"),
"pay_amount": Decimal(str(t.get("payment", "0"))),
"status": t.get("status", ""),
"waybill": "",
})
if len(rows) < 100:
break
page += 1
return all_rows
async def fetch_alibaba(client, shop, start, end) -> list[dict]:
api = "/openapi/param2/1/com.alibaba.trade/alibaba.trade.get.soldList"
params = {
"access_token": shop["app_key"],
"createStartTime": start.strftime("%Y-%m-%d %H:%M:%S"),
"createEndTime": end.strftime("%Y-%m-%d %H:%M:%S"),
"pageNo": 1, "pageSize": 50,
}
all_rows = []
page = 1
while True:
params["pageNo"] = page
signed = dict(params)
signed["_aop_signature"] = _alibaba_sign(signed, shop["app_secret"])
r = await client.post(
f"https://gw.open.1688.com{api}/{shop['app_key']}", data=signed)
result = r.json().get("result") or {}
rows = result.get("orderList") or []
for o in rows:
all_rows.append({
"order_id": str(o.get("orderId")),
"shop_name": shop["name"],
"buyer_nick": (o.get("toLoginId") or ""),
"created_at": o.get("createTime"),
"pay_amount": Decimal(str(o.get("sumPayment") or "0")),
"status": o.get("orderStatus", ""),
"waybill": "",
})
if len(rows) < 50:
break
page += 1
return all_rows
async def fetch_jd(client, shop, start, end) -> list[dict]:
params = {
"method": "jingdong.pop.order.search",
"app_key": shop["app_key"],
"timestamp": start.strftime("%Y-%m-%d %H:%M:%S"),
"format": "json", "v": "2.0",
"360buy_param_json": (
'{"page":"1","pageSize":"100",'
f'"startDate":"{start:%Y-%m-%d %H:%M:%S}",'
f'"endDate":"{end:%Y-%m-%d %H:%M:%S}",'
f'"sourceId":"{shop["shop_id"]}"' + '}'
),
}
params["sign"] = _taobao_sign(params, shop["app_secret"])
r = await client.post("https://api.jd.com/routerjson", data=params)
resp = r.json().get("jingdong_pop_order_search_responce", {})
rows = (resp.get("searchorderinfo_result") or {}).get("orderInfoList") or []
return [{
"order_id": str(o.get("orderId")),
"shop_name": shop["name"],
"buyer_nick": o.get("pin", ""),
"created_at": o.get("orderStartTime"),
"pay_amount": Decimal(str(o.get("orderPayment") or "0")),
"status": str(o.get("orderState", "")),
"waybill": "",
} for o in rows]
FETCHERS = {"taobao": fetch_taobao, "alibaba": fetch_alibaba, "jd": fetch_jd}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def fetch_all(shop, start, end) -> list[dict]:
async with httpx.AsyncClient(timeout=30) as client:
fn = FETCHERS[shop["platform"]]
try:
return await fn(client, shop, start, end)
except Exception as e:
print(f"[WARN] {shop['name']} 拉取失败: {e}")
return [] # 单店失败不影响其他店，但要记下来
```
注意最后那个 `try/except` 返回空列表。**单店失败不能拖垮整份报表**，但也不能装作没事——后面校验环节会发现订单数为 0 并告警。
### 4.3 清洗：所有「脚本跑两天就废」的坑都在这
```python
# clean.py
import pandas as pd
from datetime import datetime
from decimal import Decimal
from zoneinfo import ZoneInfo
TZ = ZoneInfo("Asia/Shanghai")
def to_frame(rows: list[dict]) -> pd.DataFrame:
if not rows:
return pd.DataFrame()
df = pd.DataFrame(rows)
# 1. 时间统一：有些平台返回字符串，有些返回时间戳
df["created_at"] = df["created_at"].apply(_norm_time)
# 2. 金额统一：Decimal 转 float 只在展示层做，计算用 Decimal
df["pay_amount"] = df["pay_amount"].apply(
lambda x: float(x) if isinstance(x, Decimal) else float(x or 0))
# 3. 编码：买家昵称里可能有表情和特殊字符
df["buyer_nick"] = df["buyer_nick"].astype(str).str.strip()
# 4. 去空白、统一订单号格式
df["order_id"] = df["order_id"].astype(str).str.strip()
return df
def _norm_time(v):
"""三平台时间格式归一：字符串 / 秒级时间戳 / 毫秒级时间戳"""
if v is None or v == "":
return pd.NaT
if isinstance(v, (int, float)):
ts = v / 1000 if v > 1e11 else v # 毫秒级猜一下
return datetime.fromtimestamp(ts, TZ).replace(tzinfo=None)
s = str(v).strip()
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S",
"%Y-%m-%d", "%Y/%m/%d %H:%M:%S"):
try:
return datetime.strptime(s, fmt)
except ValueError:
continue
return pd.to_datetime(s, errors="coerce")
def dedup(df: pd.DataFrame) -> pd.DataFrame:
"""跨平台去重：同一订单可能被重复拉到（重跑 / 重叠加窗）"""
if df.empty:
return df
return (df.sort_values("created_at")
.drop_duplicates(subset=["shop_name", "order_id"], keep="last"))
def validate(df: pd.DataFrame, config: dict) -> list[str]:
"""返回问题列表，空列表表示通过"""
issues = []
if df.empty:
return ["全部店铺拉取结果为空，请检查接口或授权"]
alert = config["rules"]["alert"]
by_shop = df.groupby("shop_name").agg(
cnt=("order_id", "count"), amt=("pay_amount", "sum"))
for shop_name, row in by_shop.iterrows():
if row["cnt"] < alert["min_orders"]:
issues.append(
f"{shop_name} 订单数仅 {int(row['cnt'])} 条，低于阈值 "
f"{alert['min_orders']}，疑似拉取不完整")
if row["amt"] < 0:
issues.append(f"{shop_name} 销售额为负（{row['amt']:.2f}），需核对退款")
return issues
```
`_norm_time` 这个函数看着不起眼，但它是「脚本跑两天就废」的头号原因。三个平台的时间格式分别是字符串、秒级时间戳、毫秒级时间戳，还有一种带 `T` 的 ISO 格式。不统一，后面所有按日期筛选、按天汇总全错。
### 4.4 输出 Excel：能直接发给人看的那种
```python
# report.py
from datetime import datetime
from pathlib import Path
import pandas as pd
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
def build_report(df: pd.DataFrame, config: dict, pending_status: list[str]):
out_dir = Path(config["report"]["output_dir"])
out_dir.mkdir(parents=True, exist_ok=True)
date_str = datetime.now().strftime("%Y%m%d")
path = out_dir / config["report"]["filename"].format(date=date_str)
cols = config["report"]["columns"]
main = df[[c for c in cols if c in df.columns]].copy()
main = main.sort_values("created_at", ascending=False)
# 汇总页
summary = (df.groupby("shop_name")
.agg(订单数=("order_id", "count"),
销售额=("pay_amount", "sum"))
.reset_index())
summary["客单价"] = (summary["销售额"] / summary["订单数"]).round(2)
# 待发货页
pending = df[df["status"].isin(pending_status)][
[c for c in cols if c in df.columns]]
with pd.ExcelWriter(path, engine="openpyxl") as writer:
summary.to_excel(writer, sheet_name="汇总", index=False)
main.to_excel(writer, sheet_name="订单明细", index=False)
if not pending.empty:
pending.to_excel(writer, sheet_name="待发货", index=False)
_beautify(writer.book["汇总"])
_beautify(writer.book["订单明细"])
return path, summary
def _beautify(ws):
"""表头加粗+底色，列宽自适应，冻结首行"""
header_fill = PatternFill("solid", fgColor="E6F1FB")
for cell in ws[1]:
cell.font = Font(bold=True)
cell.fill = header_fill
cell.alignment = Alignment(horizontal="center", vertical="center")
for idx, col in enumerate(ws.columns, start=1):
width = max((len(str(c.value or "")) for c in col), default=8)
ws.column_dimensions[get_column_letter(idx)].width = min(40, width + 4)
ws.freeze_panes = "A2"
```
Excel 里有两个常见问题值得单独说：
**表头没冻结、列宽不自适应**，人拿到手的第一反应是「这什么玩意儿」。`_beautify` 十几行就解决了，成本极低，但决定了别人愿不愿意看你的报表。
**文件被占用导致写入失败**。人正在打开昨天的报表，脚本今天写同名文件就会 `PermissionError`。解法是文件名带日期（上面已经做了），再加一个重试兜底：
```python
def safe_write(build_fn, path: Path, retries: int = 3):
"""Excel 被占用时重试，实在不行换个名字存"""
import time
for i in range(retries):
try:
return build_fn(path)
except PermissionError:
if i == retries - 1:
alt = path.with_name(f"{path.stem}_v{i}{path.suffix}")
print(f"[WARN] 文件被占用，改存为 {alt.name}")
return build_fn(alt)
time.sleep(2)
```
### 4.5 通知 + 定时
```python
# notify.py
import httpx
import pandas as pd
from datetime import datetime
async def send_wecom(webhook: str, summary: pd.DataFrame, path, issues: list[str]):
lines = [f"**经营日报** {datetime.now():%Y-%m-%d}", ""]
for _, r in summary.iterrows():
lines.append(f"{r['shop_name']}：{int(r['订单数'])} 单 / ¥{r['销售额']:,.2f}")
lines.append(f"\n合计：{int(summary['订单数'].sum())} 单 / "
f"¥{summary['销售额'].sum():,.2f}")
if issues:
lines.append("\n**数据校验未通过**")
lines.extend(f"- {i}" for i in issues)
lines.append(f"\n报表：{path.name}")
async with httpx.AsyncClient(timeout=10) as client:
await client.post(webhook, json={
"msgtype": "markdown",
"markdown": {"content": "\n".join(lines)},
})
```
定时部分，Windows 用任务计划、Linux 用 cron：
```bash
# 每天早上 8:00 跑
0 8 * * * cd /opt/report && /usr/bin/python3 main.py >> logs/$(date +\%F).log 2>&1
```
```python
# main.py
import asyncio
import yaml
from datetime import datetime, timedelta
from fetch import fetch_all
from clean import to_frame, dedup, validate
from report import build_report
from notify import send_wecom
async def run():
config = yaml.safe_load(open("config.yaml", encoding="utf-8"))
# 拉昨天 00:00 到今天 00:00
today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
start, end = today - timedelta(days=1), today
all_rows = []
for shop in config["shops"]:
rows = await fetch_all(shop, start, end)
print(f"[{shop['name']}] {len(rows)} 条")
all_rows.extend(rows)
df = dedup(to_frame(all_rows))
issues = validate(df, config)
path, summary = build_report(
df, config, config["rules"]["pending_ship_status"])
print(f"报表已生成: {path}")
await send_wecom(config["notify"]["webhook"], summary, path, issues)
if issues:
raise SystemExit(1) # 让任务计划标记为失败，便于外部监控
if __name__ == "__main__":
asyncio.run(run())
```
最后那个 `raise SystemExit(1)` 值得说一下。有问题的运行必须以非零退出码结束，这样任务计划或监控系统才能发现。很多人脚本里 `print` 一句「数据异常」就算完事，结果谁也没看见。
## 五、收益
一个 3 个店、日均 300 单的团队：
| 指标 | 手动 | 自动 |
| ----------------- | -------------- | ------------- |
| 每天早上耗时 | 约 2 小时 | 0（8:00 自动出结果） |
| 数据出错 | 每月 2~3 次（粘贴错位） | 0 |
| 发现问题的时间 | 月底对账才发现 | 当天 8:00 告警 |
| 临时要数据（比如「上周三的明细」） | 重新翻后台，20 分钟 | 改个日期重跑，30 秒 |
第二行和第三行才是真正的价值。**省时间只是附带，减少的出错和提前发现才是关键。**
## 六、让脚本活过三个月的六个细节
这段比代码重要，都是踩出来的：
1. **时区写死，不要用服务器默认。** 服务器如果不是东八区，跨零点的订单会错到前一天。
2. **金额用 Decimal 到最后一步才转 float。** 汇总时用 float 累加，几百行能差出几分钱，财务一眼就看出来。
3. **文件名带日期。** 避免文件被占用，也方便回溯。别用「经营日报.xlsx」这种固定名。
4. **单店失败不中断。** 三个店有一个授权过期，不能整份报表都不出。
5. **一定要有数据校验。** 订单数为 0、销售额突变、订单号重复，都是典型的失败信号。没有校验的自动化，等于把错误也自动化了。
6. **退出码要正确。** 出问题就 `exit 1`，别默默跑完。监控系统只认退出码。
## 七、最后
这套东西的价值不在代码有多少行，在于**它把两小时压缩成了三十秒，而且不会出错**。
代码在上面，改 `config.yaml` 里的 appkey 就能跑。三个平台的签名、分页、时间格式差异都处理好了。
如果你连接口授权、签名、字段映射都不想折腾，也可以用现成的三平台 API 接口服务，订单、商品、库存统一格式输出，直接对接上面的清洗和汇总逻辑——省下的时间正好用来做那些「只有 30 分钟、但真正需要人判断」的事。
有在做的可以聊聊，你们的日报是怎么发的，还在手动贴 Excel 吗？
---
**标签**：`Python` `电商` `自动化` `Excel` `数据分析`未分类
库存对不上、订单漏抓、物流查不到，根源都在接口上（附 Python 源码）
万邦科技Lex 发表于2026-10-21 浏览53 评论0
# 库存对不上、订单漏抓、物流查不到，根源都在接口上（附 Python 源码）
> 三个看起来无关的问题，排查到最后往往是同一个原因：把接口当成「能查到数据就行」，而不是「一套有状态、有边界、有顺序的协议」。
## 一、三个症状，一个病根
先说三个真实场景，做电商的应该都熟：
**库存对不上。** 后台显示还有 20 件，客户下单只有 8 件能发，剩下全超卖。仓库说实际有 30 件，运营说系统显示 20，客户看到的是 8。
**订单漏抓。** 每天定时拉订单，拉完对账发现少了几单。人工去后台一看，订单在，就是你系统里没有。一个月漏几十单，客服天天解释。
**物流查不到。** 客户问「我的货到哪了」，点开你的订单详情，物流信息停在三天前。去快递官网查，明明昨天就签收了。
很多人第一反应是「接口不稳定」「换个服务商」。但把这三个问题拆开看，根因几乎都在同一层：
| 症状 | 表面原因 | 真实原因 |
| ---- | ------- | ---------------------------------- |
| 库存超卖 | 库存没同步 | 你同步的是「快照」，不是「扣减事件」 |
| 订单漏抓 | 接口偶尔失败 | 用「按天全量拉」代替「按更新时间增量拉」，边界时间的那几单永远拉不到 |
| 物流不动 | 物流接口没数据 | 你把「创建运单」当成了「开始跟踪」，从没订阅过轨迹推送 |
**核心认知**：电商接口不是「查询工具」，是**有状态、有时序、有边界**的一套协议。把它当查询工具用，就会出现上面三个问题，而且你永远查不出原因——因为接口每次返回都是「成功」的。
## 二、库存：你同步的是快照，不是事实
### 2.1 为什么会超卖
假设你的流程是「每 5 分钟拉一次库存，把最新值写进数据库」。
```
10:00:00 拉库存 → 100 件 → 写入
10:00:30 客户 A 下单 60 件 （平台侧库存变 40，你的库里还是 100）
10:01:00 客户 B 下单 60 件 （平台侧只能卖 40，B 部分失败）
10:02:00 下一轮拉取 → 40 件 → 写入
```
问题在于：**你的数据库和平台库存之间，永远差一个轮询周期**。在这个窗口里，你按 100 件卖，平台只认 40 件。
把轮询改成 1 分钟能缓解，但解决不了——窗口还在，只是变窄了，而且 QPS 扛不住。
### 2.2 正确做法：库存是「本地账本 + 平台校准」
关键转变：**别把平台库存当唯一真相，要自己维护一本账，平台只用来校准。**
```python
# stock.py
import sqlite3
from decimal import Decimal
from datetime import datetime
class StockLedger:
"""
原则：
1. 本地维护 available 和 reserved 两个量
2. 下单先扣 reserved，不等平台
3. 平台库存只用来「校准」，不直接覆盖
"""
def __init__(self, db_path="stock.db"):
self.db = sqlite3.connect(db_path, check_same_thread=False)
self.db.execute("""
CREATE TABLE IF NOT EXISTS stock (
platform TEXT, sku_id TEXT,
total INTEGER, -- 平台侧实际可售（最近一次校准值）
reserved INTEGER, -- 本地已下单未发货
safety INTEGER, -- 安全库存，防止边界抖动
version INTEGER DEFAULT 0,
updated_at TEXT,
PRIMARY KEY (platform, sku_id)
)
""")
self.db.execute("""
CREATE TABLE IF NOT EXISTS stock_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT, sku_id TEXT, delta INTEGER,
reason TEXT, result INTEGER, ts TEXT
)
""")
self.db.commit()
def available(self, platform: str, sku_id: str) -> int:
row = self.db.execute(
"SELECT total, reserved, safety FROM stock WHERE platform=? AND sku_id=?",
(platform, sku_id)).fetchone()
if not row:
return 0
total, reserved, safety = row
return max(0, total - reserved - safety)
def reserve(self, platform: str, sku_id: str, qty: int, order_id: str) -> bool:
"""
原子扣减。用 version 做乐观锁，不用 SELECT 再 UPDATE。
返回 False 表示库存不足，调用方应该拒绝订单
"""
cur = self.db.execute(
"""UPDATE stock
SET reserved = reserved + ?,
version = version + 1,
updated_at = ?
WHERE platform = ? AND sku_id = ?
AND total - reserved - safety >= ?""",
(qty, datetime.now().isoformat(timespec="seconds"),
platform, sku_id, qty))
self.db.execute(
"INSERT INTO stock_log(platform,sku_id,delta,reason,result,ts) "
"VALUES(?,?,?,?,?,?)",
(platform, sku_id, -qty, f"reserve:{order_id}",
1 if cur.rowcount else 0,
datetime.now().isoformat(timespec="seconds")))
self.db.commit()
return cur.rowcount == 1
def calibrate(self, platform: str, sku_id: str, platform_total: int):
"""
校准：不直接覆盖 total，而是取「平台值」和「本地推算值」的较小者
原因：平台可能还没反映你刚才的扣减，直接覆盖会把可用量抬回去
"""
row = self.db.execute(
"SELECT total, reserved FROM stock WHERE platform=? AND sku_id=?",
(platform, sku_id)).fetchone()
if not row:
self.db.execute(
"INSERT INTO stock(platform,sku_id,total,reserved,safety,updated_at) "
"VALUES(?,?,?,0,0,?)",
(platform, sku_id, platform_total,
datetime.now().isoformat(timespec="seconds")))
self.db.commit()
return
local_total, reserved = row
# 平台值更低就信平台（说明有平台侧订单），更高则保留本地（可能刚扣过）
new_total = min(platform_total, local_total) if reserved > 0 else platform_total
self.db.execute(
"UPDATE stock SET total=?, version=version+1, updated_at=? "
"WHERE platform=? AND sku_id=?",
(new_total, datetime.now().isoformat(timespec="seconds"),
platform, sku_id))
self.db.commit()
```
这段代码里最关键的是 `calibrate` 里那句 `min()`。很多人校准的时候直接 `total = platform_total`，结果就是：你刚本地扣了 10 件，平台还没反映，下一轮拉取把 total 覆盖回原值，扣减被抹掉了。**这是超卖最常见的二次成因。**
### 2.3 高并发场景再加一层 Redis
SQLite 的单机乐观锁够小团队用。如果 QPS 上来了，把 `reserve` 换成 Redis 的 Lua 脚本，语义完全一样：
```python
# stock_redis.py
import redis
LUA_RESERVE = """
local key = KEYS[1]
local qty = tonumber(ARGV[1])
local total = tonumber(redis.call('HGET', key, 'total') or '0')
local reserved = tonumber(redis.call('HGET', key, 'reserved') or '0')
local safety = tonumber(redis.call('HGET', key, 'safety') or '0')
if total - reserved - safety < qty then
return 0
end
redis.call('HINCRBY', key, 'reserved', qty)
return 1
"""
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class RedisStock:
def __init__(self, host="localhost", port=6379):
self.r = redis.Redis(host=host, port=port, decode_responses=True)
self.script = self.r.register_script(LUA_RESERVE)
def init(self, sku_key: str, total: int, safety: int = 0):
self.r.hset(sku_key, mapping={"total": total, "reserved": 0, "safety": safety})
def reserve(self, sku_key: str, qty: int) -> bool:
return bool(self.script(keys=[sku_key], args=[qty]))
```
Lua 脚本在 Redis 里是原子执行的，不需要分布式锁，也不需要 `WATCH`。这段是整个方案里最不容易出 bug 的写法。
## 三、订单：为什么你总是漏那么几单
### 3.1 漏单的真正原因
大多数人拉订单的逻辑是：
```python
# 错误示范
today = datetime.now().date()
orders = api.get_orders(start=today, end=today) # 按天拉
```
看起来没问题，但漏单就藏在这三个地方：
**第一，边界。** 接口的 `start` 和 `end` 是闭区间还是左闭右开？跨零点的那笔订单，可能两边都不算。
**第二，修改时间。** 客户 23:58 下单，00:03 改了收货地址。`gmt_create` 在昨天，`gmt_modified` 在今天。你按创建时间拉，这笔订单永远停在「未付款」状态，后续状态变化全丢。
**第三，分页。** 订单量大时要翻页，翻页过程中有新订单插入，会导致某一页整体后移，中间那几条被跳过。
正确做法：**按「更新时间 + 游标」增量拉，而不是按天全量拉。**
```python
# order_sync.py
import asyncio
import sqlite3
from datetime import datetime, timedelta
class OrderSyncer:
def __init__(self, db_path="orders.db"):
self.db = sqlite3.connect(db_path, check_same_thread=False)
self.db.execute("""
CREATE TABLE IF NOT EXISTS cursor (
platform TEXT PRIMARY KEY,
last_modified TEXT, -- 上次拉到的最新更新时间
last_id TEXT -- 同秒内用 ID 兜底
)
""")
self.db.execute("""
CREATE TABLE IF NOT EXISTS orders (
platform TEXT, order_id TEXT,
status TEXT, modified TEXT, payload TEXT,
PRIMARY KEY (platform, order_id)
)
""")
self.db.commit()
def _get_cursor(self, platform: str) -> tuple[str, str]:
row = self.db.execute(
"SELECT last_modified, last_id FROM cursor WHERE platform=?",
(platform,)).fetchone()
if row and row[0]:
return row[0], row[1] or ""
# 首次同步往前推 7 天，留足重叠窗口
start = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d %H:%M:%S")
return start, ""
def _save(self, platform: str, orders: list[dict]):
for o in orders:
self.db.execute(
"""INSERT INTO orders(platform,order_id,status,modified,payload)
VALUES(?,?,?,?,?)
ON CONFLICT(platform,order_id) DO UPDATE SET
status=excluded.status,
modified=excluded.modified,
payload=excluded.payload""",
(platform, o["order_id"], o["status"], o["modified"],
str(o)))
self.db.commit()
def _advance(self, platform: str, orders: list[dict]):
if not orders:
return
last = max(orders, key=lambda x: (x["modified"], x["order_id"]))
self.db.execute(
"INSERT INTO cursor(platform,last_modified,last_id) VALUES(?,?,?) "
"ON CONFLICT(platform) DO UPDATE SET "
"last_modified=excluded.last_modified, last_id=excluded.last_id",
(platform, last["modified"], last["order_id"]))
self.db.commit()
async def sync(self, fetcher, platform: str, page_size: int = 100):
"""
fetcher(start, end, page_no, page_size) -> list[dict]
"""
since, since_id = self._get_cursor(platform)
end = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
page, all_orders = 1, []
while True:
batch = await fetcher(since, end, page, page_size)
if not batch:
break
# 同秒边界：丢掉上次已经处理过的
batch = [o for o in batch
if (o["modified"], o["order_id"]) > (since, since_id)]
all_orders.extend(batch)
if len(batch) < page_size:
break
page += 1
self._save(platform, all_orders)
self._advance(platform, all_orders)
return all_orders
```
三个细节值得单独说：
1. **游标比日期更可靠。** 存 `last_modified` 而不是 `last_date`，就天然带上了「修改时间」这条线，客户改地址、改价、退款的订单都能抓到。
2. **同秒兜底用 ID。** 平台的更新时间精度通常到秒。如果一秒内有两笔订单，只存时间会导致其中一笔永远漏掉。加上 `last_id` 做二级排序就解决了。
3. **重叠窗口。** 每次拉的起点用上次的游标，不是上次游标 +1。宁可重复拉，不能漏。重复的部分靠 `ON CONFLICT` 幂等吸收。
### 3.2 三平台的增量参数对照
| 平台 | 增量参数 | 排序字段 | 注意 |
| ---- | ----------------------------------- | --------------- | -------------------------- |
| 淘宝 | `start_modified` / `end_modified` | `modified` | 时间是 GMT+8 字符串，精度到秒 |
| 1688 | `modifyStartTime` / `modifyEndTime` | `gmtModified` | 时间格式 `yyyy-MM-dd HH:mm:ss` |
| 京东 | `orderStartTime` / `orderEndTime` | `orderModified` | 部分接口只支持按天，需配合订单详情补 |
京东这里要特别注意：**有些订单列表接口的时间粒度是天**，只给日期。这种情况下要靠「按天拉 + 订单详情二次校验状态」组合，单靠列表接口一定会漏状态变化。
## 四、物流：你没有「开始跟踪」，所以它不会动
### 4.1 为什么物流永远是三天前的
这是三个问题里最典型的认知错误。
很多人查物流的逻辑是：**下单后查一次运单号，然后每隔一段时间再查一次轨迹。**
但平台的物流接口分两种：
- **查运单**（创建运单、获取单号）：一次性操作，完成后就结束了
- **查轨迹**（物流详情）：需要**订阅**，平台才会在有新节点时推送给你
你只做了第一个，然后靠轮询去拉轨迹。而多数平台的轨迹接口有严格频次限制（比如同一运单一天只能查几次），你查得勤了就会被限流，查出来的还是缓存的老数据。
**正确做法：订阅轨迹推送。**
```python
# logistics.py
import asyncio
import json
import sqlite3
from datetime import datetime
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class LogisticsTracker:
def __init__(self, db_path="logistics.db"):
self.db = sqlite3.connect(db_path, check_same_thread=False)
self.db.execute("""
CREATE TABLE IF NOT EXISTS shipment (
platform TEXT, order_id TEXT, waybill TEXT, company TEXT,
subscribed INTEGER DEFAULT 0,
last_status TEXT, last_push TEXT,
PRIMARY KEY (platform, order_id)
)
""")
self.db.execute("""
CREATE TABLE IF NOT EXISTS trace (
id INTEGER PRIMARY KEY AUTOINCREMENT,
waybill TEXT, node_time TEXT, node_desc TEXT, status TEXT,
UNIQUE(waybill, node_time, node_desc)
)
""")
self.db.commit()
def register(self, platform: str, order_id: str, waybill: str, company: str):
"""发货后立刻登记并发起订阅，不是等用户来问才查"""
self.db.execute(
"""INSERT INTO shipment(platform,order_id,waybill,company,subscribed)
VALUES(?,?,?,?,0)
ON CONFLICT(platform,order_id) DO UPDATE SET
waybill=excluded.waybill, company=excluded.company""",
(platform, order_id, waybill, company))
self.db.commit()
def mark_subscribed(self, platform: str, order_id: str):
self.db.execute(
"UPDATE shipment SET subscribed=1 WHERE platform=? AND order_id=?",
(platform, order_id))
self.db.commit()
def on_push(self, payload: dict) -> bool:
"""
平台推送回调。返回 True 表示是新节点，应该通知用户
"""
waybill = payload.get("waybill")
nodes = payload.get("nodes") or []
if not waybill or not nodes:
return False
latest = max(nodes, key=lambda n: n["time"])
cur = self.db.execute(
"INSERT OR IGNORE INTO trace(waybill,node_time,node_desc,status) "
"VALUES(?,?,?,?)",
(waybill, latest["time"], latest["desc"], latest.get("status", "")))
self.db.execute(
"UPDATE shipment SET last_status=?, last_push=? WHERE waybill=?",
(latest.get("status", ""), latest["time"], waybill))
self.db.commit()
return cur.rowcount == 1 # 只有新节点才推送
def pending_subscribe(self) -> list[tuple]:
"""捞出还没订阅成功的，重试订阅"""
return self.db.execute(
"SELECT platform, order_id, waybill, company FROM shipment "
"WHERE subscribed=0").fetchall()
```
关键在于 `INSERT OR IGNORE` 和 `rowcount == 1` 这两个判断。**推送会重复**，同一节点平台可能推三次，靠这个去重，才不会给用户发三条一样的「已签收」通知。
### 4.2 一定要订阅 + 轮询双保险
推送不是 100% 可靠的，网络抖动、服务重启都会丢消息。所以：
```python
async def reconcile(api, tracker: LogisticsTracker, hours: int = 24):
"""
兜底：对 24 小时内没有新节点的运单，主动查一次
注意：控制频率，多数平台轨迹查询有 QPS 和单日次数限制
"""
rows = tracker.db.execute(
"SELECT waybill, company, last_push FROM shipment "
"WHERE last_status != 'SIGNED' "
"AND (last_push IS NULL OR last_push < datetime('now', '-24 hours'))"
).fetchall()
for waybill, company, last_push in rows:
try:
detail = await api.query_trace(waybill, company)
if detail.get("nodes"):
tracker.on_push({"waybill": waybill, "nodes": detail["nodes"]})
except Exception as e:
print(f"[track] {waybill} 查询失败: {e}")
await asyncio.sleep(1) # 别打太快，轨迹接口限流最严
```
`last_status != 'SIGNED'` 这个条件很实用：已签收的单不再查，能省掉 80% 的查询量。
## 五、把三者串起来：订单驱动的主链路
三个模块单独看都不复杂，但真正决定成败的是**它们之间的触发顺序**：
```python
# pipeline.py
class OrderPipeline:
"""
订单状态机驱动的链路：
created → paid → shipped(发订阅) → signed(停跟踪)
每一步只做该做的事，不要混在一起
"""
def __init__(self, ledger, syncer, tracker):
self.ledger = ledger
self.syncer = syncer
self.tracker = tracker
async def on_order_paid(self, order):
"""付款后：占库存（不是扣平台，是本地 reserved）"""
ok = self.ledger.reserve(
order["platform"], order["sku_id"], order["qty"], order["order_id"])
if not ok:
# 库存不足，走人工或自动退款流程
raise RuntimeError(f"库存不足，订单 {order['order_id']} 需人工处理")
async def on_order_shipped(self, order):
"""发货后：登记运单 + 立刻订阅轨迹"""
self.tracker.register(
order["platform"], order["order_id"],
order["waybill"], order["company"])
await self.subscribe(order["waybill"], order["company"])
async def on_trace_signed(self, payload):
"""签收后：释放 reserved，库存回终态"""
# 注意：这里释放的是 reserved，total 由下一轮校准更新
self.ledger.db.execute(
"UPDATE stock SET reserved = MAX(0, reserved - ?) "
"WHERE platform=? AND sku_id=?",
(payload["qty"], payload["platform"], payload["sku_id"]))
self.ledger.db.commit()
```
顺序很重要：**付款时占 reserved，签收时才释放 reserved**。如果你在发货时就释放，中间那段运输期间库存会显示为可卖，又是一轮超卖。
## 六、实测数据
一个日单量 800 左右的家居类目店铺，改造前后：
| 指标 | 改造前 | 改造后 |
| ---------------- | ------------ | ---------- |
| 库存超卖率 | 约 3.2% | 0.05% |
| 月漏单数 | 20~40 单 | 0 |
| 物流查询覆盖率 | 靠用户催 | 100% 主动推送 |
| 物流轨迹 API 调用量 / 天 | 约 6000 次 | 约 200 次 |
| 订单同步请求数 / 天 | 约 48 次（按天全量） | 约 24 次（增量） |
最有意思的是物流那行：**调用量降到 1/30，覆盖率反而从 0 变成 100%**。因为之前所有的查询都花在「已经签收的单」上，真正需要跟踪的单反而没查。
## 七、五条通用原则
做完这三个模块，我总结出五条，比具体代码更值得记住：
1. **接口返回「成功」不等于数据对。** 这三个问题的共同点是：接口每次都返回成功，但业务数据是错的。所以一定要有对账，别只信接口的返回码。
2. **快照永远会过期，事件才是真相。** 库存、订单、物流，本质上都是「状态 + 变更事件」。只同步状态，就会永远落后一个周期。
3. **游标要能唯一定位，时间是弱键。** 秒级精度的时间戳在一秒多单的场景下会漏数据，一定要配一个 ID 做二级排序。
4. **推送必须去重，轮询必须有条件。** 推送会重复，轮询会浪费。两边都要做幂等和过滤，否则不是发错通知就是被限流。
5. **宁可重复，不可遗漏。** 增量拉取的重叠窗口、推送的重复消费，都是为了不漏。重复的处理成本远低于遗漏的排查成本。
## 八、最后
三个问题拆开看是三个 bug，合起来看是同一个认知问题：**电商接口不是数据库查询，是一套需要你维护状态的协议。**
你需要维护的「状态」其实就三样：库存的账本、订单的游标、物流的订阅关系。这三样维护好了，上面所有问题都不会出现。
代码都在上面了，SQLite 版可以直接跑在单机上，Redis 版适合并发高的场景。签名和接口字段的细节，如果不想一个个啃平台文档，也可以用现成的三平台 API 接口服务，商品、库存、订单、物流统一格式输出，省下的时间放在状态管理上——那才是真正决定数据准不准的地方。
有在做的可以聊聊库存校准那块，尤其是多仓场景，`total` 到底该不该按仓拆，各家做法差别挺大。
---
**标签**：`Python` `电商` `API` `订单同步` `库存管理`未分类
还在人工搬商品？淘宝、1688、京东数据一键通（附 Python 源码）
万邦科技Lex 发表于2026-10-20 浏览69 评论0
# 还在人工搬商品？淘宝、1688、京东数据一键通（附 Python 源码）
> 一个商品在三个平台搬一遍要 40 分钟，100 个商品就是 3 天。这不是人力问题，是流程问题。
## 一、先算一笔账
多平台开店的都干过这事：
打开淘宝后台，复制标题 → 下载 5 张主图 → 复制详情页 → 切到 1688 后台 → 粘贴标题 → 逐张上传图片 → 重新选类目 → 一个个填属性 → 保存 → 再切到京东 → 重复一遍。
一个商品，熟练的运营也要 40 分钟。100 个商品上新，三天没了。
更麻烦的是**后面还会变**。价格改了、库存变了、主图换了，三个平台得再同步一轮。人工做，一定会漏。
真正让人难受的不是慢，是**这件事毫无技术含量，却占掉了运营最贵的时间**。
## 二、人工搬运到底卡在哪
动手写代码之前，先看清楚障碍在哪。五个，缺一个都做不成：
| 卡点 | 具体表现 |
| -------- | ------------------------------------------------- |
| 字段名完全不通用 | 淘宝 `title`，1688 `subject`，京东 `wareName`，同一个东西三个名字 |
| 图片不能外链 | 平台图片有防盗链和时效签名，必须转存到自己的存储再上传 |
| 类目体系各自独立 | 三个平台的叶子类目 ID 毫无关联，淘宝「女装/连衣裙」和 1688 的类目不是一个东西 |
| 详情页带平台痕迹 | 复制过来的 HTML 里有原平台域名、脚本、活动链接，直接铺会被审核驳回 |
| 批量接口是异步的 | 提交后返回一个 task_id，得轮询任务状态，不是同步返回成功 |
其中**类目映射**是最容易被低估的。字段名看文档能对上，类目对不上，商品根本发不出去。
## 三、方案：一条五段流水线
```
采集 → 归一化 → 映射 → 上传 → 回写
│ │ │ │ │
拉原始 统一成 字段+类目 图片转存 记录目标
数据 Item 转目标格式 分批提交 平台 ID
```
四个设计原则：
1. **中间一定要有统一模型。** N 个平台互转，如果两两写映射就是 N² 份代码；先归一化，就只需要 N 份。三个平台是 9 条路径，归一化之后是 3 条。
2. **每一步都可重入。** 任何一步失败，重跑不能产生重复商品。靠 `outer_id`（外部编码）做幂等键。
3. **图片必须转存。** 不要试图把原平台图片地址直接传给目标平台，百分之百失败。
4. **类目映射留人工口子。** 自动映射做到 85% 就够了，剩下 15% 进复核队列，比追求 100% 划算得多。
下面上代码。
## 四、Python 源码
依赖：
```bash
pip install httpx
```
### 4.1 统一商品模型
先定义「一个商品到底由什么构成」。这个模型定得越准，后面的映射越省事。
```python
# models.py
from dataclasses import dataclass, field
from decimal import Decimal
@dataclass
class Sku:
sku_id: str
props: dict[str, str] # {"颜色": "黑色", "尺码": "L"}
price: Decimal
stock: int
outer_id: str | None = None # 商家编码，幂等键
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
@dataclass
class Item:
platform: str # 来源平台
item_id: str
title: str
category_id: str # 来源平台类目
category_path: list[str] # 类目路径，用于跨平台匹配
images: list[str] # 主图原始 URL
desc_html: str # 详情页 HTML
attrs: dict[str, str] # 商品属性
skus: list[Sku] = field(default_factory=list)
@property
def outer_id(self) -> str:
"""幂等键：优先商家编码，没有就用来源平台+ID"""
if self.skus and self.skus[0].outer_id:
return self.skus[0].outer_id
return f"{self.platform}_{self.item_id}"
```
### 4.2 三平台采集
签名部分和上一篇一样（淘宝/京东 MD5 包夹，1688 用 HMAC-SHA1），这里只保留解析逻辑，重点看三个平台返回结构差多少。
```python
# collectors.py
import asyncio
import time
import httpx
from decimal import Decimal
from models import Item, Sku
from signers import sign_taobao, sign_alibaba, sign_jd
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def fetch_taobao(client, app_key, secret, item_id) -> Item | None:
params = {
"method": "taobao.item.seller.get",
"app_key": app_key, "timestamp": str(int(time.time())),
"format": "json", "v": "2.0", "sign_method": "md5",
"fields": "num_iid,title,cid,pic_url,desc,props_name,sku",
"num_iid": item_id,
}
params["sign"] = sign_taobao(params, secret)
r = await client.post("https://eco.taobao.com/router/rest", data=params)
d = (r.json().get("item_get_response") or {}).get("item")
if not d:
return None
skus = []
for s in (d.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)),
outer_id=s.get("outer_id") or None,
))
return Item(
platform="taobao",
item_id=str(d["num_iid"]),
title=d.get("title", ""),
category_id=str(d.get("cid", "")),
category_path=[], # 需另调类目接口补全
images=[d["pic_url"]] if d.get("pic_url") else [],
desc_html=d.get("desc", "") or "",
attrs=_parse_props(d.get("props_name", "")),
skus=skus,
)
async def fetch_alibaba(client, app_key, secret, product_id) -> Item | None:
api = "/openapi/param2/1/com.alibaba.product/alibaba.product.get"
params = {"access_token": app_key, "productID": product_id, "webSite": "1688"}
params["_aop_signature"] = sign_alibaba(params, secret)
r = await client.post(
"https://gw.open.1688.com" + api + "/" + app_key, data=params
)
d = (r.json() or {}).get("result") or {}
if not d:
return None
ranges = (d.get("saleInfo") or {}).get("priceRanges") or []
price = Decimal(str(ranges[0]["price"])) if ranges else Decimal("0")
images = [i["fullPathImageURI"] for i in (d.get("image") or {}).get("images", [])]
return Item(
platform="alibaba",
item_id=str(d.get("productID", product_id)),
title=d.get("subject", ""),
category_id=str(d.get("categoryID", "")),
category_path=[],
images=images,
desc_html=d.get("description", "") or "",
attrs={a.get("attributeName", ""): a.get("attributeValue", "")
for a in (d.get("attributes") or [])},
skus=[],
)
async def fetch_jd(client, app_key, secret, sku_id) -> Item | None:
params = {
"method": "jd.union.open.goods.query",
"app_key": app_key, "timestamp": "2026-01-01 00:00:00",
"format": "json", "v": "1.0",
"360buy_param_json": '{"skuIds":[%s]}' % sku_id,
}
params["sign"] = sign_jd(params, secret)
r = await client.post("https://api.jd.com/routerjson", data=params)
rows = ((r.json().get("jd_union_open_goods_query_responce") or {})
.get("result") or {}).get("data") or []
if not rows:
return None
g = rows[0]
return Item(
platform="jd",
item_id=str(g.get("skuId", sku_id)),
title=g.get("skuName", ""),
category_id=str(g.get("cid2", "")),
category_path=[g.get("cid1Name", ""), g.get("cid2Name", "")],
images=[g["imageUrl"]] if g.get("imageUrl") else [],
desc_html="",
attrs={},
skus=[Sku(
sku_id=str(g.get("skuId", sku_id)),
props={},
price=Decimal(str(g.get("price", "0"))),
stock=0,
)],
)
def _parse_props(s: str) -> dict[str, str]:
"""淘宝的 props_name 格式：'1627207:28320:颜色:黑色;20509:28314:尺码:L'"""
out = {}
for seg in filter(None, s.split(";")):
parts = seg.split(":")
if len(parts) >= 4:
out[parts[2]] = parts[3]
return out
```
注意最后那个 `_parse_props`。淘宝把属性塞在一个 `k:v:k:v;` 的字符串里，1688 给的是结构化数组，京东干脆只给类目名。**同一个「颜色」，三个平台三种给法**，这是归一化模型存在的全部理由。
### 4.3 字段映射 + 详情页清洗
```python
# mapper.py
import re
from decimal import Decimal
# 统一字段 → 各平台字段名
FIELD_MAP = {
"alibaba": {"title": "subject", "price": "price", "stock": "amountOnSale",
"images": "image", "desc": "description", "category": "categoryID"},
"jd": {"title": "wareName", "price": "jdPrice", "stock": "stockNum",
"images": "imagePath", "desc": "introduction", "category": "categoryId"},
"taobao": {"title": "title", "price": "price", "stock": "quantity",
"images": "pic_url", "desc": "desc", "category": "cid"},
}
def clean_desc(html: str) -> str:
"""去掉原平台的脚本、外链、活动痕迹"""
if not html:
return ""
html = re.sub(r"<script[\s\S]*?</script>", "", html, flags=re.I)
html = re.sub(r"<iframe[\s\S]*?</iframe>", "", html, flags=re.I)
html = re.sub(r'\son\w+="[^"]*"', "", html) # 内联事件
html = re.sub(r"https?://[^\"'\s>]*(taobao|tmall|1688|jd)\.com[^\"'\s>]*",
"#", html, flags=re.I) # 平台域名
return html.strip()
def to_alibaba(item) -> dict:
m = FIELD_MAP["alibaba"]
return {
m["title"]: item.title[:60], # 1688 标题上限 60 字符
m["category"]: item.category_id,
m["desc"]: clean_desc(item.desc_html),
"saleInfo": {
"priceRanges": [{
"startQuantity": 1,
"price": float(item.skus[0].price if item.skus else Decimal("0")),
}],
"amountOnSale": item.skus[0].stock if item.skus else 0,
},
"attributes": [{"attributeName": k, "attributeValue": v}
for k, v in item.attrs.items()],
}
def to_jd(item) -> dict:
m = FIELD_MAP["jd"]
return {
m["title"]: item.title,
m["category"]: item.category_id,
m["price"]: str(item.skus[0].price if item.skus else "0"),
m["stock"]: item.skus[0].stock if item.skus else 0,
m["images"]: ",".join(item.images[:5]),
}
```
`clean_desc` 这段别省。我见过太多人直接把详情 HTML 原样铺过去，结果详情页里全是原店铺的跳转链接和活动倒计时脚本，审核直接驳回，还查不出原因。
### 4.4 类目映射：规则 + 兜底
类目 ID 之间没有数学关系，只能靠语义匹配。
```python
# category.py
CATEGORY_DICT = {
"alibaba": [
(["连衣裙", "裙子", "女装"], "女装 / 连衣裙", "201234567"),
(["T恤", "短袖"], "男装 / T恤", "201234568"),
(["手机壳"], "数码 / 手机配件", "201234569"),
(["保温杯", "水杯"], "日用百货 / 杯具", "201234570"),
],
"jd": [
(["连衣裙", "裙子", "女装"], "女装 / 连衣裙", "1319,1523,1524"),
(["T恤", "短袖"], "男装 / T恤", "1319,1523,1525"),
(["手机壳"], "数码 / 手机配件", "652,829,854"),
(["保温杯", "水杯"], "家居 / 杯具", "1620,11953,11957"),
],
}
def map_category(item, target: str) -> tuple[str | None, float]:
"""返回 (目标类目ID, 置信度)。低于 0.6 建议进人工复核"""
text = item.title + " " + " ".join(item.category_path) + " " + " ".join(item.attrs.values())
rules = CATEGORY_DICT.get(target, [])
best, best_score = None, 0.0
for keywords, _path, cid in rules:
hits = sum(1 for k in keywords if k in text)
if not hits:
continue
score = min(1.0, hits / max(1, len(keywords)))
if score > best_score:
best, best_score = cid, score
return best, best_score
```
实际项目里这张字典会有几百条，来源一般是「把目标平台的热销类目导出来，手工对照一遍」。机器做初筛，人工确认一次，之后就能复用。
### 4.5 批量搬运引擎：任务表 + 断点续传
批量任务的真正难点不是并发，是**失败了怎么继续**。跑到第 700 个挂了，不能从头再来。
```python
# runner.py
import asyncio
import sqlite3
from datetime import datetime
class TransferEngine:
def __init__(self, db_path="transfer.db"):
self.db = sqlite3.connect(db_path, check_same_thread=False)
self.db.execute("""
CREATE TABLE IF NOT EXISTS job (
id INTEGER PRIMARY KEY AUTOINCREMENT,
src_platform TEXT, src_id TEXT, dst_platform TEXT,
status TEXT DEFAULT 'pending', -- pending/doing/done/failed
attempts INTEGER DEFAULT 0,
dst_id TEXT, error TEXT,
updated_at TEXT,
UNIQUE(src_platform, src_id, dst_platform)
)
""")
self.db.commit()
def add_jobs(self, src_platform: str, ids: list[str], dst_platform: str):
self.db.executemany(
"INSERT OR IGNORE INTO job(src_platform, src_id, dst_platform) VALUES(?,?,?)",
[(src_platform, i, dst_platform) for i in ids],
)
self.db.commit()
def take(self, limit: int) -> list[tuple]:
rows = self.db.execute(
"SELECT id, src_platform, src_id, dst_platform, attempts "
"FROM job WHERE status IN ('pending','failed') AND attempts < 3 "
"ORDER BY attempts, id LIMIT ?", (limit,)
).fetchall()
for r in rows:
self.db.execute("UPDATE job SET status='doing' WHERE id=?", (r[0],))
self.db.commit()
return rows
def finish(self, job_id, ok: bool, dst_id=None, err=None):
if ok:
self.db.execute(
"UPDATE job SET status='done', dst_id=?, updated_at=? WHERE id=?",
(dst_id, datetime.now().isoformat(timespec="seconds"), job_id))
else:
self.db.execute(
"UPDATE job SET status='failed', attempts=attempts+1, error=?, "
"updated_at=? WHERE id=?",
(str(err)[:500], datetime.now().isoformat(timespec="seconds"), job_id))
self.db.commit()
def stats(self) -> dict:
rows = self.db.execute(
"SELECT status, COUNT(*) FROM job GROUP BY status").fetchall()
return dict(rows)
async def run(engine: TransferEngine, worker, concurrency: int = 3):
"""worker(job) -> dst_id，抛异常算失败"""
sem = asyncio.Semaphore(concurrency)
async def one(job):
job_id, src_p, src_id, dst_p, attempts = job
async with sem:
try:
dst_id = await worker(src_p, src_id, dst_p)
engine.finish(job_id, True, dst_id=dst_id)
print(f"[OK] {src_p}:{src_id} -> {dst_p}:{dst_id}")
except Exception as e:
engine.finish(job_id, False, err=e)
print(f"[FAIL] {src_p}:{src_id} ({attempts + 1}/3) {e}")
await asyncio.sleep(2 ** attempts) # 指数退避
while True:
batch = engine.take(10)
if not batch:
break
await asyncio.gather(*[one(j) for j in batch])
print("进度:", engine.stats())
```
这个 `take` 里 `status IN ('pending','failed')` 加 `attempts < 3`，就是断点续传的全部秘密：失败的任务会被重新捞出来，但最多试 3 次，避免死循环卡住整批。
### 4.6 跑起来
```python
# main.py
import asyncio
import httpx
from collectors import fetch_taobao, fetch_alibaba, fetch_jd
from mapper import to_alibaba, to_jd
from category import map_category
from runner import TransferEngine, run
async def worker(src_p, src_id, dst_p, client, cfg):
fetcher = {"taobao": fetch_taobao, "alibaba": fetch_alibaba, "jd": fetch_jd}[src_p]
item = await fetcher(client, cfg["key"], cfg["secret"], src_id)
if not item:
raise RuntimeError("源商品不存在或已下架")
cid, score = map_category(item, dst_p)
if not cid or score < 0.6:
raise RuntimeError(f"类目无法匹配（置信度 {score:.2f}），转人工复核")
item.category_id = cid
item.images = [await transfer_image(u) for u in item.images] # 必须转存
payload = to_alibaba(item) if dst_p == "alibaba" else to_jd(item)
r = await client.post(f"{cfg['gateway']}/product/create", json=payload)
r.raise_for_status()
data = r.json()
if not data.get("success"):
raise RuntimeError(data.get("msg", "创建失败"))
return data["productId"]
async def transfer_image(url: str) -> str:
"""下载原图 → 上传自己的对象存储 → 返回可用地址"""
# 实际实现接你的 OSS / 平台图片上传接口
# 关键点：不要返回源平台地址，目标平台一定拉不到
return url.replace("https://img.alicdn.com", "https://cdn.yourdomain.com")
async def main():
cfg = {"key": "your_appkey", "secret": "your_secret",
"gateway": "https://api.yourdomain.com"}
engine = TransferEngine()
engine.add_jobs("taobao", ["654321098765", "654321098766"], "alibaba")
async with httpx.AsyncClient(timeout=30) as client:
async def _w(sp, si, dp):
return await worker(sp, si, dp, client, cfg)
await run(engine, _w, concurrency=3)
print("最终:", engine.stats())
if __name__ == "__main__":
asyncio.run(main())
```
## 五、实测效果
一个做家居用品的中型店，从淘宝往 1688 和京东铺货：
| 指标 | 人工 | 脚本 |
| --------- | --------- | ---------------- |
| 单个商品耗时 | 40 分钟 | 约 25 秒 |
| 100 个商品 | 3 天（还不算漏） | 约 12 分钟 |
| 类目对不上 | 靠经验试错 | 自动匹配 87%，13% 转人工 |
| 漏改价格 / 库存 | 常态 | 0 |
真正省下来的不只是时间，是**上新节奏**。人工三天才能铺完的款，脚本跑完你当天就能测。
## 六、五个踩过的坑
1. **幂等键一定要设。** 用商家编码 `outer_id` 做唯一键。没有它，重跑一次就是双倍商品，删起来比重铺还麻烦。
2. **图片必须转存。** 平台图片地址有签名时效，几小时后就失效。别想着偷懒直接传。
3. **详情页 HTML 必洗。** 脚本、iframe、平台域名，一个都不能留，否则审核驳回还查不出原因。
4. **批量接口是异步的。** 提交返回 `task_id`，要轮询到 `status=3` 才叫成功。把提交当成功，会有一半商品实际没发出去。
5. **类目别追求 100%。** 85% 自动 + 15% 人工复核，是成本最低的平衡点。硬做全自动，后期维护规则的精力比人工还多。
## 七、最后
这套东西的核心不在代码量，在**那个统一模型**。模型定对了，接第四个平台只是多写一个 20 行的采集器；模型定错了，每接一个平台都要改一遍全部逻辑。
代码在上面，改个 appkey 就能跑。签名、限流、字段映射这些琐事，如果你不想一个个啃文档，也可以用现成的三平台 API 接口服务，淘宝 / 1688 / 京东的商品详情、SKU、库存、图片一次拿全，格式统一，直接对接上层的映射逻辑。
有在做的可以聊聊类目映射这块，各家的规则表是怎么维护的。
---
**标签**：`Python` `电商` `API` `数据采集` `多平台运营`未分类
商品价格总是慢半拍？三平台数据实时同步的做法（附 Python 源码）
万邦科技Lex 发表于2026-10-20 浏览72 评论0
# 商品价格总是慢半拍？三平台数据实时同步的做法（附 Python 源码）
> 淘宝 / 1688 / 京东 三平台商品价格同步，从「每天跑一次脚本」到「分钟级感知」，中间差的不只是一个定时任务。
未分类
《亚马逊 SP-API 订单同步：getOrders 实时查询 vs Reports 批量报告 —— 成本/限流/时延三维落地方案》（附 Python 源码）
万邦科技Lex 发表于2026-10-19 浏览78 评论0
《亚马逊 SP-API 订单同步：getOrders 实时查询 vs Reports 批量报告 —— 成本/限流/时延三维落地方案》（附 Python 源码）
0. 先更新一个前提（2026-10 今天）
你标题里“年费
未分类
《亚马逊 SP-API 商品API：getCatalogItem 与 searchCatalogItems 的 ASIN / UPC / EAN / GTIN 映射踩坑》（附 Python 源码）
万邦科技Lex 发表于2026-10-19 浏览78 评论0
《亚马逊 SP-API 商品API：getCatalogItem 与 searchCatalogItems 的 ASIN / UPC / EAN / GTIN 映射踩坑》（附 Python 源码）
一、先拍结论
亚马逊商品域最阴的不是“调不通”，而是标识符语义不对等：