Step-by-Step: 接入淘宝商品详情 API 并解析返回数据

pip install aiohttp pycryptodome beautifulsoup4 pandas

import hmac
import hashlib

def generate_taobao_sign(params: dict, app_secret: str) -> str:
    """
    生成淘宝API请求签名（HMAC-SHA256）
    :param params: 所有请求参数（公共参数+业务参数）
    :param app_secret: 应用的App Secret
    :return: 小写签名字符串
    """
    # 1. 按参数名ASCII升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 2. 拼接为"key=value"格式字符串（无分隔符）
    canonical_str = ''.join([f"{k}{v}" for k, v in sorted_params])
    # 3. 前后拼接App Secret后签名
    sign_str = app_secret + canonical_str + app_secret
    sign = hmac.new(
        app_secret.encode('utf-8'),
        sign_str.encode('utf-8'),
        hashlib.sha256
    ).hexdigest().lower()
    return sign

from datetime import datetime

def build_item_detail_params(
    app_key: str,
    app_secret: str,
    access_token: str,
    num_iid: str,
    is_promotion: bool = False
) -> dict:
    """
    构造商品详情API的请求参数（含签名）
    :param app_key: App Key
    :param app_secret: App Secret
    :param access_token: Access Token
    :param num_iid: 商品ID（num_iid）
    :param is_promotion: 是否获取促销价
    :return: 完整请求参数（含sign）
    """
    # 1. 构造公共参数
    public_params = {
        "app_key": app_key,
        "method": "taobao.item_get",
        "format": "json",
        "v": "2.0",
        "sign_method": "hmac-sha256",
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "access_token": access_token
    }

    # 2. 构造业务参数
    business_params = {
        "num_iid": num_iid,
        "is_promotion": str(is_promotion).lower()  # 转为小写字符串（API要求）
    }

    # 3. 合并参数并生成签名
    all_params = {**public_params, **business_params}
    all_params["sign"] = generate_taobao_sign(all_params, app_secret)

    return all_params

import asyncio
import aiohttp
from bs4 import BeautifulSoup
from typing import Dict, Optional

async def fetch_item_detail(
    session: aiohttp.ClientSession,
    app_key: str,
    app_secret: str,
    access_token: str,
    num_iid: str
) -> Optional[Dict]:
    """
    异步请求商品详情API并解析核心数据
    :param session: Aiohttp会话（复用连接）
    :param app_key: App Key
    :param app_secret: App Secret
    :param access_token: Access Token
    :param num_iid: 商品ID
    :return: 解析后的商品详情字典（失败返回None）
    """
    # 1. 构造请求参数
    params = build_item_detail_params(
        app_key=app_key,
        app_secret=app_secret,
        access_token=access_token,
        num_iid=num_iid,
        is_promotion=True  # 获取促销价
    )

    # 2. 发起异步请求（添加随机延迟，避免限流）
    api_url = "https://eco.taobao.com/router/rest"  # 淘宝API网关
    try:
        await asyncio.sleep(asyncio.sleep(random.uniform(0.3, 0.8)))  # 0.3-0.8秒延迟
        async with session.get(api_url, params=params, timeout=10) as response:
            # 3. 检查HTTP响应状态
            if response.status != 200:
                print(f"商品{num_iid}请求失败，HTTP状态码：{response.status}")
                return None

            # 4. 解析JSON响应
            result = await response.json()

            # 5. 检查API错误（如权限不足、商品不存在）
            if "error_response" in result:
                error_code = result["error_response"]["code"]
                error_msg = result["error_response"]["msg"]
                print(f"商品{num_iid}API调用失败：[{error_code}] {error_msg}")
                return None

            # 6. 提取核心详情数据（从"item_get_response.item"字段获取）
            raw_item = result["item_get_response"]["item"]
            parsed_item = parse_raw_item_data(raw_item)  # 后续实现解析函数
            print(f"商品{num_iid}详情解析成功，获取{len(parsed_item)}个核心字段")
            return parsed_item

    except Exception as e:
        print(f"商品{num_iid}请求异常：{str(e)}")
        return None

