## 一、整体技术架构

一键上货系统的数据流如下：

```

货源平台（1688/淘宝等）          目标平台（淘宝/拼多多/抖店）

        │                              ▲

        ▼                              │

  ┌───────────┐    ┌───────────┐    ┌───────────┐

  │ 商品数据获取 │ →  │ 数据清洗与  │ →  │ 格式适配与  │

  │  (item.get) │    │  字段映射   │    │  批量发布   │

  └───────────┘    └───────────┘    └───────────┘

```

整个链路依赖两类API：**货源数据获取接口**（从上游拉取商品信息）和**商品发布接口**（向目标平台上架商品）。跨平台场景下，不同平台的字段体系差异很大，需要自建映射字典做字段转换——例如1688的“颜色分类”在部分海外平台叫“变体选项”，国内的计量单位需要转换成目标平台要求的格式。

## 二、第一步：获取货源商品数据

以1688商品详情接口为例，传入商品ID即可获取标题、价格、SKU、主图、详情图、属性等全量字段。

```python

import requests

import hashlib

import time

def get_source_product(app_key, app_secret, num_iid):

    """从1688获取货源商品数据"""

    url = "https://eco.taobao.com/router/rest"

    

    params = {

        "method": "alibaba.item.get",

        "app_key": app_key,

        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),

        "format": "json",

        "v": "2.0",

        "sign_method": "md5",

        "num_iid": num_iid,

        "fields": "title,price,pic_url,desc,sku_list,cid"

    }

    

    # 生成签名

    sorted_params = "".join(

        sorted([f"{k}{v}" for k, v in params.items()])

    )

    sign_str = app_secret + sorted_params + app_secret

    params["sign"] = hashlib.md5(sign_str.encode()).hexdigest()

    

    response = requests.post(url, data=params)

    result = response.json()

    

    if "error_response" in result:

        raise Exception(f"获取货源失败: {result['error_response']['msg']}")

    

    return result["item_get_response"]["item"]

```

调用后返回的JSON结构包含：`title`（标题）、`price`（价格）、`pic_url`（主图）、`sku_list`（SKU列表）、`desc`（详情HTML）等字段。

## 三、第二步：认证授权

调用商品发布接口前，必须完成平台的身份认证。主流电商平台普遍采用 **OAuth 2.0** 协议获取访问令牌。

```python

def get_access_token(client_id, client_secret):

    """通过OAuth 2.0获取访问令牌"""

    url = "https://api.platform.com/auth/token"

    payload = {

        "grant_type": "client_credentials",

        "client_id": client_id,

        "client_secret": client_secret

    }

    response = requests.post(url, data=payload)

    token_data = response.json()

    return token_data["access_token"]

```

不同平台的认证细节有所差异：

| 平台 | 认证方式 | 关键参数 |

|------|---------|---------|

| 淘宝 | AppKey + AppSecret + Session | session 通过OAuth授权获取 |

| 拼多多 | AppKey + AppSecret + access_token | access_token 通过OAuth获取 |

| 抖店 | app_key + access_token + sign | sign使用hmac-sha256签名 |

| Shopify | Access Token | X-Shopify-Access-Token 请求头 |

以抖店为例，其公共参数包括 `method`、`app_key`、`access_token`、`param_json`、`timestamp`、`sign`，签名算法推荐使用 **hmac-sha256**。

## 四、第三步：数据清洗与字段映射

货源平台和目标平台的字段命名和结构往往不同，需要做清洗和映射。这一步是保证上货成功率的关键。

```python

def clean_and_map(source_item, target_platform):

    """数据清洗与字段映射"""

    

    # 通用清洗：过滤异常数据

    if not source_item.get("title") or float(source_item.get("price", 0)) <= 0:

        raise ValueError("商品数据异常：标题为空或价格无效")

    

    # 平台字段映射表

    field_mapping = {

        "taobao": {

            "title": "title",

            "price": "price",

            "pic_url": "pic_url",

            "desc": "desc",

            "num": "num",

            "cid": "cid",

        },

        "pdd": {

            "title": "goods_name",

            "price": "price",

            "pic_url": "image_url",

            "desc": "description",

            "num": "quantity",

            "cid": "cat_id",

        },

        "doudian": {

            "title": "name",

            "price": "price",

            "pic_url": "pic",

            "desc": "description",

            "num": "stock_num",

            "cid": "category_leaf_id",

        }

    }

    

    mapping = field_mapping.get(target_platform, {})

    cleaned = {}

    for src_field, tgt_field in mapping.items():

        if src_field in source_item:

            cleaned[tgt_field] = source_item[src_field]

    

    # 价格微调：在货源价基础上加价

    if "price" in cleaned:

        cleaned["price"] = round(float(cleaned["price"]) * 1.3, 2)

    

    # 标题长度截断（抖店要求至少8个字符、最多60个字符）

    if target_platform == "doudian" and "name" in cleaned:

        cleaned["name"] = cleaned["name"][:60]

    

    return cleaned

```

