淘宝商品详情接口（item_get）企业级全解析：参数配置、签名机制与 Python 代码实战

import requests
import hashlib
import time
import json
from threading import Lock
from datetime import datetime
class TaobaoItemDetailAPI:
    """淘宝商品详情接口企业级客户端（支持签名、重试、结构化解析）"""
    def __init__(self, app_key, app_secret, timeout=10, max_retries=3, request_interval=1):
        """
        初始化客户端
        :param app_key: 开放平台app_key
        :param app_secret: 开放平台app_secret（需妥善保管）
        :param timeout: 请求超时时间（秒）
        :param max_retries: 失败重试次数
        :param request_interval: 请求间隔（秒，控制QPS）
        """
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "https://eco.taobao.com/router/rest"
        self.timeout = timeout
        self.max_retries = max_retries
        self.request_interval = request_interval
        self.last_request_time = 0
        self.request_lock = Lock()  # 线程安全控制
        self.session = self._init_session()  # 初始化请求会话
    def _init_session(self):
        """初始化会话，配置自动重试"""
        session = requests.Session()
        retry_adapter = requests.adapters.HTTPAdapter(
            max_retries=requests.packages.urllib3.util.retry.Retry(
                total=self.max_retries,
                status_forcelist=[429, 500, 502, 503, 504],
                backoff_factor=0.5
            )
        )
        session.mount("https://", retry_adapter)
        return session
    def _generate_sign(self, params):
        """生成MD5签名（严格遵循淘宝规范）"""
        # 1. 参数ASCII升序排序
        sorted_items = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接字符串
        sign_str = "&".join([f"{k}={v}" for k, v in sorted_items])
        # 3. 首尾加app_secret
        sign_str = f"{self.app_secret}{sign_str}{self.app_secret}"
        # 4. MD5加密转大写
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    def _validate_fields(self, fields):
        """过滤无效字段，避免接口报错"""
        supported_fields = [
            "title", "price", "stock", "item_id", "seller_id", "shop_name",
            "main_image", "detail_images", "skus", "category", "brand"
        ]
        if not fields:
            return ",".join(supported_fields)
        return ",".join([f for f in fields.split(",") if f.strip() in supported_fields])
    def _control_request_interval(self):
        """控制请求频率，避免超限（建议QPS≤5）"""
        with self.request_lock:
            current_time = time.time()
            if current_time - self.last_request_time < self.request_interval:
                time.sleep(self.request_interval - (current_time - self.last_request_time))
            self.last_request_time = current_time
    def _parse_item_data(self, raw_data):
        """结构化解析商品数据"""
        if not raw_data or "item_get_response" not in raw_data:
            return None
        item = raw_data["item_get_response"].get("item", {})
        # 基础信息
        base_info = {
            "item_id": item.get("num_iid", ""),
            "title": item.get("title", ""),
            "create_time": item.get("created", ""),
            "update_time": item.get("modified", "")
        }
        # 价格信息
        price_info = {
            "current_price": float(item.get("price", 0)),
            "original_price": float(item.get("original_price", 0)),
            "promotion_price": float(item.get("promotion_price", 0))
        }
        # 库存与销量
        inventory = {
            "total_stock": int(item.get("stock", 0)),
            "sales_count": int(item.get("sales", 0)),
            "skus": self._parse_skus(item.get("skus", {}))
        }
        # 图片信息
        images = {
            "main_images": item.get("pic_urls", []),
            "detail_images": self._extract_detail_images(item.get("desc", "")),
            "sku_images": item.get("sku_pics", {})
        }
        # 其他核心模块
        return {
            "base_info": base_info,
            "price_info": price_info,
            "inventory": inventory,
            "images": images,
            "category_brand": {
                "category": item.get("category", ""),
                "brand": item.get("brand", "")
            },
            "seller_info": {
                "seller_id": item.get("seller_id", ""),
                "shop_name": item.get("shop_name", "")
            },
            "specifications": item.get("specs", []),
            "parse_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }
    def _parse_skus(self, sku_data):
        """解析SKU规格与库存"""
        skus = []
        for sku in sku_data.get("sku", []):
            skus.append({
                "sku_id": sku.get("sku_id", ""),
                "specs": sku.get("specs", ""),
                "price": float(sku.get("price", 0)),
                "stock": int(sku.get("stock", 0))
            })
        return skus
    def _extract_detail_images(self, desc_html):
        """从HTML描述中提取详情图"""
        import re
        return re.findall(r'src="(https?://[^"]+\.jpg|https?://[^"]+\.png)"', desc_html)
    def get_item_detail(self, item_id, fields=None):
        """
        核心方法：获取商品详情
        :param item_id: 商品ID
        :param fields: 需返回的字段（逗号分隔）
        :return: 结构化商品数据（None表示失败）
        """
        # 1. 字段验证
        valid_fields = self._validate_fields(fields)
        # 2. 构建基础参数
        base_params = {
            "app_key": self.app_key,
            "method": "taobao.item_get",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "item_id": item_id,
            "fields": valid_fields
        }
        # 3. 生成签名
        base_params["sign"] = self._generate_sign(base_params)
        # 4. 控制请求频率
        self._control_request_interval()
        # 5. 发送请求（带错误处理）
        retry_count = 0
        while retry_count < self.max_retries:
            try:
                response = self.session.post(
                    url=self.base_url,
                    data=base_params,
                    timeout=self.timeout,
                    headers={"Content-Type": "application/x-www-form-urlencoded"}
                )
                response.raise_for_status()  # 捕获4xx/5xx错误
                # 解析响应
                raw_result = response.json()
                if "error_response" in raw_result:
                    error_msg = raw_result["error_response"].get("msg", "未知错误")
                    print(f"接口报错：{error_msg}（code:{raw_result['error_response']['code']}）")
                    # 签名/参数错误无需重试
                    if raw_result["error_response"]["code"] in [15, 16]:
                        return None
                    retry_count += 1
                    time.sleep(1)
                    continue
                # 结构化解析
                return self._parse_item_data(raw_result)
            except requests.exceptions.RequestException as e:
                print(f"网络异常：{str(e)}")
                retry_count += 1
                time.sleep(1)
            except json.JSONDecodeError:
                print("响应格式错误，无法解析JSON")
                retry_count += 1
                time.sleep(1)
        print(f"超过{self.max_retries}次重试，获取失败")
        return None

def single_item_demo():
    # 替换为自身的app_key和app_secret
    APP_KEY = "your_taobao_appkey"
    APP_SECRET = "your_taobao_appsecret"
    TARGET_ITEM_ID = "123456789"  # 目标商品ID
    # 初始化客户端
    api_client = TaobaoItemDetailAPI(
        app_key=APP_KEY,
        app_secret=APP_SECRET,
        request_interval=0.3  # QPS≈3
    )
    # 获取详情
    item_detail = api_client.get_item_detail(
        item_id=TARGET_ITEM_ID,
        fields="title,price,stock,main_images,shop_name"  # 指定字段
    )
    # 打印结果
    if item_detail:
        print(f"商品名称：{item_detail['base_info']['title']}")
        print(f"售价：¥{item_detail['price_info']['current_price']}")
        print(f"库存：{item_detail['inventory']['total_stock']}件")
        print(f"店铺：{item_detail['seller_info']['shop_name']}")
if __name__ == "__main__":
    single_item_demo()

from concurrent.futures import ThreadPoolExecutor, as_completed
def batch_item_demo():
    APP_KEY = "your_taobao_appkey"
    APP_SECRET = "your_taobao_appsecret"
    BATCH_ITEM_IDS = ["123456789", "987654321", "112233445"]  # 批量商品ID
    MAX_WORKERS = 5  # 并发线程数≤5
    api_client = TaobaoItemDetailAPI(APP_KEY, APP_SECRET)
    results = {}
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_tasks = {
            executor.submit(api_client.get_item_detail, item_id): item_id
            for item_id in BATCH_ITEM_IDS
        }
        for future in as_completed(future_tasks):
            item_id = future_tasks[future]
            try:
                detail = future.result()
                results[item_id] = "成功" if detail else "失败"
            except Exception as e:
                results[item_id] = f"异常：{str(e)}"
    # 输出统计
    print(f"批量结果：成功{list(results.values()).count('成功')}个，失败{list(results.values()).count('失败')}个")