淘宝关键词搜索 API 批量搜索优化：分页策略、并发控制与频率适配

python
total_pages = min(math.ceil(total_results / page_size), max_pages)  # 限制最大查询页数

python
def search_all_items(keyword, max_pages=10):
    all_items = []
    first_page = search_items(keyword, page_no=1)  # 假设search_items为单页查询函数
    total_results = first_page["items_search_response"]["total_results"]
    total_pages = min(math.ceil(total_results / 20), max_pages)  # 每页20条
    for page in range(1, total_pages + 1):
        result = search_items(keyword, page_no=page)
        all_items.extend(result["items_search_response"]["items"]["item"])
        time.sleep(1)  # 避免触发限流
    return all_items

python
async def fetch_items(keywords, max_concurrent=10):
    semaphore = asyncio.Semaphore(max_concurrent)  # 限制最大并发数
    async def fetch_single(keyword):
        async with semaphore:
            params = {"q": keyword, "page_no": 1, "page_size": 20}
            params["sign"] = generate_sign(params, app_secret)  # 生成签名
            async with session.get(API_URL, params=params) as resp:
                return await resp.json()
    tasks = [fetch_single(kw) for kw in keywords]
    return await asyncio.gather(*tasks)

python
import asyncio
import aiohttp
import hashlib
import time
import math
from queue import Queue
from threading import Thread
class TaobaoAPIClient:
    def __init__(self, app_key, app_secret, max_concurrent=10):
        self.app_key = app_key
        self.app_secret = app_secret
        self.max_concurrent = max_concurrent
        self.session = aiohttp.ClientSession()
    def generate_sign(self, params):
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = self.app_secret + ''.join([f"{k}{v}" for k, v in sorted_params if k != 'sign']) + self.app_secret
        return hashlib.md5(sign_str.encode()).hexdigest().upper()
    async def fetch_page(self, keyword, page_no, page_size=20):
        params = {
            "app_key": self.app_key,
            "method": "taobao.items.search",
            "format": "json",
            "v": "2.0",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "sign_method": "md5",
            "q": keyword,
            "page_no": page_no,
            "page_size": page_size,
            "fields": "num_iid,title,price,pic_url,sales"
        }
        params["sign"] = self.generate_sign(params)
        async with self.session.get("https://eco.taobao.com/router/rest", params=params) as resp:
            return await resp.json()
    async def fetch_all_pages(self, keyword, max_pages=10):
        first_page = await self.fetch_page(keyword, page_no=1)
        total_results = first_page["items_search_response"]["total_results"]
        total_pages = min(math.ceil(total_results / 20), max_pages)
        tasks = [self.fetch_page(keyword, page) for page in range(1, total_pages + 1)]
        results = await asyncio.gather(*tasks)
        all_items = []
        for result in results:
            all_items.extend(result["items_search_response"]["items"]["item"])
        return all_items
    async def close(self):
        await self.session.close()
# 使用示例
async def main():
    client = TaobaoAPIClient(app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET")
    items = await client.fetch_all_pages("手机")
    print(f"获取到 {len(items)} 条商品数据")
    await client.close()
asyncio.run(main())