优化实践：提升 1688 商品详情 API 接口稳定性和数据获取效率

import requests
import time
import random
from concurrent.futures import ThreadPoolExecutor, as_completed
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 1688 API核心配置
API_CONFIG = {
    "base_url": "https://gw.open.1688.com/openapi/param2/1/",
    "app_key": "your_app_key",  # 替换为实际AppKey
    "app_secret": "your_app_secret",  # 替换为实际AppSecret
    "api_method": "alibaba.item.get",  # 商品详情API方法名
    "timeout": 10,  # 请求超时时间（秒）
    "max_retries": 3,  # 最大重试次数
    "retry_delay": (1, 3),  # 重试延迟范围（秒），随机取值避免固定间隔
    "max_workers": 10,  # 并发线程数
    "rate_limit": 20,  # 每分钟最大请求数
}

# 初始化请求会话，配置重试策略
def init_request_session():
    """初始化带重试策略的requests会话"""
    session = requests.Session()
    # 定义重试规则：针对连接超时、5xx错误重试
    retry_strategy = Retry(
        total=API_CONFIG["max_retries"],
        backoff_factor=0.5,  # 重试间隔递增因子
        status_forcelist=[500, 502, 503, 504],  # 触发重试的状态码
        allowed_methods=["GET", "POST"]  # 允许重试的请求方法
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

# 初始化会话
session = init_request_session()

class RateLimiter:
    """请求限流器：控制每分钟请求数"""
    def __init__(self, max_requests_per_minute):
        self.max_requests = max_requests_per_minute
        self.requests = []  # 存储请求时间戳

    def acquire(self):
        """获取请求权限，超出限制则等待"""
        now = time.time()
        # 过滤1分钟前的请求记录
        self.requests = [t for t in self.requests if now - t < 60]
        if len(self.requests) >= self.max_requests:
            # 计算需要等待的时间
            wait_time = 60 - (now - self.requests[0])
            if wait_time > 0:
                time.sleep(wait_time)
        self.requests.append(time.time())

# 初始化限流器
rate_limiter = RateLimiter(API_CONFIG["rate_limit"])

def get_item_detail(item_id, session):
    """
    获取1688商品详情
    :param item_id: 商品ID
    :param session: requests会话
    :return: 商品详情字典/None
    """
    try:
        # 1. 限流控制
        rate_limiter.acquire()

        # 2. 构造请求参数（需根据1688 API规则签名，此处简化）
        params = {
            "app_key": API_CONFIG["app_key"],
            "method": API_CONFIG["api_method"],
            "item_id": item_id,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            # 签名参数（实际需结合app_secret生成，此处省略签名逻辑）
        }

        # 3. 发送请求
        response = session.get(
            API_CONFIG["base_url"],
            params=params,
            timeout=API_CONFIG["timeout"]
        )
        response.raise_for_status()  # 触发HTTP状态码异常

        # 4. 解析数据
        result = response.json()
        if result.get("error_response"):
            print(f"商品{item_id}请求失败：{result['error_response']}")
            return None

        # 5. 数据清洗：处理字段缺失/类型异常
        item_info = result.get("item_get_response", {}).get("item", {})
        clean_item_info = {
            "item_id": item_info.get("item_id", ""),
            "title": item_info.get("title", "").strip(),
            "price": float(item_info.get("price", 0)) if item_info.get("price") else 0.0,
            "sales": int(item_info.get("sales", 0)) if item_info.get("sales") else 0,
            "image_url": item_info.get("image_url", ""),
            "category": item_info.get("category", {}).get("name", "")
        }
        return clean_item_info

    except requests.exceptions.Timeout:
        print(f"商品{item_id}请求超时，重试中...")
        # 随机延迟后重试（避免重试风暴）
        time.sleep(random.uniform(*API_CONFIG["retry_delay"]))
        return get_item_detail(item_id, session)  # 递归重试
    except requests.exceptions.HTTPError as e:
        print(f"商品{item_id}HTTP错误：{e.response.status_code}")
        return None
    except Exception as e:
        print(f"商品{item_id}请求异常：{str(e)}")
        return None

def batch_get_item_details(item_ids):
    """
    批量获取商品详情
    :param item_ids: 商品ID列表
    :return: 商品详情列表
    """
    item_details = []
    # 初始化线程池
    with ThreadPoolExecutor(max_workers=API_CONFIG["max_workers"]) as executor:
        # 提交任务
        future_to_item = {
            executor.submit(get_item_detail, item_id, session): item_id
            for item_id in item_ids
        }
        # 获取结果
        for future in as_completed(future_to_item):
            item_id = future_to_item[future]
            try:
                detail = future.result()
                if detail:
                    item_details.append(detail)
                    print(f"商品{item_id}详情获取成功")
            except Exception as e:
                print(f"商品{item_id}批量处理异常：{str(e)}")
    return item_details

# 测试示例
if __name__ == "__main__":
    # 待获取的商品ID列表
    test_item_ids = ["123456789", "987654321", "112233445"]
    # 批量获取商品详情
    item_details = batch_get_item_details(test_item_ids)
    # 输出结果
    print("\n最终获取的商品详情：")
    for detail in item_details:
        print(detail)