需要注意的清洗要点：

- **过滤异常商品**：价格≤0、无主图、SKU为空的商品应直接跳过。

- **标题长度适配**：抖店要求商品名称至少8个字符、最多60个字符，不能含emoji。

- **图片处理**：抖店商品轮播图用“|”分隔，最多5张，每张至少600×600像素，大小不超过5M。

## 五、第四步：格式适配与商品发布

### 5.1 淘宝商品发布

淘宝使用 `taobao.item.add` 接口发布商品，请求参数包括标题、价格、库存、类目ID、图片URL、详情描述等。

```python

def publish_to_taobao(api_key, secret_key, session_key, product_data):

    """发布商品到淘宝店铺"""

    url = "https://api.taobao.com/router/rest"

    

    params = {

        "method": "taobao.item.add",

        "app_key": api_key,

        "session": session_key,

        "timestamp": str(int(time.time())),

        "format": "json",

        "v": "2.0",

        "sign_method": "md5",

        "title": product_data["title"],

        "price": str(product_data["price"]),

        "num": str(product_data["num"]),

        "cid": product_data["cid"],

        "desc": product_data.get("desc", ""),

        "pic_url": product_data["pic_url"],

    }

    

    # 生成签名

    sorted_params = "".join(

        sorted([f"{k}{v}" for k, v in params.items()])

    )

    sign_str = secret_key + sorted_params + secret_key

    params["sign"] = hashlib.md5(sign_str.encode()).hexdigest()

    

    response = requests.post(url, data=params)

    result = response.json()

    

    if "error_response" in result:

        error = result["error_response"]

        raise Exception(f"淘宝上货失败: {error['code']} - {error['msg']}")

    

    item = result["item_add_response"]["item"]

    return {

        "num_iid": item["num_iid"],

        "status": item["status"],

        "msg": "商品发布成功"

    }

```

成功返回示例：包含 `num_iid`（商品ID）、`title`、`price`、`status`（onsale）等字段。

### 5.2 拼多多商品发布

拼多多使用 `pdd.goods.add` 接口，结合预设的商品信息模板（如Excel或数据库）批量上传商品信息。

```python

def publish_to_pdd(access_token, product_data):

    """发布商品到拼多多"""

    url = "https://open-api.pinduoduo.com/api/goods/add"

    headers = {"Authorization": f"Bearer {access_token}"}

    

    payload = {

        "goods_name": product_data["title"],

        "price": int(float(product_data["price"]) * 100),  # 单位：分

        "quantity": product_data["num"],

        "image_url": product_data["pic_url"],

        "description": product_data.get("desc", ""),

        "cat_id": product_data["cid"],

    }

    

    response = requests.post(url, json=payload, headers=headers)

    result = response.json()

    

    if result.get("error_code"):

        raise Exception(f"拼多多上货失败: {result['error_msg']}")

    

    return result["goods_id"]

```

### 5.3 抖店商品发布

抖店使用 `/product/addV2` 接口，请求参数需要按照参数名字符串大小排序后传入 `param_json`。

```python

def publish_to_doudian(app_key, app_secret, access_token, product_data):

    """发布商品到抖店"""

    url = "https://openapi-fxg.jinritemai.com/product/addV2"

    

    param_json = json.dumps({

        "name": product_data["title"],

        "category_leaf_id": product_data["cid"],

        "pic": product_data["pic_url"],

        "description": product_data.get("desc", ""),

        "price": product_data["price"],

        "stock_num": product_data["num"],

        "product_type": 0,  # 0-普通商品

    }, sort_keys=True)

    

    params = {

        "method": "product.addV2",

        "app_key": app_key,

        "access_token": access_token,

        "param_json": param_json,

        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),

        "v": "2",

        "sign_method": "hmac-sha256",

    }

    

    # hmac-sha256签名

    sorted_params = "".join(

        sorted([f"{k}{v}" for k, v in params.items()])

    )

    import hmac

    sign = hmac.new(

        app_secret.encode(), sorted_params.encode(), hashlib.sha256

    ).hexdigest()

    params["sign"] = sign

    

    response = requests.post(url, data=params)

    result = response.json()

    

    if result.get("code") != 0:

        raise Exception(f"抖店上货失败: {result.get('message')}")

    

    return result["data"]["product_id"]

```

