如何处理淘宝 API 的请求限流与数据缓存策略

import time
import random
import requests
from typing import Dict, Optional
from functools import wraps

# 淘宝API配置（请替换为你的真实信息）
TAOBAO_APP_KEY = "your_app_key"
TAOBAO_APP_SECRET = "your_app_secret"
TAOBAO_API_URL = "https://eco.taobao.com/router/rest"

# 限流控制配置
RATE_LIMIT = 5  # 每秒最大调用次数（根据淘宝API实际限额调整）
DAILY_LIMIT = 10000  # 单日最大调用次数
TOKEN_BUCKET = RATE_LIMIT  # 令牌桶初始令牌数
LAST_TOKEN_REFILL_TIME = time.time()  # 上次令牌补充时间
DAILY_CALL_COUNT = 0  # 当日已调用次数

def token_bucket_decorator(func):
    """令牌桶限流装饰器：控制请求频率"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        global TOKEN_BUCKET, LAST_TOKEN_REFILL_TIME, DAILY_CALL_COUNT
        
        # 1. 检查单日配额是否用尽
        if DAILY_CALL_COUNT >= DAILY_LIMIT:
            raise Exception(f"单日调用配额已用尽（限额：{DAILY_LIMIT}）")
        
        # 2. 补充令牌桶（按时间比例补充）
        current_time = time.time()
        time_elapsed = current_time - LAST_TOKEN_REFILL_TIME
        new_tokens = time_elapsed * RATE_LIMIT
        TOKEN_BUCKET = min(TOKEN_BUCKET + new_tokens, RATE_LIMIT)
        LAST_TOKEN_REFILL_TIME = current_time
        
        # 3. 等待获取令牌
        while TOKEN_BUCKET < 1:
            time.sleep(0.01)
            current_time = time.time()
            time_elapsed = current_time - LAST_TOKEN_REFILL_TIME
            new_tokens = time_elapsed * RATE_LIMIT
            TOKEN_BUCKET = min(TOKEN_BUCKET + new_tokens, RATE_LIMIT)
            LAST_TOKEN_REFILL_TIME = current_time
        
        # 4. 消耗令牌并执行请求
        TOKEN_BUCKET -= 1
        DAILY_CALL_COUNT += 1
        return func(*args, **kwargs)
    return wrapper

def exponential_backoff_retry(max_retries: int = 3):
    """指数退避重试装饰器：处理限流错误"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    # 检测淘宝API限流错误码
                    if "isv.access-limited" in str(e) or "system.busy" in str(e):
                        retries += 1
                        if retries >= max_retries:
                            raise Exception(f"重试{max_retries}次后仍限流，请求失败：{e}")
                        # 指数退避等待：2^retries + 随机值（避免并发重试冲突）
                        wait_time = (2 ** retries) + random.uniform(0, 1)
                        print(f"触发限流，{wait_time:.2f}秒后重试（第{retries}次）")
                        time.sleep(wait_time)
                    else:
                        # 非限流错误直接抛出
                        raise e
            return None
        return wrapper
    return decorator

@token_bucket_decorator
@exponential_backoff_retry(max_retries=3)
def call_taobao_api(method: str, params: Dict) -> Optional[Dict]:
    """
    调用淘宝API的核心函数
    :param method: API接口名称（如taobao.item.get）
    :param params: API请求参数
    :return: API返回的JSON数据
    """
    # 组装通用参数
    common_params = {
        "app_key": TAOBAO_APP_KEY,
        "method": method,
        "format": "json",
        "v": "2.0",
        "sign_method": "md5",
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        "partner_id": "apidoc"
    }
    # 合并参数（业务参数 + 通用参数）
    all_params = {**common_params, **params}
    
    # 生成签名（淘宝API要求，此处简化，实际需按官方签名规则实现）
    # 注：生产环境需严格按照淘宝开放平台文档实现签名逻辑
    sorted_params = sorted(all_params.items(), key=lambda x: x[0])
    sign_str = TAOBAO_APP_SECRET + "".join([f"{k}{v}" for k, v in sorted_params]) + TAOBAO_APP_SECRET
    all_params["sign"] = sign_str.lower()  # 示例签名，实际需MD5加密
    
    # 发送请求
    response = requests.get(TAOBAO_API_URL, params=all_params, timeout=10)
    result = response.json()
    
    # 检查返回结果中的错误
    if "error_response" in result:
        error_code = result["error_response"]["code"]
        error_msg = result["error_response"]["msg"]
        raise Exception(f"API调用失败：{error_code} - {error_msg}")
    
    return result

# 示例：调用淘宝商品查询API
if __name__ == "__main__":
    try:
        # 查询商品基础信息（示例参数）
        item_params = {
            "num_iid": "123456789"  # 商品ID
        }
        result = call_taobao_api("taobao.item.get", item_params)
        print(f"商品信息查询成功：{result}")
    except Exception as e:
        print(f"请求失败：{e}")

import hashlib
import redis
from typing import Dict, Optional

# 初始化Redis连接（请替换为你的Redis地址）
REDIS_CLIENT = redis.Redis(
    host="127.0.0.1",
    port=6379,
    db=0,
    decode_responses=True  # 自动将返回值转为字符串
)

def generate_cache_key(method: str, params: Dict) -> str:
    """生成唯一的缓存Key"""
    # 对参数排序后MD5，避免参数顺序不同导致Key不同
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    params_str = "".join([f"{k}{v}" for k, v in sorted_params])
    params_md5 = hashlib.md5(params_str.encode()).hexdigest()
    return f"taobao:{method}:{params_md5}"

def cache_decorator(ttl: int = 3600):
    """Redis缓存装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(method: str, params: Dict):
            # 1. 生成缓存Key
            cache_key = generate_cache_key(method, params)
            
            # 2. 尝试从缓存获取数据
            cached_data = REDIS_CLIENT.get(cache_key)
            if cached_data:
                import json
                return json.loads(cached_data)
            
            # 3. 缓存未命中，调用API获取数据
            api_data = func(method, params)
            
            # 4. 将数据存入缓存（处理空值，防止缓存穿透）
            if api_data is None:
                # 空值缓存，TTL设为5分钟
                REDIS_CLIENT.setex(cache_key, 300, json.dumps(None))
            else:
                REDIS_CLIENT.setex(cache_key, ttl, json.dumps(api_data))
            
            return api_data
        return wrapper
    return decorator

# 集成缓存的淘宝API调用函数
@cache_decorator(ttl=3600)  # 缓存1小时
@token_bucket_decorator
@exponential_backoff_retry(max_retries=3)
def call_taobao_api_with_cache(method: str, params: Dict) -> Optional[Dict]:
    """带缓存的淘宝API调用函数"""
    # 复用之前的call_taobao_api逻辑（此处省略重复代码，实际需整合）
    return call_taobao_api(method, params)

# 示例：调用带缓存的商品查询API
if __name__ == "__main__":
    try:
        item_params = {"num_iid": "123456789"}
        # 第一次调用：缓存未命中，走API
        result1 = call_taobao_api_with_cache("taobao.item.get", item_params)
        print(f"第一次调用结果：{result1}")
        
        # 第二次调用：缓存命中，直接返回
        result2 = call_taobao_api_with_cache("taobao.item.get", item_params)
        print(f"第二次调用结果（缓存）：{result2}")
    except Exception as e:
        print(f"请求失败：{e}")