淘宝开放平台店铺全量商品接口实战：分类筛选+防风控+批量导出（附Python完整代码）

import requests
import hashlib
import time
import json
import csv

class TaoBaoShopProductsAPI:
    """淘宝店铺全量商品接口客户端（生产级，分类筛选+批量导出）"""
    def __init__(self, app_key, app_secret, session_key, nick):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session_key = session_key  # 店铺授权session_key
        self.nick = nick  # 店铺昵称（需与授权账号一致）
        self.api_url = "https://eco.taobao.com/router/rest"
        self.timeout = 12
        self.retry_count = 2
        self.qps_limit = 5  # 适配平台QPS限制
        self.last_request_time = 0

    def _generate_sign(self, params):
        """淘宝TOP API标准MD5签名（网上教程常缺失参数排序与编码逻辑）"""
        # 1. 按key ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接参数（key+value），无分隔符
        param_str = "".join([f"{k}{v}" for k, v in sorted_params])
        # 3. 首尾拼接app_secret，MD5加密后转大写
        sign_raw = f"{self.app_secret}{param_str}{self.app_secret}"
        return hashlib.md5(sign_raw.encode("utf-8")).hexdigest().upper()

    def _control_qps(self):
        """QPS限流控制，避免触发平台风控"""
        current_time = time.time()
        interval = 1 / self.qps_limit
        if current_time - self.last_request_time < interval:
            time.sleep(interval - (current_time - self.last_request_time))
        self.last_request_time = current_time

    def get_shop_products(self, cid=None, status="onsale", page_size=20, export_csv=False):
        """
        获取店铺全量商品（支持分类筛选、状态过滤、批量导出）
        :param cid: 商品分类ID（可选，筛选指定分类商品）
        :param status: 商品状态（onsale-在售，instock-库存中，outstock-下架）
        :param page_size: 每页数量（最大20）
        :param export_csv: 是否导出为CSV文件
        :return: 全量商品结构化数据
        """
        page_no = 1
        all_products = []
        while True:
            # 1. 构造请求参数
            params = {
                "method": "taobao.items.onsale.get",
                "app_key": self.app_key,
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),  # 官方要求格式
                "format": "json",
                "v": "2.0",
                "sign_method": "md5",
                "session": self.session_key,
                "nick": self.nick,
                "page_no": page_no,
                "page_size": page_size,
                "status": status
            }
            # 可选：分类筛选
            if cid:
                params["cid"] = cid

            # 2. 生成签名
            params["sign"] = self._generate_sign(params)

            # 3. 防风控+重试机制
            for attempt in range(self.retry_count + 1):
                try:
                    self._control_qps()
                    response = requests.post(
                        self.api_url,
                        data=params,
                        timeout=self.timeout
                    )
                    result = response.json()

                    # 业务错误判断
                    if "error_response" in result:
                        err_msg = result["error_response"].get("sub_msg", "接口调用失败")
                        return {"code": -1, "msg": err_msg}

                    # 解析商品数据
                    data = result.get("items_onsale_get_response", {})
                    products = data.get("items", {}).get("item", [])
                    if not products:
                        break  # 无更多商品，终止分页

                    # 结构化解析，剔除冗余字段
                    for product in products:
                        all_products.append({
                            "num_iid": product.get("num_iid"),  # 商品ID
                            "title": product.get("title"),
                            "price": product.get("price"),
                            "stock": product.get("num"),
                            "cid": product.get("cid"),  # 分类ID
                            "category_name": product.get("category_name"),
                            "pic_url": product.get("pic_url"),
                            "status": product.get("status"),
                            "created": product.get("created")  # 上架时间
                        })

                    # 分页判断
                    total_page = data.get("total_results", 0) // page_size + 1
                    if page_no >= total_page:
                        break
                    page_no += 1
                    break
                except Exception as e:
                    if attempt == self.retry_count:
                        return {"code": 500, "msg": f"请求异常：{str(e)}"}
                    # 限流重试，延长间隔
                    if "403" in str(e):
                        time.sleep(2 ** attempt)
                    else:
                        time.sleep(1)

        # 可选：导出CSV
        if export_csv and all_products:
            with open("shop_products.csv", "w", encoding="utf-8-sig", newline="") as f:
                writer = csv.DictWriter(f, fieldnames=all_products[0].keys())
                writer.writeheader()
                writer.writerows(all_products)

        return {
            "code": 200,
            "msg": "success",
            "total": len(all_products),
            "products": all_products,
            "exported": export_csv
        }

# 调用示例
if __name__ == "__main__":
    # 替换为淘宝开放平台获取的真实信息
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    SESSION_KEY = "your_session_key"  # 店铺授权后获取
    SHOP_NICK = "your_shop_nick"  # 店铺昵称

    # 初始化客户端
    api = TaoBaoShopProductsAPI(APP_KEY, APP_SECRET, SESSION_KEY, SHOP_NICK)

    # 调用示例：获取指定分类、在售商品，并导出CSV
    result = api.get_shop_products(
        cid="123456",  # 替换为实际分类ID
        status="onsale",
        page_size=20,
        export_csv=True
    )

    print(json.dumps(result, ensure_ascii=False, indent=2))