Python 异步方案助力京东 API 数据采集：提升实时数据获取效率

import asyncio
import aiohttp
import json
import time
from typing import List, Dict, Optional
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class JDAsyncCrawler:
    def __init__(self, api_key: str, max_concurrent: int = 10):
        """
        初始化京东API异步采集器
        :param api_key: 京东API访问密钥
        :param max_concurrent: 最大并发数
        """
        self.api_key = api_key
        self.base_url = "https://api.jd.com/routerjson"
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session = None
        
    async def __aenter__(self):
        """创建aiohttp会话"""
        self.session = aiohttp.ClientSession()
        return self
        
    async def __aexit__(self, exc_type, exc, tb):
        """关闭aiohttp会话"""
        if self.session:
            await self.session.close()
    
    async def fetch_product(self, product_id: str) -> Optional[Dict]:
        """
        异步获取单个商品信息
        :param product_id: 商品ID
        :return: 商品信息字典，失败则返回None
        """
        async with self.semaphore:
            params = {
                "method": "jd.union.open.goods.info.query",
                "app_key": self.api_key,
                "format": "json",
                "v": "1.0",
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
                "param_json": json.dumps({"skuIds": [product_id]})
            }
            
            try:
                start_time = time.time()
                async with self.session.get(self.base_url, params=params) as response:
                    if response.status == 200:
                        data = await response.json()
                        logger.info(f"获取商品 {product_id} 成功，耗时: {time.time() - start_time:.2f}秒")
                        return data
                    else:
                        logger.error(f"获取商品 {product_id} 失败，状态码: {response.status}")
                        return None
            except Exception as e:
                logger.error(f"获取商品 {product_id} 发生错误: {str(e)}")
                return None
    
    async def fetch_products_batch(self, product_ids: List[str]) -> Dict[str, Optional[Dict]]:
        """
        批量异步获取商品信息
        :param product_ids: 商品ID列表
        :return: 以商品ID为键，商品信息为值的字典
        """
        logger.info(f"开始批量获取 {len(product_ids)} 个商品信息")
        start_time = time.time()
        
        # 创建所有任务
        tasks = [self.fetch_product(pid) for pid in product_ids]
        
        # 并发执行所有任务
        results = await asyncio.gather(*tasks)
        
        # 整理结果
        product_data = {pid: data for pid, data in zip(product_ids, results)}
        
        logger.info(f"批量获取完成，总耗时: {time.time() - start_time:.2f}秒，"
                   f"平均每个商品耗时: {(time.time() - start_time)/len(product_ids):.2f}秒")
        
        return product_data

async def main():
    # 替换为实际的API密钥
    API_KEY = "your_jd_api_key"
    
    # 要查询的商品ID列表
    product_ids = [
        "100012345678", "100009876543", "100005678901",
        "100001234567", "100008765432", "100002345678",
        "100003456789", "100004567890", "100005678901",
        "100006789012", "100007890123", "100008901234",
        "100009012345", "100000123456", "100001234560"
    ]
    
    # 使用异步上下文管理器创建采集器
    async with JDAsyncCrawler(API_KEY, max_concurrent=5) as crawler:
        # 批量获取商品信息
        products_data = await crawler.fetch_products_batch(product_ids)
        
        # 处理获取到的数据（这里仅简单统计成功数量）
        success_count = sum(1 for data in products_data.values() if data is not None)
        logger.info(f"数据采集完成，成功获取 {success_count}/{len(product_ids)} 个商品信息")
        
        # 可以在这里添加数据存储或进一步处理的逻辑
        # 例如保存到数据库、写入文件等

if __name__ == "__main__":
    # 运行异步主函数
    asyncio.run(main())