def parse_raw_item_data(raw_item: Dict) -> Dict:
    """
    解析商品详情原始数据，提取核心字段
    :param raw_item: API返回的原始商品详情字典
    :return: 结构化的商品详情字典
    """
    # 1. 基础信息（标题、价格、库存等）
    base_info = {
        "商品ID": raw_item.get("num_iid", ""),
        "商品标题": raw_item.get("title", "").replace("<span class='H'>", "").replace("</span>", ""),
        "原价": raw_item.get("price", ""),
        "促销价": raw_item.get("promotion_price", raw_item.get("price", "")),  # 无促销价则用原价
        "库存数量": raw_item.get("total_sold", ""),  # 已售数量
        "库存状态": "有货" if int(raw_item.get("stock", 0)) > 0 else "无货",
        "商品链接": raw_item.get("detail_url", ""),
        "主图链接": raw_item.get("pic_url", "")
    }

    # 2. 店铺信息
    shop_info = {
        "店铺名称": raw_item.get("seller", {}).get("shop_name", ""),
        "店铺类型": raw_item.get("seller", {}).get("shop_type", ""),  # 天猫/淘宝
        "卖家昵称": raw_item.get("seller", {}).get("nick", ""),
        "卖家ID": raw_item.get("seller", {}).get("user_id", "")
    }

    # 3. 规格信息（如颜色、尺寸，嵌套在"sku"字段中）
    sku_info = []
    skus = raw_item.get("sku", {}).get("sku_list", [])  # 规格列表（可能为空）
    for sku in skus:
        sku_info.append({
            "规格ID": sku.get("sku_id", ""),
            "规格名称": sku.get("sku_name", ""),  # 如"颜色:黑色;尺寸:XL"
            "规格价格": sku.get("price", ""),
            "规格库存": sku.get("stock", 0)
        })

    # 4. 详情描述图片（从HTML中提取，raw_item["desc"]是HTML字符串）
    desc_html = raw_item.get("desc", "")
    soup = BeautifulSoup(desc_html, "html.parser")
    desc_images = [img["src"] for img in soup.find_all("img") if "src" in img.attrs]

    # 5. 售后信息
    after_sale_info = {
        "支持7天无理由": raw_item.get("freight", {}).get("support_7d_return", "未知"),
        "发货地": raw_item.get("freight", {}).get("from", "未知"),
        "运费": raw_item.get("freight", {}).get("express_fee", "未知")
    }

    # 合并所有字段，返回结构化数据
    return {
        **base_info,
        **shop_info,
        "规格列表": sku_info,
        "详情图片列表": desc_images,
        **after_sale_info
    }

import time
import random
import pandas as pd
from typing import List

async def batch_fetch_item_details(
    app_key: str,
    app_secret: str,
    access_token: str,
    num_iid_list: List[str],
    max_concurrency: int = 30
) -> List[Dict]:
    """
    批量采集商品详情
    :param app_key: App Key
    :param app_secret: App Secret
    :param access_token: Access Token
    :param num_iid_list: 商品ID列表（如["123456789", "987654321"]）
    :param max_concurrency: 最大并发数（建议30-50，避免限流）
    :return: 解析后的商品详情列表
    """
    start_time = time.time()
    all_details = []

    # 1. 创建Aiohttp会话（复用TCP连接）
    timeout = aiohttp.ClientTimeout(total=30)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        # 2. 创建信号量（控制并发）
        semaphore = asyncio.Semaphore(max_concurrency)

        # 3. 生成所有异步任务（绑定信号量）
        tasks = []
        for num_iid in num_iid_list:
            # 绑定信号量到任务（避免直接在函数内创建，便于复用）
            task = asyncio.create_task(
                fetch_item_detail(
                    session=session,
                    app_key=app</doubaocanvas>