一、前言

做电商 SaaS、ERP 进销存、多店管理系统时，最核心需求之一就是一次性拉取目标店铺全量商品，完成商品库同步、库存价格监控、数据对账、多平台铺货等业务。 很多新手直接硬写循环翻页，极易遇到签名错误、QPS 限流、10 万条翻页限制、授权失效、SKU 信息缺失等坑。本文基于 2026 淘宝开放平台 TOP 官方接口，给出一套稳定可落地的批量抓取架构、接口选型、签名、分页限流、异常重试、增量同步完整方案，附带可运行 Python 工程代码。

适用场景

1. 自有淘宝店铺 ERP 数据同步
2. 第三方 SaaS 多店铺商品统一管理
3. 店铺商品批量导出、数据盘点
4. 定时增量同步价格、库存、标题

二、前置准备工作

2.1 开放平台账号与应用

1. 注册淘宝开放平台，完成企业 / 个人实名认证
2. 创建第三方应用 / 自用商家应用，拿到 app_key、app_secret（密钥禁止明文泄露、禁止前端存放）
3. 申请接口权限：
   • 基础列表：taobao.items.onsale.get（在售）、taobao.items.inventory.get（仓库下架）
   • 商品详情：taobao.item.seller.get（拿 SKU、详情、属性、库存）
4. OAuth2 授权商家店铺，拿到长期有效 session（access_token），无 session 无法读取私有店铺商品数据。

2.2 核心接口选型对比

抓取全店商品分列表分页+详情补全两步，不要用单一接口硬扛：

表格

接口名称	作用	限制	使用场景
taobao.items.onsale.get	获取在售商品列表	page_size 最大 200；翻页总量超 10 万报错	主力拉取在售商品 ID、基础信息
taobao.items.inventory.get	获取仓库 / 下架商品列表	同在售接口	抓取仓库未上架、下架商品
taobao.item.seller.get	单商品完整详情（SKU、描述、规格、库存）	单条调用	拿到 num_iid 后批量补全详情
taobao.items.list.get	批量多商品 ID 查基础信息	单次最多 20 个 num_iid	批量精简查询，减少请求次数

全店完整商品 = 在售列表 + 仓库下架列表，两者合并才是店铺全部商品池。

三、整体批量抓取架构方案

3.1 标准执行流程

1. 初始化配置：app_key、密钥、session、QPS 阈值、数据库存储地址
2. 分页循环拉取在售商品：page_no 从 1 递增，读取 total_results 计算总页数
3. 分页循环拉取仓库下架商品
4. 合并所有商品 num_iid 集合，去重
5. 分组批量调用详情接口，补齐 SKU、图文、规格、真实库存
6. 数据清洗、格式化，入库 / 导出 CSV
7. 增量同步优化：按start_modified修改时间分段抓取，无需全量刷新
8. 全局限流、失败重试、错误日志告警

3.2 关键约束风险点（官方硬性规则）

1. 翻页上限：page_no × page_size > 100000 直接报错，大店铺必须用时间分片拆分抓取
2. QPS 限流：普通应用 QPS≈3~5，企业签约可提升；超速返回 403 限流错误
3. Session 有效期：普通授权一般 1 年，过期需重新引导商家授权
4. Fields 字段按需填写：只拉业务需要字段，减少返回包体积、提升速度

四、淘宝 TOP API 标准签名实现（MD5）

所有请求必须携带 sign 签名，签名错误 100% 调用失败，严格遵循官方规则：

1. 收集全部请求参数（除 sign 本身）
2. 按键名 ASCII 升序排序
3. 拼接字符串：app_secret + k1v1k2v2... + app_secret
4. MD5 加密，结果转大写作为 sign 值

Python 签名工具函数：

import hashlib

def create_sign(params: dict, app_secret: str) -> str:
    # 按键排序
    sorted_items = sorted(params.items(), key=lambda x: x[0])
    sign_src = app_secret
    for k, v in sorted_items:
        if v is not None and str(v).strip():
            sign_src += f"{k}{v}"
    sign_src += app_secret
    md5_res = hashlib.md5(sign_src.encode("utf-8")).hexdigest()
    return md5_res.upper()

通用请求固定参数

每次调用都必须携带：

base_params = {
    "app_key": APP_KEY,
    "v": "2.0",
    "format": "json",
    "sign_method": "md5",
    "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
    "session": SHOP_SESSION
}

网关固定地址：https://gw.api.taobao.com/router/rest

五、分页批量拉取商品列表核心逻辑

5.1 在售商品分页封装

python

运行

import requests
import time

GATEWAY_URL = "https://gw.api.taobao.com/router/rest"
# 全局限流时间戳
last_req_time = 0
QPS_LIMIT = 4  # 控制4次/秒，安全阈值

def qps_control():
    global last_req_time
    now = time.time()
    interval = 1 / QPS_LIMIT
    diff = now - last_req_time
    if diff < interval:
        time.sleep(interval - diff)
    last_req_time = time.time()

def fetch_item_list(method: str, page_no: int, page_size: int = 100, start_modified: str = None):
    qps_control()
    params = base_params.copy()
    params["method"] = method
    params["page_no"] = page_no
    params["page_size"] = page_size
    # 按需指定返回字段，精简带宽
    params["fields"] = "num_iid,title,price,nick,cid,pic_url,modified,num,outer_id"
    if start_modified:
        params["start_modified"] = start_modified
    
    # 生成签名
    sign = create_sign(params, APP_SECRET)
    params["sign"] = sign
    
    resp = requests.get(GATEWAY_URL, params=params, timeout=15)
    data = resp.json()
    
    # 错误捕获
    if "error_response" in data:
        err = data["error_response"]
        code = err.get("code")
        msg = err.get("msg")
        print(f"接口异常 code:{code}, msg:{msg}")
        # 限流错误自动重试2次
        if code in [403, 15] and retry_count < 2:
            time.sleep(2)
            return fetch_item_list(method, page_no, page_size, start_modified)
        return None
    
    resp_key = f"{method.replace('.','_')}_response"
    res_body = data[resp_key]
    total = res_body.get("total_results", 0)
    items = res_body.get("items", {}).get("item", [])
    # 单条转列表兼容空数据
    if not isinstance(items, list):
        items = [items]
    return {"total": total, "items": items}

