淘宝评论 API 性能优化：缓存与并发控制

import redisimport jsonimport timefrom functools import lru_cachefrom datetime import datetime, timedelta# -------------------------- 缓存配置 --------------------------# Redis配置（二级缓存）REDIS_CONFIG = {
    "host": "localhost",
    "port": 6379,
    "db": 0,
    "decode_responses": True,
    "password": ""  # 无密码则留空}REDIS_CLIENT = redis.Redis(**REDIS_CONFIG)# 缓存过期时间MEMORY_CACHE_TTL = 300  # 内存缓存5分钟（秒）REDIS_CACHE_TTL = 3600   # Redis缓存1小时（秒）PERSIST_CACHE_TTL = 86400  # 持久化缓存24小时（秒）# 一级缓存：内存LRU缓存（限制最多缓存100个商品）@lru_cache(maxsize=100)def memory_cache_get(key):
    """从内存缓存获取数据"""
    return None  # 仅作为缓存标记，实际数据存在Redisdef memory_cache_set(key):
    """设置内存缓存标记"""
    memory_cache_get(key)  # 利用lru_cache的缓存特性# -------------------------- 通用缓存操作 --------------------------def get_cache_key(product_id, page_no):
    """生成缓存Key：taobao_review:{商品ID}:{页码}"""
    return f"taobao_review:{product_id}:{page_no}"def cache_get(product_id, page_no):
    """
    多级缓存读取：内存→Redis→持久化（MySQL）
    :return: 评论数据（None表示无缓存）
    """
    cache_key = get_cache_key(product_id, page_no)
    
    # 1. 一级缓存：内存（仅检查是否存在，避免存储大体积数据）
    if memory_cache_get(cache_key) is not None:
        # 2. 二级缓存：Redis读取实际数据
        redis_data = REDIS_CLIENT.get(cache_key)
        if redis_data:
            return json.loads(redis_data)
    
    # 3. 三级缓存：MySQL（示例，实际需结合前文数据库表）
    # 此处省略MySQL读取逻辑，可参考前文taobao_reviews表查询
    return Nonedef cache_set(product_id, page_no, data):
    """
    多级缓存写入：内存→Redis→持久化（MySQL）
    """
    cache_key = get_cache_key(product_id, page_no)
    # 1. 一级缓存：设置内存标记
    memory_cache_set(cache_key)
    # 2. 二级缓存：写入Redis（带过期时间）
    REDIS_CLIENT.setex(cache_key, REDIS_CACHE_TTL, json.dumps(data, ensure_ascii=False))
    # 3. 三级缓存：写入MySQL（可选，持久化）
    # 此处省略MySQL写入逻辑

from top.api import TbkItemReviewGetRequest, TopClient# 淘宝API配置APP_KEY = "你的App Key"APP_SECRET = "你的App Secret"SERVER_URL = "http://gw.api.taobao.com/router/rest"client = TopClient(appkey=APP_KEY, appsecret=APP_SECRET, url=SERVER_URL)def get_review_with_cache(product_id, page_no=1, page_size=20):
    """
    带缓存的评论获取函数：优先读缓存，缓存未命中再调用API
    :return: 评论数据 + 是否命中缓存
    """
    # 1. 尝试读取缓存
    cache_data = cache_get(product_id, page_no)
    if cache_data:
        print(f"✅ 缓存命中：商品{product_id} 第{page_no}页")
        return cache_data, True
    
    # 2. 缓存未命中，调用API
    print(f"🔄 缓存未命中，调用API：商品{product_id} 第{page_no}页")
    req = TbkItemReviewGetRequest()
    req.num_iid = product_id
    req.page_no = page_no
    req.page_size = page_size
    req.fields = "content,created,user_nick,rate"

    try:
        # API调用（带简单重试）
        resp = client.execute(req)
        if "tbk_item_review_get_response" not in resp:
            return [], False
        
        # 解析数据
        response = resp["tbk_item_review_get_response"]
        reviews = response.get("results", {}).get("tbk_item_review", [])
        parsed_data = [{
            "content": r.get("content", "").strip(),
            "rate": int(r.get("rate", 0)),
            "created": r.get("created", ""),
            "user_nick": r.get("user_nick", "")
        } for r in reviews]

        # 写入缓存
        cache_set(product_id, page_no, parsed_data)
        return parsed_data, False
    except Exception as e:
        print(f"❌ API调用失败：{e}")
        return [], False

import threadingimport timeclass TokenBucket:
    """令牌桶算法：控制API调用速率"""
    def __init__(self, capacity=2, rate=1):
        """
        :param capacity: 令牌桶容量（最大并发数）
        :param rate: 令牌生成速率（个/秒，适配QPS限制）
        """
        self.capacity = capacity
        self.rate = rate
        self.tokens = capacity
        self.last_refill_time = time.time()
        self.lock = threading.Lock()

    def get_token(self, timeout=5):
        """获取令牌，超时返回False"""
        start_time = time.time()
        while True:
            with self.lock:
                # 补充令牌
                now = time.time()
                refill_tokens = (now - self.last_refill_time) * self.rate
                self.tokens = min(self.capacity, self.tokens + refill_tokens)
                self.last_refill_time = now                # 尝试获取令牌
                if self.tokens >= 1:
                    self.tokens -= 1
                    return True
            
            # 超时判断
            if time.time() - start_time > timeout:
                return False
            time.sleep(0.01)  # 10ms轮询# 初始化令牌桶：容量2，速率1个/秒（适配QPS=1）token_bucket = TokenBucket(capacity=2, rate=1)

