🛒 一键铺货神器：利用1688 API实现淘宝/抖音自动同步货源（附Python源码）

⚠️ 合规说明：本文示例使用1688开放平台官方商品查询接口（需企业实名应用），淘宝/抖音需使用各自开放平台商品发布API并拥有店铺授权。

一、铺货链路架构

┌──────────────┐   1688 API    ┌────────────────────┐  平台发布API   ┌──────────────┐
│  ERP/铺货后台│ ────────────▶ │ ali1688_client      │ ────────────▶ │ 淘宝/抖音小店│
│  (输入链接/ID)│              │ get item_id+SKU+图  │   (taobao.item.add  │
└──────────────┘              │ +详情HTML+类目映射   │    / spu.add)   └──────────────┘
                              └────────────────────┘

二、Python：1688货源抓取 + 标准化铺货DTO

# ali1688_distributor.py
import hashlib
import time
import requests
import urllib.parse
from typing import Dict, List, Optional
from dataclasses import dataclass, field


# ──────────────────────────────────────────────
# 铺货中间对象（与平台无关）
# ──────────────────────────────────────────────
@dataclass
class DistributeGoods:
    outer_id: str            # 1688 offerId
    title: str               # 商品标题
    price: str                # 起批价（字符串）
    moq: int                  # 最小起订量
    main_images: List[str] = field(default_factory=list)
    detail_html: str = ""     # 商品详情HTML（注意：部分需单独接口/权限）
    skus: List[Dict] = field(default_factory=list)
    # 淘宝/抖音类目映射（需手动维护 or 规则引擎）
    target_category_id: str = ""

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class Ali1688SourceClient:
    """1688 货源抓取 Client（复用标准签名）"""
    GATEWAY = "https://gw.open.1688.com/openapi/http/2/1"

    def __init__(self, app_key: str, app_secret: str, access_token: str = None):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token

    def _sign(self, params: Dict) -> str:
        filtered = sorted((k, v) for k, v in params.items()
                          if v is not None and str(v).strip() != '')
        qs = ''.join(f"{k}{v}" for k, v in filtered)
        raw = f"{self.app_secret}{qs}{self.app_secret}"
        return hashlib.md5(raw.encode('utf-8')).hexdigest().upper()

    def _call(self, method: str, biz: Dict) -> Dict:
        api_params = {
            "method": method,
            "app_key": self.app_key,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
        }
        if self.access_token:
            api_params["session"] = self.access_token
        api_params["param2"] = urllib.parse.quote_plus(
            str(biz).replace("'", '"')
        )
        api_params["sign"] = self._sign(api_params)

        resp = requests.get(self.GATEWAY, params=api_params, timeout=15)
        resp.raise_for_status()
        data = resp.json()
        if "error_response" in data:
            err = data["error_response"]
            raise Exception(f"1688 Err[{err.get('code')}]: {err.get('msg')}")
        return data.get(list(data.keys()-{'error_response'})[0], {})

    def fetch_goods(self, offer_id: str) -> DistributeGoods:
        """
        根据1688 offerId 拉取铺货所需全量信息
        注意：detail_html 有时需 alibaba.item.getDesc 接口单独取
        """
        biz = {
            "item_id": offer_id,
            "fields": "item_id,title,price,sku_list,pics,spec_info,"
                     "min_order_quantity,desc"
        }
        res = self._call("alibaba.item.get", biz)
        item = res.get("item", {})

        goods = DistributeGoods(
            outer_id=str(offer_id),
            title=item.get("title", ""),
            price=str(item.get("price", "")),
            moq=int(item.get("min_order_quantity") or 1),
            main_images=item.get("pics", [])[:5],   # 取前5张主图
            skus=item.get("sku_list", []),
        )

        # 详情HTML（部分返回字段含desc，若无需单独调 desc 接口）
        if "desc" in item:
            goods.detail_html = item.get("desc", "")

        return goods

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ──────────────────────────────────────────────
# 示例：转换为淘宝发布参数（淘宝 top SDK 类似结构）
# ──────────────────────────────────────────────
def to_taobao_item_add_params(g: DistributeGoods,
                               shop_cat_id: str,
                               freight_tpl_id: str) -> Dict:
    """
    映射为淘宝 taobao.item.add 业务参数
    （实际需用 top-sdk-python 调用，此处仅展示字段映射）
    """
    # SKU 处理示例（取第一个规格文本）
    sku_attrs = ", ".join(
        s.get("spec_attributes", "") for s in g.skus[:3]
    ) if g.skus else ""

    return {
        "num_iid": None,          # 新增为空
        "title": g.title[:60],    # 淘宝标题限字
        "cat_id": shop_cat_id,    # 需提前匹配类目
        "price": g.price,
        "num": 999,                # 淘宝需填库存（可用MOQ或自定义）
        "stuff_status": "new",
        "location.city": "杭州",   # 按实际填
        "image_urls": ";".join(g.main_images),
        "description": g.detail_html or g.title,
        "sku_properties": sku_attrs,
        "freight_payer": "seller",
        "post_fee": "0",
        "express_free_fee": "0",
        "has_invoice": "false",
        "is_taobao": "true",
        "is_ex": "false",
        "template_id": freight_tpl_id
    }

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ──────────────────────────────────────────────
# 示例：转换为抖音小店 SPU 创建参数（简化）
# ──────────────────────────────────────────────
def to_douyin_spu_params(g: DistributeGoods,
                          category_id: str) -> Dict:
    return {
        "name": g.title[:60],
        "category_id": category_id,
        "main_images": g.main_images,
        "price_info": [{
            "price": int(float(g.price) * 100),  # 抖音单位：分
            "sku_id": g.outer_id
        }],
        "description": g.detail_html or g.title,
        "outer_id": g.outer_id,
        "specs": [
            {"name": "规格", "values": [s.get("spec_attributes","默认")
             for s in g.skus]} if g.skus else [{"name":"规格","values":["默认"]}]
        ]
    }


