淘宝店铺所有商品 API 接口全解析：批量获取全店商品数据（2026 最新版）

https://oauth.taobao.com/authorize?response_type=code&client_id=你的AppKey&redirect_uri=你的回调地址&view=web

pip install requests

import requests
import time
import json
import hashlib
class TaobaoShopClient:
    def __init__(self, app_key, app_secret, access_token):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.api_url = "https://eco.taobao.com/router/rest"  # 淘宝开放平台网关
    def generate_sign(self, params):
        """生成淘宝API签名（MD5加密）"""
        # 1. 参数按key字典排序
        sorted_params = sorted(params.items())
        # 2. 拼接参数串
        param_str = "".join([f"{k}{v}" for k, v in sorted_params])
        # 3. 前后加AppSecret，MD5加密
        sign_str = f"{self.app_secret}{param_str}{self.app_secret}"
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    def request(self, method, params):
        """通用API请求方法"""
        # 公共参数
        public_params = {
            "method": method,
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "access_token": self.access_token
        }
        # 合并公共参数与业务参数
        all_params = {**public_params, **params}
        # 生成签名
        all_params["sign"] = self.generate_sign(all_params)
        try:
            # 发送POST请求
            resp = requests.post(self.api_url, data=all_params, timeout=15)
            resp.raise_for_status()  # 抛出HTTP错误
            return resp.json()
        except Exception as e:
            return {"error": f"请求失败：{str(e)}"}
    def get_shop_all_items(self, shop_nick, page_no=1, page_size=200, status="onsale"):
        """获取淘宝店铺所有商品（分页）"""
        params = {
            "nick": shop_nick,  # 店铺账号昵称
            "page_no": page_no,
            "page_size": page_size,
            "status": status,  # 在售商品：onsale，全部：all
            "fields": "num_iid,title,price,sku,stock,volume,category_id,pic_url,publish_time"  # 指定返回字段
        }
        return self.request("taobao.shop.get.items", params)
    def parse_shop_items(self, json_data):
        """解析店铺商品数据（结构化）"""
        try:
            # 淘宝API返回结构：第一层为接口方法名+_response
            items_response = json_data.get("taobao_shop_get_items_response", {})
            result = items_response.get("shop_item_get_response", {})
            total_results = result.get("total_results", 0)  # 商品总数
            items = result.get("items", {}).get("item", [])  # 商品列表
            # 结构化解析
            parsed_items = []
            for item in items:
                parsed_items.append({
                    "商品ID": item.get("num_iid"),
                    "商品标题": item.get("title"),
                    "售价": item.get("price"),
                    "库存": item.get("stock", 0),
                    "销量": item.get("volume", 0),
                    "类目ID": item.get("category_id"),
                    "主图": item.get("pic_url"),
                    "发布时间": item.get("publish_time"),
                    "SKU": item.get("sku", "")  # SKU详情需单独解析
                })
            return {
                "total": total_results,
                "items": parsed_items
            }
        except Exception as e:
            return {"error": f"解析失败：{str(e)}"}
# ===================== 配置信息（替换为自己的） =====================
APP_KEY = "你的淘宝开放平台AppKey"
APP_SECRET = "你的淘宝开放平台AppSecret"
ACCESS_TOKEN = "店铺授权获取的access_token"
SHOP_NICK = "目标店铺账号昵称"  # 如：xx旗舰店
# 初始化客户端
client = TaobaoShopClient(APP_KEY, APP_SECRET, ACCESS_TOKEN)
# ===================== 批量获取全店商品 =====================
if __name__ == "__main__":
    all_items = []
    total_items = 0
    page = 1
    # 循环获取所有分页数据
    while True:
        print(f"正在获取第 {page} 页商品...")
        # 获取当前页商品
        response = client.get_shop_all_items(SHOP_NICK, page_no=page, page_size=200)
        # 解析数据
        parsed_data = client.parse_shop_items(response)
        if "error" in parsed_data:
            print(f"获取失败：{parsed_data['error']}")
            break
        # 累加商品数据
        items = parsed_data["items"]
        if not items:
            break
        all_items.extend(items)
        total_items = parsed_data["total"]
        # 分页终止条件：当前页商品数 < 每页条数，或已获取全部
        if len(items) < 200:
            break
        page += 1
        time.sleep(1)  # 限流，避免触发平台风控
    # 输出结果
    print(f"共获取 {total_items} 个商品，实际解析 {len(all_items)} 个")
    # 保存至JSON文件
    with open("taobao_shop_items.json", "w", encoding="utf-8") as f:
        json.dump(all_items, f, ensure_ascii=False, indent=2)
    print("商品数据已保存至 taobao_shop_items.json")

def get_shop_items_by_search(shop_nick, page_no=1, page_size=100):
    """通过搜索接口获取店铺商品（非商家授权）"""
    params = {
        "q": shop_nick,  # 店铺昵称作为关键词
        "page_no": page_no,
        "page_size": page_size,
        "fields": "num_iid,title,price,volume,pic_url"
    }
    return client.request("taobao.items.search", params)
# 调用示例
search_response = get_shop_items_by_search("xx旗舰店")
print(search_response)