技术解析：如何通过 1688 API 高效获取商品详情实时数据

import requests
import time
import hashlib
import json
from urllib.parse import urlencode
import logging
from functools import lru_cache

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

class Ali1688APIClient:
    def __init__(self, app_key, app_secret, access_token=None):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.base_url = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/"
        self.timeout = 10
        self.max_retries = 3
        
    def get_access_token(self, grant_type="client_credentials"):
        """获取访问令牌"""
        url = "https://gw.open.1688.com/auth/token"
        
        params = {
            "grant_type": grant_type,
            "client_id": self.app_key,
            "client_secret": self.app_secret
        }
        
        try:
            response = requests.get(url, params=params, timeout=self.timeout)
            result = response.json()
            
            if "access_token" in result:
                self.access_token = result["access_token"]
                logger.info(f"成功获取access_token，有效期: {result.get('expires_in')}秒")
                return self.access_token
            else:
                logger.error(f"获取access_token失败: {result}")
                return None
        except Exception as e:
            logger.error(f"获取access_token异常: {str(e)}")
            return None
    
    def _generate_signature(self, params):
        """生成签名"""
        # 按参数名升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 拼接参数
        sign_str = self.app_secret
        for key, value in sorted_params:
            sign_str += f"{key}{value}"
        sign_str += self.app_secret
        # 计算MD5
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def _api_request(self, api_name, params, need_token=True):
        """通用API请求方法"""
        # 基础参数
        request_params = {
            "app_key": self.app_key,
            "format": "json",
            "v": "1.0",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "sign_method": "md5"
        }
        
        # 添加访问令牌
        if need_token and self.access_token:
            request_params["access_token"] = self.access_token
        
        # 合并业务参数
        request_params.update(params)
        
        # 生成签名
        request_params["sign"] = self._generate_signature(request_params)
        
        # 构建请求URL
        url = f"{self.base_url}{api_name}"
        
        # 发送请求（带重试机制）
        for retry in range(self.max_retries):
            try:
                logger.info(f"调用API: {api_name}, 参数: {json.dumps(params, ensure_ascii=False)}")
                response = requests.get(url, params=request_params, timeout=self.timeout)
                result = response.json()
                
                # 检查是否请求成功
                if "error_response" in result:
                    error_code = result["error_response"].get("code")
                    error_msg = result["error_response"].get("msg")
                    logger.error(f"API调用错误 [retry={retry}]: {error_code} - {error_msg}")
                    
                    # 如果是令牌过期，尝试重新获取令牌
                    if error_code == 100006:
                        logger.info("令牌已过期，尝试重新获取...")
                        if self.get_access_token():
                            continue  # 重新获取令牌后重试
                    
                    # 达到最大重试次数
                    if retry == self.max_retries - 1:
                        return {"success": False, "error": f"{error_code} - {error_msg}"}
                else:
                    return {"success": True, "data": result}
                    
            except Exception as e:
                logger.error(f"API请求异常 [retry={retry}]: {str(e)}")
                if retry == self.max_retries - 1:
                    return {"success": False, "error": str(e)}
            
            # 重试前等待一段时间
            time.sleep(1 * (retry + 1))
        
        return {"success": False, "error": "达到最大重试次数"}
    
    @lru_cache(maxsize=1000)
    def get_product_detail(self, product_id, fields=None):
        """
        获取商品详情
        
        :param product_id: 商品ID
        :param fields: 需要返回的字段列表，None则返回所有字段
        :return: 商品详情数据
        """
        if not product_id:
            return {"success": False, "error": "商品ID不能为空"}
        
        # 默认返回的字段
        default_fields = "product_id,subject,price,rprice,amount,on_sale,detail_url,pic_url,sale_info"
        if fields:
            # 如果指定了字段，使用指定的
            fields_str = ",".join(fields)
        else:
            fields_str = default_fields
        
        params = {
            "product_id": product_id,
            "fields": fields_str
        }
        
        return self._api_request("alibaba.product.get", params)
    
    def batch_get_product_details(self, product_ids, fields=None, max_concurrent=5):
        """
        批量获取商品详情
        
        :param product_ids: 商品ID列表
        :param fields: 需要返回的字段列表
        :param max_concurrent: 最大并发数
        :return: 商品详情列表
        """
        if not product_ids or not isinstance(product_ids, list):
            return {"success": False, "error": "商品ID列表不能为空"}
        
        results = []
        # 分批处理，控制并发
        for i in range(0, len(product_ids), max_concurrent):
            batch = product_ids[i:i+max_concurrent]
            batch_results = []
            
            # 并发获取当前批次的商品详情
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrent) as executor:
                # 提交所有任务
                futures = {executor.submit(self.get_product_detail, pid, fields): pid for pid in batch}
                
                # 获取结果
                for future in concurrent.futures.as_completed(futures):
                    pid = futures[future]
                    try:
                        result = future.result()
                        batch_results.append({
                            "product_id": pid,
                            "result": result
                        })
                    except Exception as e:
                        logger.error(f"批量获取商品 {pid} 详情异常: {str(e)}")
                        batch_results.append({
                            "product_id": pid,
                            "result": {"success": False, "error": str(e)}
                        })
            
            results.extend(batch_results)
            # 每批请求后休息一下，避免触发限流
            time.sleep(1)
        
        return {"success": True, "data": results}

# 使用示例
if __name__ == "__main__":
    # 替换为你的App Key和App Secret
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    
    # 初始化客户端
    client = Ali1688APIClient(APP_KEY, APP_SECRET)
    
    # 获取访问令牌
    if not client.get_access_token():
        print("无法获取访问令牌，程序退出")
        exit(1)
    
    # 单个商品详情获取
    product_id = "610784549587"  # 示例商品ID
    print(f"获取商品 {product_id} 详情...")
    result = client.get_product_detail(product_id)
    if result["success"]:
        print(json.dumps(result["data"], ensure_ascii=False, indent=2))
    else:
        print(f"获取失败: {result['error']}")
    
    # 批量获取商品详情
    product_ids = ["610784549587", "626644646828", "625847634767"]  # 示例商品ID列表
    print(f"\n批量获取商品 {product_ids} 详情...")
    batch_result = client.batch_get_product_details(product_ids)
    if batch_result["success"]:
        for item in batch_result["data"]:
            print(f"\n商品 {item['product_id']} 详情:")
            if item["result"]["success"]:
                print(json.dumps(item["result"]["data"], ensure_ascii=False, indent=2))
            else:
                print(f"获取失败: {item['result']['error']}")
    else:
        print(f"批量获取失败: {batch_result['error']}")