淘宝开放平台获取店铺全量商品列表接口实战：自动分页 + 去重 + 风控兼容（附 Python 生产级代码）

import requests
import hashlib
import time

class TaoBaoShopAllItemsAPI:
    """淘宝获取店铺所有商品列表接口（官方API·生产级封装）"""
    def __init__(self, app_key, app_secret, session_key):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session_key = session_key
        self.api_url = "https://eco.taobao.com/router/rest"
        self.item_id_set = set()  # 自动去重

    def _make_sign(self, params):
        """淘宝标准MD5签名（全网最标准实现）"""
        sorted_items = sorted(params.items())
        sign_str = self.app_secret + "".join(f"{k}{v}" for k, v in sorted_items) + self.app_secret
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()

    def get_all_shop_items(self, max_page=50):
        """获取店铺所有商品（自动分页+去重+防风控）"""
        page = 1
        all_items = []

        while page <= max_page:
            # 公共参数
            params = {
                "method": "taobao.items.list.get",
                "app_key": self.app_key,
                "session": self.session_key,
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
                "format": "json",
                "v": "2.0",
                "sign_method": "md5",
                "page_no": page,
                "page_size": 20,
                "fields": "num_iid,title,price,num,pic_url,created"
            }

            # 签名
            params["sign"] = self._make_sign(params)

            try:
                # 请求
                resp = requests.post(self.api_url, data=params, timeout=10)
                result = resp.json()

                # 错误判断
                if "error_response" in result:
                    err = result["error_response"]
                    return {"code": -1, "msg": err.get("sub_msg", "接口异常")}

                # 解析商品
                data = result.get("items_list_get_response", {})
                items = data.get("items", {}).get("item", [])

                if not items:
                    break  # 无商品，结束分页

                # 去重 + 结构化
                for item in items:
                    num_iid = item.get("num_iid")
                    if num_iid not in self.item_id_set:
                        self.item_id_set.add(num_iid)
                        all_items.append({
                            "num_iid": num_iid,
                            "title": item.get("title"),
                            "price": item.get("price"),
                            "stock": item.get("num"),
                            "pic_url": item.get("pic_url"),
                            "created": item.get("created")
                        })

                page += 1
                time.sleep(1)  # 风控间隔

            except Exception as e:
                return {"code": 500, "msg": f"请求异常：{str(e)}"}

        return {
            "code": 200,
            "msg": "success",
            "total": len(all_items),
            "items": all_items
        }

# ———— 调用示例 ————
if __name__ == "__main__":
    # 替换为开放平台密钥
    APP_KEY = "你的app_key"
    APP_SECRET = "你的app_secret"
    SESSION_KEY = "你的session_key"

    api = TaoBaoShopAllItemsAPI(APP_KEY, APP_SECRET, SESSION_KEY)
    result = api.get_all_shop_items(max_page=50)
    print(f"店铺商品总数：{result['total']}")