1688 API 开发避坑指南：解析限流策略、频控规则及异常状态码的标准化处理

┌───────────┬─────────────────────────┬─────────────────────────┐
│ 规则类型  │ 核心目的                │ 触发后果                │
├───────────┼─────────────────────────┼─────────────────────────┤
│ 限流      │ 管控资源占用，分配额度  │ 单次请求失败，返回限流码│
│ 频控      │ 防范恶意调用，保障安全  │ 接口临时封禁，批量失败  │
└───────────┴─────────────────────────┴─────────────────────────┘

import requests
import time
import hashlib
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# ===================== 1688 API 配置常量（替换为自己的配置）=====================
APP_KEY = "your_1688_appkey"          # 开放平台应用AppKey
APP_SECRET = "your_1688_appsecret"    # 开放平台应用AppSecret
BASE_URL = "https://gw.open.1688.com/openapi/param2/1/"  # 1688 API网关地址
TIMEOUT = 10  # 请求超时时间（秒）
MAX_RETRY_TIMES = 3  # 最大重试次数

# ===================== 自定义异常类（标准化异常体系）=====================
class Api1688Exception(Exception):
    """1688 API 基础异常类"""
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message
        super().__init__(f"1688 API异常 [code:{code}]：{message}")

class AuthFailedException(Api1688Exception):
    """认证失败异常（401）"""
    pass

class PermissionDeniedException(Api1688Exception):
    """权限不足异常（403）"""
    pass

class RateLimitException(Api1688Exception):
    """限流/频控异常（429）"""
    pass

class PlatformServerException(Api1688Exception):
    """平台服务异常（500/503）"""
    pass

# ===================== 核心API调用基类（避坑核心）=====================
class Api1688Client:
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = requests.Session()
        self.last_request_time = 0  # 记录上一次请求时间，实现基础频控

    def _generate_sign(self, params: dict) -> str:
        """生成1688 API签名（必填，身份校验核心）"""
        # 1. 按字典序排序参数
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接参数字符串
        sign_str = self.app_secret
        for k, v in sorted_params:
            if v is not None and str(v).strip():
                sign_str += f"{k}{v}"
        sign_str += self.app_secret
        # 3. MD5加密并转大写
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()

    def _base_request(self, api_method: str, params: dict) -> dict:
        """基础请求方法：处理签名、请求发送、状态码校验"""
        # 1. 基础频控：强制间隔0.2秒，避免短时间突发调用（避坑频控核心）
        current_time = time.time()
        interval = current_time - self.last_request_time
        if interval < 0.2:
            time.sleep(0.2 - interval)
        self.last_request_time = time.time()

        # 2. 拼接公共参数（1688 API必传）
        public_params = {
            "app_key": self.app_key,
            "method": api_method,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5"
        }
        # 合并公共参数与业务参数
        all_params = {**public_params, **params}
        # 生成签名
        all_params["sign"] = self._generate_sign(all_params)

        # 3. 发送请求
        try:
            response = self.session.get(
                url=BASE_URL + api_method.replace(".", "/"),
                params=all_params,
                timeout=TIMEOUT
            )
            response.raise_for_status()  # 抛出HTTP状态码异常
            result = response.json()
        except requests.exceptions.RequestException as e:
            raise Api1688Exception(-1, f"请求发送失败：{str(e)}")

        # 4. 标准化状态码处理（核心避坑：全量校验code）
        code = result.get("code", -1)
        message = result.get("message", "未知异常")

        if code == 0:
            return result.get("data", {})  # 成功：返回业务数据
        elif code == 401:
            raise AuthFailedException(code, message)
        elif code == 403:
            raise PermissionDeniedException(code, message)
        elif code == 429:
            raise RateLimitException(code, message)
        elif code in [500, 503]:
            raise PlatformServerException(code, message)
        else:
            raise Api1688Exception(code, message)

    @retry(
        stop=stop_after_attempt(MAX_RETRY_TIMES),  # 最大重试3次
        wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避：2s→4s→8s
        retry=retry_if_exception_type((RateLimitException, PlatformServerException)),  # 仅限流/平台异常重试
        reraise=True  # 重试失败后，重新抛出异常
    )
    def call_api(self, api_method: str, params: dict = None) -> dict:
        """对外暴露的API调用方法：集成智能重试，实现限流自动降级"""
        if params is None:
            params = {}
        return self._base_request(api_method, params)

def test_1688_api():
    """1688 API 业务调用示例：商品详情查询"""
    # 1. 初始化客户端（传入自己的AppKey/AppSecret）
    client = Api1688Client(APP_KEY, APP_SECRET)

    try:
        # 2. 调用商品详情接口（示例：替换为实际需要的API方法）
        # 接口文档参考：https://open.1688.com/api/api.htm?spm=a260k.home.category.demo_0.391b33e15a9009&cat=product
        api_method = "com.alibaba.product:alibaba.product.info.get-1"
        business_params = {
            "productId": "699999999999",  # 1688商品ID
            "fields": "productId,productName,price,stock"  # 需要查询的字段
        }
        # 3. 发起调用（自动处理签名、限流、重试、异常）
        product_data = client.call_api(api_method, business_params)
        print("✅ 商品详情查询成功：")
        print(product_data)

    except AuthFailedException as e:
        print(f"❌ 认证失败：{e.message}，请检查AppKey/AppSecret是否正确")
    except PermissionDeniedException as e:
        print(f"❌ 权限不足：{e.message}，请开通商品查询接口权限")
    except RateLimitException as e:
        print(f"❌ 限流触发：{e.message}，已达最大重试次数，建议降级处理")
    except PlatformServerException as e:
        print(f"❌ 平台服务异常：{e.message}，请稍后重试")
    except Api1688Exception as e:
        print(f"❌ API调用失败：{e.message}")
    except Exception as e:
        print(f"❌ 系统异常：{str(e)}")

if __name__ == "__main__":
    test_1688_api()