高效调用淘宝商品详情 API：异步处理、批量化与性能调优实战

pip install aiohttp ratelimit requests

import asyncio
import aiohttp
import time
from ratelimit import limits, sleep_and_retry
import hashlib
import json

# 淘宝API配置（替换为自己的真实信息）
TAOBAO_APP_KEY = "your_app_key"
TAOBAO_APP_SECRET = "your_app_secret"
TAOBAO_API_URL = "https://gw.api.taobao.com/router/rest"

# 限流配置：QPS限制为5（根据平台实际限流规则调整）
QPS_LIMIT = 5
CALLS = QPS_LIMIT
PERIOD = 1

def generate_sign(params, app_secret):
    """
    生成淘宝API请求签名（简化版，实际需严格按淘宝签名规则）
    """
    # 按参数名升序排列
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 拼接字符串
    sign_str = app_secret + "".join([f"{k}{v}" for k, v in sorted_params]) + app_secret
    # MD5加密并转大写
    sign = hashlib.md5(sign_str.encode()).hexdigest().upper()
    return sign

def build_request_params(item_id):
    """
    构建淘宝商品详情API的请求参数
    """
    params = {
        "method": "taobao.item.get",
        "app_key": TAOBAO_APP_KEY,
        "format": "json",
        "v": "2.0",
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        "item_id": item_id,
        "fields": "num_iid,title,price,sales,stock,pic_url"  # 需要查询的字段
    }
    # 生成签名
    params["sign"] = generate_sign(params, TAOBAO_APP_SECRET)
    return params

import requests

def sync_get_item_detail(item_id):
    """
    同步调用淘宝商品详情API
    """
    try:
        params = build_request_params(item_id)
        response = requests.get(TAOBAO_API_URL, params=params, timeout=10)
        response.raise_for_status()  # 抛出HTTP异常
        return response.json()
    except Exception as e:
        print(f"同步查询商品{item_id}失败：{str(e)}")
        return None

def sync_batch_query(item_ids):
    """
    同步批量查询商品详情
    """
    start_time = time.time()
    results = []
    for item_id in item_ids:
        result = sync_get_item_detail(item_id)
        if result:
            results.append(result)
    end_time = time.time()
    print(f"同步调用完成，共查询{len(results)}个商品，耗时：{end_time - start_time:.2f}秒")
    return results

@sleep_and_retry
@limits(calls=CALLS, period=PERIOD)
async def async_get_item_detail(session, item_id):
    """
    异步调用淘宝商品详情API（带限流）
    """
    try:
        params = build_request_params(item_id)
        async with session.get(TAOBAO_API_URL, params=params, timeout=10) as response:
            response.raise_for_status()
            result = await response.json()
            return {"item_id": item_id, "data": result}
    except Exception as e:
        print(f"异步查询商品{item_id}失败：{str(e)}")
        return {"item_id": item_id, "data": None}

async def async_batch_query(item_ids):
    """
    异步批量查询商品详情
    """
    start_time = time.time()
    # 创建异步HTTP会话（复用连接，提升性能）
    async with aiohttp.ClientSession() as session:
        # 创建任务列表
        tasks = [async_get_item_detail(session, item_id) for item_id in item_ids]
        # 并发执行任务
        results = await asyncio.gather(*tasks)
        # 过滤掉查询失败的结果
        valid_results = [res for res in results if res["data"] is not None]
    
    end_time = time.time()
    print(f"异步调用完成，共查询{len(valid_results)}个商品，耗时：{end_time - start_time:.2f}秒")
    return valid_results

async def batch_async_query(item_ids, batch_size=50):
    """
    分批次异步查询（最终优化版）
    :param item_ids: 商品ID列表
    :param batch_size: 每批次查询的商品数量
    """
    start_time = time.time()
    all_results = []
    
    # 拆分批次
    for i in range(0, len(item_ids), batch_size):
        batch_item_ids = item_ids[i:i+batch_size]
        print(f"正在处理第{i//batch_size + 1}批次，共{len(batch_item_ids)}个商品")
        # 执行当前批次异步查询
        batch_results = await async_batch_query(batch_item_ids)
        all_results.extend(batch_results)
        # 可选：批次间短暂休眠，进一步降低限流风险
        await asyncio.sleep(0.5)
    
    end_time = time.time()
    print(f"分批次异步调用完成，总计查询{len(all_results)}个商品，总耗时：{end_time - start_time:.2f}秒")
    return all_results

if __name__ == "__main__":
    # 测试用商品ID列表（替换为真实ID）
    test_item_ids = ["123456789", "987654321", "1122334455", "5544332211"] * 50  # 共200个商品ID
    
    # 1. 同步调用测试
    print("===== 同步调用开始 =====")
    sync_results = sync_batch_query(test_item_ids)
    
    # 2. 异步调用测试
    print("\n===== 异步调用开始 =====")
    async_results = asyncio.run(batch_async_query(test_item_ids, batch_size=50))