高性能采集方案：淘宝商品 API 的并发调用与数据实时处理

pip install aiohttp python-dotenv pymongo  # 异步HTTP/环境变量/数据库

import asyncio
import aiohttp
import hashlib
import time
import json
from datetime import datetime
from typing import List, Dict, Optional
from dotenv import load_dotenv
import os
from pymongo import MongoClient
from queue import Queue
from threading import Thread

# 加载环境变量（安全存储密钥）
load_dotenv()
APP_KEY = os.getenv("TAOBAO_APP_KEY")
APP_SECRET = os.getenv("TAOBAO_APP_SECRET")
MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017/")
MAX_CONCURRENT = int(os.getenv("MAX_CONCURRENT", 10))  # 最大并发数
RETRY_MAX = 3  # 最大重试次数
API_URL = "https://eco.taobao.com/router/rest"


class TaobaoAPIClient:
    """淘宝API异步客户端"""
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = aiohttp.ClientSession()

    def _generate_sign(self, params: Dict) -> str:
        """生成签名"""
        # 过滤空值并排序
        filtered = {k: v for k, v in params.items() if v is not None and k != "sign"}
        sorted_params = sorted(filtered.items(), key=lambda x: x[0])
        # 拼接签名字符串
        sign_str = "&".join([f"{k}={v}" for k, v in sorted_params]) + self.app_secret
        return hashlib.md5(sign_str.encode()).hexdigest().upper()

    async def fetch_item(self, item_id: str, retry_count: int = 0) -> Optional[Dict]:
        """异步获取商品详情"""
        params = {
            "app_key": self.app_key,
            "method": "taobao.item.detail.get",
            "format": "json",
            "v": "2.0",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "sign_method": "md5",
            "item_id": item_id,
            "fields": "item:id,title,price,desc,pics,sales,stock,shop_name"
        }
        params["sign"] = self._generate_sign(params)

        try:
            async with self.session.get(API_URL, params=params, timeout=10) as resp:
                result = await resp.json()
                
                if "error_response" in result:
                    error = result["error_response"]
                    # 限流或服务器错误时重试
                    if error["code"] in [11, 42, 500] and retry_count < RETRY_MAX:
                        wait_time = 2 **retry_count  # 指数退避
                        await asyncio.sleep(wait_time)
                        return await self.fetch_item(item_id, retry_count + 1)
                    print(f"商品{item_id}获取失败: {error['msg']}")
                    return None
                return result.get("item_detail_get_response", {}).get("item")
        
        except Exception as e:
            if retry_count < RETRY_MAX:
                wait_time = 2** retry_count
                await asyncio.sleep(wait_time)
                return await self.fetch_item(item_id, retry_count + 1)
            print(f"商品{item_id}请求异常: {str(e)}")
            return None

    async def close(self):
        """关闭客户端会话"""
        await self.session.close()


class DataProcessor:
    """数据处理器（消费者）"""
    def __init__(self, queue: Queue, db_name: str = "taobao_data"):
        self.queue = queue
        self.client = MongoClient(MONGO_URI)
        self.db = self.client[db_name]
        self.collection = self.db["items"]
        self.running = True
        # 启动处理线程
        self.thread = Thread(target=self.process, daemon=True)
        self.thread.start()

    def process(self):
        """处理队列中的数据"""
        while self.running or not self.queue.empty():
            item = self.queue.get()
            if not item:
                continue
            try:
                # 数据清洗与转换
                processed = {
                    "item_id": item["id"],
                    "title": item["title"],
                    "price": float(item["price"]),
                    "sales": int(item["sales"]),
                    "stock": int(item["stock"]),
                    "shop_name": item["shop_name"],
                    "pics": item["pics"],
                    "crawl_time": datetime.now()
                }
                # 存储到MongoDB（存在则更新）
                self.collection.update_one(
                    {"item_id": processed["item_id"]},
                    {"$set": processed},
                    upsert=True
                )
                print(f"已处理商品: {processed['item_id']}")
            except Exception as e:
                print(f"数据处理失败: {str(e)}")
            finally:
                self.queue.task_done()

    def stop(self):
        """停止处理器"""
        self.running = False
        self.thread.join()
        self.client.close()


async def batch_fetch(items: List[str], result_queue: Queue):
    """批量并发获取商品数据"""
    client = TaobaoAPIClient(APP_KEY, APP_SECRET)
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)  # 控制并发量

    async def bounded_fetch(item_id: str):
        """受信号量控制的抓取任务"""
        async with semaphore:
            data = await client.fetch_item(item_id)
            if data:
                result_queue.put(data)

    # 并发执行所有任务
    tasks = [bounded_fetch(item_id) for item_id in items]
    await asyncio.gather(*tasks)
    await client.close()


def main():
    # 1. 准备任务列表（示例：100个商品ID）
    item_ids = [f"654321{str(i).zfill(3)}" for i in range(100)]  # 实际应从文件/数据库读取

    # 2. 初始化结果队列和数据处理器
    result_queue = Queue(maxsize=1000)
    processor = DataProcessor(result_queue)

    # 3. 执行并发抓取
    start_time = time.time()
    asyncio.run(batch_fetch(item_ids, result_queue))
    
    # 4. 等待所有数据处理完成
    result_queue.join()
    processor.stop()

    # 5. 输出统计信息
    end_time = time.time()
    total_time = end_time - start_time
    print(f"\n采集完成！总耗时: {total_time:.2f}秒")
    print(f"平均每秒处理: {len(item_ids)/total_time:.2f}个商品")


if __name__ == "__main__":
    main()