# 还在人工搬商品？淘宝、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` `数据采集` `多平台运营`