# =========================================================
# 使用示例
# =========================================================
if __name__ == "__main__":
    client = Ali1688SourceClient(
        app_key="YOUR_APP_KEY",
        app_secret="YOUR_APP_SECRET",
        access_token=None   # 商品详情可不传
    )

    try:
        goods = client.fetch_goods("612345678901")   # ← 1688商品ID
        print(f"📦 商品: {goods.title}")
        print(f"   起批价: ¥{goods.price}  MOQ:{goods.moq}")
        print(f"   主图: {goods.main_images[0] if goods.main_images else 'N/A'}")
        print(f"   SKU数: {len(goods.skus)}")

        # —— 映射到平台参数（实际传给各自 top-sdk / douyin-open-sdk）——
        tb_params = to_taobao_item_add_params(goods, "50011740", "1")
        dy_params = to_douyin_spu_params(goods, "1234567890")

        print("\n✅ 已生成淘宝/抖音铺货参数（调用各自SDK上传即可）")

    except Exception as e:
        print(f"❌ 铺货源数据获取失败: {e}")

三、关键对接点说明

1688 → 淘宝/抖音 字段映射要点

类目映射（必须处理）

1688_leaf_category_id  →  taobao_cid  /  douyin_category_id

四、生产级建议

1. 图片转存：1688返回的是1688 CDN图片URL，发布到淘宝/抖音通常要求先下载转存到自己图床或平台素材空间，再传URL

2. 详情HTML清洗：1688详情含1688域名图片，建议正则替换 <img src=为转存后地址

3. SKU价格阶梯：1688有批发阶梯价，淘宝通常取最低价或自定义零售加价公式 retail_price = wholesale * markup

4. 草稿而非直发：先调平台API创建草稿状态，人工审核图片/标题后再上架

5. 幂等：以 outer_id=1688_offerId做幂等，重复铺货更新草稿不新建重复商品

五、面试/方案一句话

一键铺货 = 1688 alibaba.item.get取结构化商品数据 → 映射淘宝/抖音商品字段 → 调平台发布API建草稿，全程用官方API合法合规；重点是类目映射、图片转存和SKU规格对齐。