《1688开放平台基础API真免费？QPS 10的限制怎么破（附GuardedClient源码）》

1. 客户端自律：令牌桶限速在 8~9 QPS（留余量防触发风控）。

2. 架构级扩容：申请多个同企业实名下的AppKey做轮询分担（理论QPS = 单Key上限 × N）。

3. 官方提额：在开放平台购买资源包将单Key提至50/100/200 QPS，或升级企业应用。

# guarded_1688_client.py
"""
1688开放平台 GuardedClient
- 多AppKey轮询突破单Key QPS限制
- 令牌桶单Key限速 (防触发ISP_FLOW_CONTROL_LIMIT)
- 捕获限流异常自动CoolDown + 指数退避重试
依赖: pip install requests
"""
import hashlib
import json
import time
import requests
from typing import Dict, List
from threading import Lock

# ------------------ 配置区 ------------------
APP_KEYS_SECRETS = [
    # 同一企业实名下创建的多个自用型应用
    {"key": "YOUR_APP_KEY_1", "secret": "YOUR_APP_SECRET_1"},
    {"key": "YOUR_APP_KEY_2", "secret": "YOUR_APP_SECRET_2"},
    # {"key": "...", "secret": "..."},
]
GW = "https://gw.api.1688.com/openapi"
MAX_QPS_PER_KEY = 8          # 单Key保守限速（官方≈10，留余量）
TOKEN_BUCKET_CAP = MAX_QPS_PER_KEY
# --------------------------------------------

封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ===== 单Key令牌桶 =====
class TokenBucket:
    def __init__(self, rate: float):
        self.rate = rate
        self.tokens = rate
        self.ts = time.monotonic()
        self.lock = Lock()

    def wait(self):
        with self.lock:
            now = time.monotonic()
            self.token = min(self.rate, self.tokens + (now - self.ts) * self.rate)
            self.ts = now
            if self.tokens < 1:
                time.sleep((1 - self.tokens) / self.rate + 0.005)
                self.tokens = 0
            else:
                self.tokens -= 1


# ===== 多Key轮询Guard =====
class Guarded1688Client:
    def __init__(self, apps: List[Dict], gw=GW):
        self.apps = apps
        self.gw = gw
        self.idx = 0
        self.idx_lock = Lock()
        # 为每个Key初始化独立令牌桶
        self.buckets = [TokenBucket(MAX_QPS_PER_KEY) for _ in apps]
        self.err_streak = [0] * len(apps)  # 连续限流计数

    def _next_app(self) -> (int, Dict, TokenBucket):
        with self.idx_lock:
            # 简单轮询，可改为按err_streak跳过冷却Key
            i = self.idx
            self.idx = (self.idx + 1) % len(self.apps)
            return i, self.apps[i], self.buckets[i]

    def _sign_1688(self, params: Dict, secret: str) -> str:
        """
        1688签名: md5(secret + sort(KV) + secret) -> 大写
        注意: 1688部分网关用 HmacMD5，这里以通用MD5为例，实际请对齐文档
        """
        filtered = sorted(
            (k, v) for k, v in params.items()
            if v is not None and str(v).strip() != "" and k != "sign"
        )
        qs = "".join(f"{k}{v}" for k, v in filtered)
        raw = f"{secret}{qs}{secret}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def safe_call(self, method: str, biz: Dict, token: str = None, max_retry=5):
        """
        method: 如 alibaba.item.get
        biz: 业务参数
        """
        for att in range(max_retry):
            i, app, bucket = self._next_app()
            bucket.wait()  # 单Key QPS守护

            # 构造1688 param2风格或router风格参数（示意，按你网关调整）
            api_params = {
                "method": method,
                "app_key": app["key"],
                "timestamp": str(int(time.time() * 1000)),  # 1688用毫秒
                "format": "json",
                "v": "2.0",
                "sign_method": "md5",
                **biz
            }
            if token:
                api_params["access_token"] = token
            api_params["sign"] = self._sign_1688(api_params, app["secret"])

            try:
                # 1688 param2 路径示例: /openapi/param2/1/{method}/{app_key}
                url = f"{self.gw}/param2/1/{method}/{app['key']}"
                resp = requests.post(url, data=api_params, timeout=15)
                resp.raise_for_status()
                d = resp.json()

                # 1688成功通常在顶层，错误在 error_response 或 code!=200
                if "error_response" in d:
                    err = d["error_response"]
                    code = str(err.get("code", ""))
                    msg = err.get("msg", "")
                    # 限流判定
                    if "FLOW_CONTROL" in msg or code == "15" or resp.status_code == 429:
                        self.err_streak[i] += 1
                        back = min(2 ** self.err_streak[i], 8)
                        print(f"⚠️ Key[{i}] 限流，冷却 {back}s 后重试...")
                        time.sleep(back)
                        continue
                    raise Exception(f"1688[{code}]: {msg}")

                # 成功重置错误计数
                self.err_streak[i] = 0
                return d

            except requests.exceptions.HTTPError as e:
                if "429" in str(e):
                    self.err_streak[i] += 1
                    time.sleep(2 ** att)
                    continue
                raise

        raise RuntimeError("1688 API 多次重试仍限流，建议降低QPS或增加AppKey数量")


# ===================== 使用示例 =====================
if __name__ == "__main__":
    client = Guarded1688Client(APP_KEYS_SECRETS)

    # 查商品详情（免费接口，受QPS限制）
    try:
        result = client.safe_call(
            "alibaba.item.get",
            {"item_id": "612345678901", "fields": "title,price,sku_list,amount_on_sale"},
            token=None  # 商品公开字段可不传token
        )
        print("✅ 1688商品数据:", result.get("alibaba_item_get_response", {}).get("result", {}).get("title"))
    except Exception as e:
        print("❌", e)

关键设计说明

1. 多Key轮询：通过 Guarded1688Client 在多个AppKey间分发请求，理论上N个Key可将总QPS提升至 N × 10。

2. 令牌桶限速：每个Key独立令牌桶，强制单Key ≤ 8~9 QPS，避免触碰官方阈值被拉黑。

3. 限流自愈：捕获 FLOW_CONTROL / 429 后对该Key做指数退避冷却，并自动切换下一个Key重试。

4. 合规提醒：多Key必须是同一企业主体下合法创建的不同应用，用于分散合法业务流量，非绕过封禁。

如果日调用量极大（>50万次）或对实时性要求高，首选买官方资源包提单Key QPS至50/100，成本远低于维护多应用复杂度；多Key方案更适合初创期零成本扛中等并发的场景。