Python 异步框架 (Async/Aiohttp) 调用淘宝 API：实现万级商品数据异步采集

pip install aiohttp pycryptodome pandas

import hmac
import hashlib
import urllib.parse

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
    # 4. HMAC-SHA256签名并转为小写
    sign = hmac.new(
        app_secret.encode('utf-8'),
        sign_str.encode('utf-8'),
        hashlib.sha256
    ).hexdigest().lower()
    return sign

import asyncio
import aiohttp
import time
import random
from datetime import datetime
import pandas as pd
from typing import List, Dict

# -------------------------- 配置参数（需根据自身情况修改） --------------------------
APP_KEY = "你的App Key"          # 替换为你的App Key
APP_SECRET = "你的App Secret"    # 替换为你的App Secret
ACCESS_TOKEN = "你的Access Token"# 替换为你的Access Token（若接口需要）
KEYWORD = "手机"                 # 搜索关键词（可替换为其他商品类别，如“连衣裙”“笔记本电脑”）
PAGE_SIZE = 100                  # 每页商品数量（淘宝API最大支持100条/页）
TOTAL_PAGES = 100                # 总采集页数（100页 × 100条/页 = 10000条数据）
MAX_CONCURRENCY = 50             # 最大并发数（建议50-200，过高易被限流）
API_URL = "https://eco.taobao.com/router/rest"  # 淘宝API网关地址
# -----------------------------------------------------------------------------------

async def fetch_taobao_items(
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore,
    page_no: int
) -> List[Dict]:
    """
    异步采集单页商品数据
    :param session: aiohttp客户端会话（复用连接，提升效率）
    :param semaphore: 信号量（控制并发数）
    :param page_no: 当前页码
    :return: 单页商品数据列表（若失败返回空列表）
    """
    async with semaphore:  # 控制并发
        try:
            # 1. 构造公共参数（淘宝API必填）
            public_params = {
                "app_key": APP_KEY,
                "method": "taobao.item.search",  # 商品搜索接口
                "format": "json",               # 响应格式
                "v": "2.0",                     # API版本
                "sign_method": "hmac-sha256",   # 签名方式
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),  # 时间戳
                "access_token": ACCESS_TOKEN    # 若接口需要，否则可删除
            }

            # 2. 构造业务参数（商品搜索条件）
            business_params = {
                "q": KEYWORD,          # 搜索关键词
                "page_no": page_no,    # 当前页码
                "page_size": PAGE_SIZE # 每页数量
                # 可选参数：添加筛选条件（如价格区间、销量排序等）
                # "start_price": 1000,  # 最低价格
                # "end_price": 5000,    # 最高价格
                # "sort": "sales_desc"  # 按销量降序（默认按综合排序）
            }

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

            # 4. 发起异步GET请求（添加随机延迟，避免被限流）
            await asyncio.sleep(random.uniform(0.1, 0.5))  # 0.1-0.5秒随机延迟
            async with session.get(API_URL, params=all_params) as response:
                if response.status != 200:
                    print(f"页码{page_no}请求失败，HTTP状态码：{response.status}")
                    return []

                # 5. 解析响应数据
                result = await response.json()
                # 检查API返回是否成功（淘宝API错误码参考：https://open.taobao.com/doc.htm?docId=101370&docType=1）
                if "error_response" in result:
                    error_msg = result["error_response"]["msg"]
                    print(f"页码{page_no}API调用失败：{error_msg}")
                    return []

                # 6. 提取商品核心字段（根据需求调整字段）
                items = result["item_search_response"]["items"]["item"]
                item_list = []
                for item in items:
                    item_list.append({
                        "商品ID": item.get("num_iid", ""),
                        "商品标题": item.get("title", "").replace("<span class='H'>", "").replace("</span>", ""),  # 去除标题中的高亮标签
                        "售价": item.get("price", ""),
                        "销量": item.get("sales", ""),
                        "卖家昵称": item.get("nick", ""),
                        "店铺名称": item.get("shop_name", ""),
                        "商品链接": item.get("detail_url", ""),
                        "图片链接": item.get("pic_url", "")
                    })

                print(f"页码{page_no}采集成功，获取{len(item_list)}条商品数据")
                return item_list

        except Exception as e:
            print(f"页码{page_no}采集异常：{str(e)}")
            return []

async def main():
    """
    主函数：创建异步会话、调度协程、收集数据并存储
    """
    start_time = time.time()
    all_items = []  # 存储所有商品数据

    # 1. 创建aiohttp客户端会话（复用TCP连接，提升效率）
    timeout = aiohttp.ClientTimeout(total=30)  # 超时时间30秒
    async with aiohttp.ClientSession(timeout=timeout) as session:
        # 2. 创建信号量（控制并发数）
        semaphore = asyncio.Semaphore(MAX_CONCURRENCY)

        # 3. 生成所有页码的协程任务
        tasks = [
            fetch_taobao_items(session, semaphore, page_no)
            for page_no in range(1, TOTAL_PAGES + 1)
        ]

        # 4. 并发执行所有任务并收集结果
        results = await asyncio.gather(*tasks)

        # 5. 合并所有页的商品数据
        for page_items in results:
            if page_items:
                all_items.extend(page_items)

    # 6. 数据存储（保存为Excel文件）
    if all_items:
        df = pd.DataFrame(all_items)
        output_file = f"淘宝商品数据_{KEYWORD}_{datetime.now().strftime('%Y%m%d%H%M%S')}.xlsx"
        df.to_excel(output_file, index=False, engine="openpyxl")
        print(f"\n采集完成！共获取{len(all_items)}条商品数据，已保存至：{output_file}")
    else:
        print("\n采集失败，未获取到任何商品数据")

    # 7. 输出采集耗时
    end_time = time.time()
    print(f"总耗时：{round(end_time - start_time, 2)}秒")

if __name__ == "__main__":
    # 运行异步主函数（Python 3.7+支持asyncio.run()）
    asyncio.run(main())