抖店发布商品时有一个常见坑点：商品属性中若包含特殊字符（如“>”），直接用接口返回的 `name` 字段即可，手动修改会导致 **40003 参数错误**。

## 六、第五步：批量上传与限流处理

### 6.1 分批处理

单次批量不宜过大，建议每批50条，避免触发平台限流。

```python

def batch_upload(products, publish_func, batch_size=50):

    """分批批量上传商品"""

    results = {"success": [], "failed": []}

    

    for i in range(0, len(products), batch_size):

        batch = products[i:i + batch_size]

        for product in batch:

            try:

                product_id = publish_func(product)

                results["success"].append({

                    "title": product["title"],

                    "id": product_id

                })

            except Exception as e:

                results["failed"].append({

                    "title": product.get("title", "未知"),

                    "error": str(e)

                })

        

        print(f"已完成 {min(i + batch_size, len(products))}/{len(products)}")

        

        # 批次间休眠，避免触发频率限制

        time.sleep(1)

    

    return results

```

### 6.2 限流与重试

各平台API都有调用频率限制（如淘宝每分钟100次），超出后返回 **429** 错误码。建议实现带退避策略的重试机制：

```python

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(

    stop=stop_after_attempt(3),

    wait=wait_exponential(multiplier=1, min=1, max=10)

)

def api_request_with_retry(url, payload, headers=None):

    """带重试的API请求"""

    response = requests.post(url, json=payload, headers=headers)

    

    if response.status_code == 429:

        retry_after = int(response.headers.get("Retry-After", 5))

        time.sleep(retry_after)

        raise Exception("触发限流，等待重试")

    

    response.raise_for_status()

    return response

def throttled_request(url, min_interval=0.01):

    """令牌桶限流"""

    global _last_request_time

    elapsed = time.time() - _last_request_time

    if elapsed < min_interval:

        time.sleep(min_interval - elapsed)

    _last_request_time = time.time()

    return requests.post(url)

```

对于批量上货场景，建议使用消息队列（如Celery + Redis）将上传任务异步化，避免因单次请求超时导致整个批量任务中断。

## 七、错误处理与常见问题

商品发布过程中常见的错误类型及处理方式：

| 错误类型 | 典型错误码 | 解决方案 |

|---------|-----------|---------|

| 参数缺失 | 40003 | 检查必填参数，参考平台API文档逐一核对 |

| 类目属性错误 | 类目属性不存在 | 确认类目ID有效，必选属性完整填写 |

| 频率超限 | 429 | 降低请求频率，增加批次间隔 |

| Token过期 | 401 | 刷新access_token后重试 |

| 数据校验失败 | 价格/重量异常 | 检查价格、重量、尺寸是否在合理范围内 |

淘宝上架时，接口会进行**全量校验**，如果缺少必选属性（如“流行款式名称”），即使其他参数都正确也会报错。建议在发布前先调用类目属性查询接口，确认所有必填字段已覆盖。

## 八、安全与工程化建议

1. **敏感数据加密**：AppSecret、access_token等凭证使用环境变量或密钥管理服务存储，不要硬编码在代码中。

2. **审计日志**：关键操作（商品创建、修改、删除）记录审计日志，便于问题追溯。

3. **幂等性保障**：使用 `outer_product_id`（外部商家编码）作为幂等键，避免重复发布同一商品。抖店推荐使用 `outer_product_id` 字段做唯一标识。

4. **监控告警**：部署监控系统实时检测API异常，当失败率超过阈值时自动告警。

5. **接口版本兼容**：各平台API版本迭代频繁，需关注版本变更公告，及时适配新版本。

## 九、总结

一键上货的技术链路清晰但细节繁多：**获取货源数据 → OAuth认证 → 数据清洗映射 → 格式适配 → 批量发布**，每一步都有平台特定的规则需要遵循。核心开发要点包括：

- 不同平台的字段体系和接口参数差异较大，统一抽象层和映射字典是工程化的关键；

- 限流处理和重试机制是保证批量任务稳定运行的必备能力；

- 错误码分类处理能够显著提升问题定位效率；

- 建议先用少量商品验证全流程，再逐步扩大批量规模。