🏪 淘宝店铺全量商品列表API批量拉取：参数配置・分页・限流处理（附Python源码）

一、接口速览

⚠️ 必须用卖家自身AccessToken调用，不能用买家账号查别人店铺全量商品（淘宝隐私限制）。第三方ISV需商户授权后查其店铺。

二、Python：全量分页拉取 + 令牌桶限速 + 限流重试

# top_items_batch.py
"""
淘宝店铺全量商品批量拉取
依赖: top_api_client.py 中的 TaobaoTopClient
"""
import time
from typing import List, Dict, Generator
from top_api_client import TaobaoTopClient
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

class _TokenBucket:
    def __init__(self, rate=5, cap=None):
        self.rate = rate
        self.cap = cap or int(rate)
        self.tok = float(self.cap)
        self.ts = time.monotonic()

    def wait(self):
        now = time.monotonic()
        self.tok = min(self.cap, self.tok + (now - self.ts) * self.rate)
        self.ts = now
        if self.tok >= 1:
            self.tok -= 1
            return
        time.sleep((1 - self.tok) / self.rate + 0.01)
        self.tok = 0


class TopShopItemsSyncer:
    """
    店铺商品全量同步器
    默认查【在售】，inventory=True 查仓库中
    """

    def __init__(self, app_key: str, app_secret: str,
                 session: str, sandbox=False, qps=5):
        self.client = TaobaoTopClient(app_key, app_secret,
                                      sandbox=sandbox)
        self.session = session
        self.bucket = _TokenBucket(rate=qps)

    def iter_all_items(self,
                       fields: str = None,
                       page_size: int = 100,
                       inventory: bool = False,
                       max_pages: int = 500) -> Generator[Dict, None, None]:
        """
        生成器逐条yield商品item，自动翻页至结束
        """
        method = "taobao.items.inventory.get" if inventory else "taobao.items.onsale.get"
        fields = fields or (
            "num_iid,title,price,num,pic_url,outer_id,cid,"
            "approve_status,list_time,modified"
        )
        page_no = 1
        total = None

        while page_no <= max_pages:
            self.bucket.wait()   # ← QPS控制

            try:
                result = self.client.call(
                    method,
                    biz_params={
                        "page_no": page_no,
                        "page_size": min(page_size, 100),
                        "fields": fields
                    },
                    session=self.session
                )
            except Exception as e:
                # TOP限流 code=7 / SubMsg=ISP_FLOW_CONTROL
                if "FLOW_CONTROL" in str(e) or "code=7" in str(e):
                    print(f"⚠️  限流！等待3s后退页重试 p{page_no}")
                    time.sleep(3)
                    continue
                raise

            items = result.get("items", []) or []
            if total is None:
                total = int(result.get("total_results", 0))
                print(f"📦 店铺共有商品 {total} 件，开始全量同步...")

            for it in items:
                yield it

            # 翻页判断
            if not items or (page_no * page_size >= total):
                break
            page_no += 1

    # ───────── 便捷：一次返回列表 ─────────
    def fetch_all(self, **kwargs) -> List[Dict]:
        return list(self.iter_all_items(**kwargs))


# =========================================================
# 使用示例
# =========================================================
if __name__ == "__main__":
    syncer = TopShopItemsSyncer(
        app_key="YOUR_TOP_APP_KEY",
        app_secret="YOUR_TOP_APP_SECRET",
        session="YOUR_SELLER_ACCESS_TOKEN",  # ← 卖家OAuth授权所得
        sandbox=True,   # 生产切 False
        qps=5           # 淘宝默认约 5~10/s 免费，留余量
    )

    try:
        count = 0
        for item in syncer.iter_all_items(
                fields="num_iid,title,price,num,outer_id,approve_status",
                page_size=100
        ):
            count += 1
            if count <= 3:
                print(f"  • {item.get('num_iid')} {item.get('title')} ¥{item.get('price')} 库:{item.get('num')}")
            if count == 4:
                print("  ...")

        print(f"\n✅ 全量同步完成，共 {count} 件商品")

    except Exception as e:
        print("❌", e)

三、关键参数说明与避坑

▶ 分页正确姿势

# 终止条件（必须！）
if len(items) == 0 or page_no * page_size >= total_results:
    break

▶ fields 务必指定

▶ QPS 建议

▶ session 必传

四、返回关键字段映射（同步ERP用）

五、全量 + 增量同步建议

每日凌晨：
  ① 全量翻页（this script）→ upsert ERP商品表
     WHERE approve_status='onsale'

  ② 增量（可选）：调 taobao.items.onsale.get 按 modified 筛选近1天
     → 更新变动商品（减少翻页量）

六、一句话总结（面试版）

淘宝店铺全量商品用 taobao.items.onsale.get+ fields指定字段 + 按 total_results判断终止翻页；必须用卖家AccessToken调用；客户端加令牌桶QPS≤免费上限，遇限流(code=7)同页退避重试，避免触发封禁。