5.2 全分页循环抓取函数

def get_all_shop_items():
    all_num_iid = []
    # 1. 抓取在售商品
    page = 1
    while True:
        ret = fetch_item_list("taobao.items.onsale.get", page, page_size=100)
        if not ret:
            break
        total = ret["total"]
        item_list = ret["items"]
        if not item_list:
            break
        for item in item_list:
            all_num_iid.append(item["num_iid"])
        print(f"在售第{page}页，拿到{len(item_list)}件，总计:{total}")
        if page * 100 >= total:
            break
        page += 1
    
    # 2. 抓取仓库下架商品
    page = 1
    while True:
        ret = fetch_item_list("taobao.items.inventory.get", page, page_size=100)
        if not ret:
            break
        total = ret["total"]
        item_list = ret["items"]
        if not item_list:
            break
        for item in item_list:
            all_num_iid.append(item["num_iid"])
        print(f"仓库第{page}页，拿到{len(item_list)}件，总计:{total}")
        if page * 100 >= total:
            break
        page += 1
    
    # 去重
    unique_iid = list(set(all_num_iid))
    print(f"店铺商品总数(去重)：{len(unique_iid)}")
    return unique_iid

六、批量补全商品 SKU 详情

拿到所有num_iid后，单条调用taobao.item.seller.get获取完整规格、库存、详情描述；也可以用taobao.items.list.get每 20 个 ID 批量查询提速：

def fetch_item_detail(num_iid: int):
    qps_control()
    params = base_params.copy()
    params["method"] = "taobao.item.seller.get"
    params["num_iid"] = num_iid
    # 全量核心字段
    params["fields"] = "num_iid,title,price,desc,sku,skus,item_img,props,stock_num,outer_id"
    params["sign"] = create_sign(params, APP_SECRET)
    resp = requests.get(GATEWAY_URL, params=params, timeout=15)
    data = resp.json()
    if "error_response" in data:
        print(f"详情查询失败 {num_iid}: {data['error_response']}")
        return None
    return data["taobao_item_seller_get_response"]["item"]

# 批量循环获取所有详情
iid_list = get_all_shop_items()
full_goods = []
for iid in iid_list:
    detail = fetch_item_detail(iid)
    if detail:
        full_goods.append(detail)
# 此处可写入MySQL/导出CSV

七、企业级优化方案

7.1 大店铺 10 万条翻页限制解决

当店铺商品超 10 万，直接 page 翻页会报错，改用时间分片抓取：

1. 首次全量抓取记录每条商品modified更新时间
2. 下次同步按时间段拆分：start_modified="2026-01-01 00:00:00"、end_modified="2026-03-01 00:00:00"分段循环抓取
3. 日常定时任务只抓最近 7 天修改商品，实现增量同步，速度提升 90%

7.2 限流与稳定性优化

1. QPS 保守设置 3~5，不要拉满；企业签约后可申请调高配额
2. 失败重试机制：限流、网络超时重试 2 次，间隔 2s；授权失效触发告警通知商家重授权
3. 异步队列架构：生产环境用 Redis 队列分发商品 ID，多消费线程隔离限流，防止同步阻塞主业务
4. 缓存机制：不变基础信息（标题、类目）本地缓存，只同步价格库存变动字段

7.3 数据存储与清洗规范

1. 核心存储字段：num_iid（唯一主键）、title、price、stock、sku 列表、status、modified、outer_id（商家编码关联 ERP）
2. SKU 单独分表存储，一条商品对应多条 SKU 记录，方便库存精准更新
3. 增加同步时间戳、同步状态标记，便于排查漏抓、失败商品

八、常见错误码排查

错误 Code	原因	解决办法
403	QPS 超限 / 调用频率太高	降低 QPS、增加 sleep、拆分任务
15	Session 失效 / 无权限	引导商家重新 OAuth 授权
27	签名错误	检查排序、密钥拼接、MD5 大写、无空值参数
20001	page 翻页总量超 10 万	启用时间分片分段抓取
52001	接口权限未申请	开放平台后台申请对应 API 权限并签约

九、合规重要提醒

1. 仅抓取已授权自有合作店铺数据，禁止无授权爬取陌生店铺（属于违规，平台封禁应用）
2. 数据仅限自身 ERP/SaaS 内部业务使用，禁止倒卖、批量分发商品数据
3. 严格使用官方 TOP API，禁止模拟网页爬虫抓页面，极易封号、封 IP
4. 定时同步频率不要过高，非促销期 1~4 小时同步一次足够

十、总结

批量抓取淘宝全店商品标准链路：授权获取 session → 分页拉取在售 + 仓库商品 ID → 去重 → 批量补全 SKU 详情 → 入库 / 增量定时同步。 本文代码可直接改造集成到 Python 电商 SaaS、ERP 后端；Java/Go 只需移植签名、请求、分页逻辑，架构思路完全通用。这套方案已经稳定支撑数十家多店管理系统，完美规避翻页上限、限流、签名三大高频坑点。