import asyncioimport aiohttpimport jsonfrom top.api import TbkItemReviewGetRequest, TopClient# 异步API调用封装（适配topapi，若不支持则直接调用HTTP接口）async def async_get_review(product_id, page_no, page_size=20):
    """
    异步调用淘宝评论API（直接构造HTTP请求，避免topapi同步限制）
    """
    # 1. 获取令牌（速率控制）
    if not token_bucket.get_token(timeout=10):
        print(f"⚠️ 令牌获取超时：商品{product_id} 第{page_no}页")
        return []
    
    # 2. 构造API请求参数（TOP API签名逻辑，简化版）
    # 注：完整签名需按淘宝规则实现，此处可参考topapi源码
    params = {
        "method": "taobao.tbk.item.review.get",
        "app_key": APP_KEY,
        "format": "json",
        "v": "2.0",
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "num_iid": product_id,
        "page_no": page_no,
        "page_size": page_size,
        "fields": "content,created,user_nick,rate"
    }
    # 补充签名（需实现淘宝TOP API的签名逻辑，此处省略，可复用topapi的签名函数）
    # params["sign"] = generate_sign(params, APP_SECRET)

    # 3. 异步请求
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(SERVER_URL, params=params, timeout=15) as resp:
                if resp.status != 200:
                    return []
                raw_data = await resp.json()
                # 解析数据（同前文）
                if "tbk_item_review_get_response" not in raw_data:
                    return []
                reviews = raw_data["tbk_item_review_get_response"].get("results", {}).get("tbk_item_review", [])
                return [{
                    "content": r.get("content", ""),
                    "rate": int(r.get("rate", 0)),
                    "created": r.get("created", "")
                } for r in reviews]
        except Exception as e:
            print(f"❌ 异步请求失败：{e}")
            return []# 批量异步采集多商品async def batch_async_crawl(product_ids, max_pages=5):
    """
    异步采集多个商品的评论
    :param product_ids: 商品ID列表
    :param max_pages: 单商品最大采集页数
    """
    tasks = []
    for product_id in product_ids:
        for page_no in range(1, max_pages + 1):
            # 每个商品的分页串行，避免单商品高频调用
            task = asyncio.create_task(async_get_review(product_id, page_no))
            tasks.append(task)
            await asyncio.sleep(0.1)  # 单商品分页间隔
    
    # 等待所有任务完成
    results = await asyncio.gather(*tasks)
    # 合并结果
    all_reviews = []
    for idx, res in enumerate(results):
        product_id = product_ids[idx // max_pages]
        page_no = (idx % max_pages) + 1
        all_reviews.extend([{**r, "product_id": product_id, "page_no": page_no} for r in res])
    return all_reviews# 执行异步采集if __name__ == "__main__":
    PRODUCT_IDS = ["123456789", "987654321", "456789123"]  # 待采集商品ID
    start_time = time.time()
    # 运行异步函数
    reviews = asyncio.run(batch_async_crawl(PRODUCT_IDS, max_pages=5))
    end_time = time.time()
    print(f"✅ 采集完成：共{len(reviews)}条评论，耗时{end_time - start_time:.2f}秒")

def retry_with_backoff(func, max_retry=3):
    """指数退避重试装饰器"""
    def wrapper(*args, **kwargs):
        retry_count = 0
        while retry_count < max_retry:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                # 仅重试限流/网络错误
                if "quota-api-limit" in str(e) or "timeout" in str(e):
                    sleep_time = 2 ** retry_count                    print(f"⚠️ 重试{retry_count+1}/{max_retry}，等待{sleep_time}秒：{e}")
                    time.sleep(sleep_time)
                    retry_count += 1
                else:
                    break
        return None
    return wrapper# 应用装饰器@retry_with_backoffdef get_review_with_retry(product_id, page_no):
    return get_review_with_cache(product_id, page_no)

import asyncioimport timefrom functools import lru_cacheimport redis# -------------------------- 整合缓存与并发 --------------------------async def crawl_review_optimized(product_id, page_no):
    """
    优化后的评论采集函数：缓存+速率控制+异步+重试
    """
    # 1. 优先读缓存
    cache_data = cache_get(product_id, page_no)
    if cache_data:
        return cache_data    
    # 2. 速率控制（获取令牌）
    if not token_bucket.get_token(timeout=10):
        return []
    
    # 3. 异步调用API（带重试）
    try:
        reviews = await async_get_review(product_id, page_no)
        # 4. 写入缓存
        if reviews:
            cache_set(product_id, page_no, reviews)
        return reviews    except Exception as e:
        print(f"❌ 采集失败：{product_id} 第{page_no}页 - {e}")
        return []# 批量采集示例async def batch_crawl_optimized(product_ids, max_pages=5):
    start_time = time.time()
    tasks = []
    for product_id in product_ids:
        for page_no in range(1, max_pages + 1):
            tasks.append(crawl_review_optimized(product_id, page_no))
    results = await asyncio.gather(*tasks)
    # 合并结果
    all_reviews = []
    for res in results:
        all_reviews.extend(res)
    end_time = time.time()
    print(f"✅ 优化后采集完成：共{len(all_reviews)}条评论，耗时{end_time - start_time:.2f}秒")
    return all_reviews# 执行if __name__ == "__main__":
    PRODUCT_IDS = ["123456789", "987654321", "456789123"]
    asyncio.run(batch_crawl_optimized(PRODUCT_IDS, max_pages=5))