# 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 源码）

> 驳回理由是「商品信息不符合规范」，七个字，什么信息都没给你。真正的答案在类目属性接口里。

## 一、先看一个真实的比例

一个做小家电的商家，从淘宝铺 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` `电商` `京东` `商品发布` `数据校验`