《1688开放平台认证与权限：AppKey/Secret 管理、Token 刷新与接口调用限流踩坑》（附Python源码）

1688 开放平台里最容易被“能调通”骗了的，就是认证与限流层。

AppKey/AppSecret 是应用身份，access_token 是用户授权，refresh_token 是续命凭证，QPS 是平台给的命。

生产系统必须做到：密钥不落地、token 前置刷新、按接口分级限流、限流/网络/业务错误分开处理、多实例用分布式限流。

谁把 Secret 写进前端、谁在 401 上死重试、谁在 429 上立即重试、谁用“多 Key 爬商品”绕限流——谁就会被封应用或拿不到真实库存。

一、凭证体系：先分清楚 4 样东西

⚠️ 网上很多文章写“token 24h / 7d / 30d”，1688 官方口径是 access_token 过期超 10 小时后用 refresh_token 换；refresh_token 半年有效。

别把淘宝/抖音/微信的 token 模型直接套过来。

二、AppKey / Secret 管理：别犯低级错误

1. 红线

• Secret 不进代码仓库（用 KMS / Vault / 环境变量）

• 不进前端 JS / 客户端 App

• 不写进日志

• 不同环境（沙箱/生产）不同 Key

• 下单 Key / 商品同步 Key / 消息消费 Key 隔离

2. 最小密钥管理器

# auth/secret_store.py
import os
from dataclasses import dataclass


@dataclass(frozen=True)
class AppCredential:
    app_key: str
    app_secret: str
    env: str


class SecretStore:
    def get(self, env: str, usage: str) -> AppCredential:
        prefix = f"ALI1688_{env.upper()}_{usage.upper()}"
        key = os.environ[f"{prefix}_APPKEY"]
        secret = os.environ[f"{prefix}_SECRET"]
        if not key or not secret:
            raise RuntimeError(f"missing credential {prefix}")
        return AppCredential(key, secret, env)

cred = SecretStore().get("prod", "trade")   # 下单用 trade key
cred_search = SecretStore().get("prod", "catalog")  # 商品同步用 catalog key

三、签名：1688 TOP 风格

1. 空值参数不参加签名

2. 参数按 key ASCII 升序

3. 拼成 key1value1key2value2

4. 首尾拼 secret：secret + 拼接串 + secret

5. MD5，转大写

6. timestamp 用毫秒

# auth/signer.py
import hashlib
import time
from typing import Any, Dict


class Ali1688Signer:
    def __init__(self, app_secret: str):
        self.secret = app_secret

    def sign(self, params: Dict[str, Any]) -> str:
        flat = {
            k: v for k, v in params.items()
            if v is not None and k != "sign"
        }
        qs = "".join(f"{k}{v}" for k, v in sorted(flat.items()))
        raw = f"{self.secret}{qs}{self.secret}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def public_params(self, app_key: str, method: str, access_token: str = None):
        p = {
            "method": method,
            "app_key": app_key,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
        }
        if access_token:
            p["access_token"] = access_token
        return p

• 把 sign 自己又拼进去

• 中文先 urlencode 再签名（错）

• 毫秒写成秒

• None 当成 "None" 拼进去

• secret 只放前面不放后面

四、Token 管理：前置刷新，不等 401

1. token 缓存结构

# auth/token_store.py
import time
from dataclasses import dataclass


@dataclass
class TokenBundle:
    access_token: str
    refresh_token: str
    expires_at_ms: int        # access_token 过期时间
    refresh_expires_at_ms: int
    member_id: str = ""

    @property
    def access_almost_expired(self, skew_ms=10 * 60 * 1000) -> bool:
        return time.time() * 1000 >= (self.expires_at_ms - skew_ms)

    @property
    def refresh_still_valid(self) -> bool:
        return time.time() * 1000 < self.refresh_expires_at_ms

2. 刷新客户端

# auth/oauth_client.py
import requests

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class Ali1688OAuth:
    TOKEN_URL = "https://gw.open.1688.com/openapi/http/1/system.oauth2/getToken/YOUR_APPKEY"

    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret

    def refresh(self, refresh_token: str) -> dict:
        # 官方：grant_type=refresh_token，POST，HTTPS，不签名 
        resp = requests.post(
            self.TOKEN_URL.replace("YOUR_APPKEY", self.app_key),
            data={
                "grant_type": "refresh_token",
                "client_id": self.app_key,
                "client_secret": self.app_secret,
                "refresh_token": refresh_token,
            },
            timeout=5,
        )
        data = resp.json()
        if "access_token" not in data:
            raise RuntimeError(f"refresh failed: {data}")
        return data

3. 分布式刷新：别让 10 个 worker 同时刷

# auth/token_manager.py
import json
import time


class TokenManager:
    def __init__(self, redis, oauth, key: str, member_id: str):
        self.r = redis
        self.oauth = oauth
        self.redis_key = f"ali1688:token:{key}:{member_id}"
        self.lock_key = f"ali1688:token_lock:{key}:{member_id}"

    def load(self) -> TokenBundle | None:
        raw = self.r.get(self.redis_key)
        return TokenBundle(**json.loads(raw)) if raw else None

    def save(self, tb: TokenBundle):
        self.r.set(self.redis_key, json.dumps(tb.__dict__), ex=60 * 60 * 24 * 200)

    def acquire_refresh_lock(self) -> bool:
        return bool(self.r.set(self.lock_key, "1", nx=True, ex=30))

    def get_valid_token(self) -> str:
        tb = self.load()
        if tb and not tb.access_almost_expired:
            return tb.access_token

        # 快过期：抢锁刷新
        if self.acquire_refresh_lock():
            try:
                tb = self.load()
                if tb and tb.access_almost_expired and tb.refresh_still_valid:
                    data = self.oauth.refresh(tb.refresh_token)
                    new_tb = TokenBundle(
                        access_token=data["access_token"],
                        refresh_token=data.get("refresh_token", tb.refresh_token),
                        expires_at_ms=int(time.time() * 1000) + int(data.get("expires_in", 3600)) * 1000,
                        refresh_expires_at_ms=tb.refresh_expires_at_ms,
                        member_id=tb.member_id,
                    )
                    self.save(new_tb)
                    return new_tb.access_token
            finally:
                self.r.delete(self.lock_key)

        # 没抢到锁：等别人刷完再读
        time.sleep(0.3)
        return self.load().access_token

• access_token 过期前 10 分钟主动刷

• refresh_token 失效 → 清缓存、让用户重新 OAuth 跳转

• 多实例：Redis 锁，避免“刷新互踢”

• 401 回来再刷一次可以，但别把 401 当主流程

五、限流：1688 不是“日调用次数”游戏

1. 官方/实战口径

• 应用创建后有流量上限（如 5000）

• 商品搜索/详情类：社区经验约 10 QPS/Key

• 订单/物流类：社区经验约 20 QPS/Key

• 高级实时库存 / 跨境 / 分销：要资源包/权限，不是默认开

• 超限返回：ISP_FLOW_CONTROL_LIMIT / isv.freq-limit / 429

1688 更看重单 Key 瞬时 QPS，不是“今天还剩多少条”。

所以本地令牌桶比“每天 5000 次”更有用。

2. 本地令牌桶

# ratelimit/token_bucket.py
import time


class TokenBucket:
    def __init__(self, rate: float, capacity: float = None, jitter: float = 0.05):
        self.rate = rate                      # 留余量：官方10就设8
        self.capacity = capacity or rate
        self.tokens = self.capacity
        self.last = time.monotonic()
        self.jitter = jitter

    def acquire(self, n: int = 1):
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
            self.last = now
            if self.tokens >= n:
                self.tokens -= n
                return
            wait = (n - self.tokens) / self.rate + self.jitter
            time.sleep(wait)

3. 分布式限流（多实例必须）

# ratelimit/redis_sliding_window.py
import time


class RedisRateLimiter:
    def __init__(self, redis, key: str, qps: int):
        self.r = redis
        self.key = key
        self.qps = qps

    def allow(self) -> bool:
        now = time.time()
        pipe = self.r.pipeline()
        pipe.zremrangebyscore(self.key, 0, now - 1)              # 1s 滑动窗口
        pipe.zadd(self.key, {str(now): now})
        pipe.expire(self.key, 2)
        _, _, _ = pipe.execute()
        size = self.r.zcard(self.key)
        return size <= self.qps

• 单 Key 实际打 8~9 QPS，别顶满 10

• 多业务用多 Key：下单 / 商品同步 / 图搜 / 消息确认 分开

• 集群总 QPS = 各 Key 配额之和，用 Redis 统一管控

• 图搜/以图搜货：严控（3 QPS 内），最容易被风控

六、重试策略：网络/限流/业务要分开

# client/retry.py
import time


class RetriableError(Exception):
    pass


class FatalError(Exception):
    pass


def call_with_policy(fn, max_retry=3):
    for i in range(max_retry):
        try:
            return fn()
        except RetriableError as e:      # 网络超时 / 502 / 503 / 限流
            if i == max_retry - 1:
                raise
            # 限流：退避更长；网络抖动：短退避
            sleep = 2 ** i
            if "FLOW_CONTROL" in str(e) or "429" in str(e):
                sleep = 30 * (i + 1)
            time.sleep(sleep)
        except FatalError as e:           # 401 刷新后仍失败 / 403 / 参数错 / 商品下架
            raise                         # 不重试

七、完整客户端骨架

# client/ali1688_client.py
import requests


class Ali1688Client:
    def __init__(self, cred: AppCredential, token_manager, bucket: TokenBucket, timeout=5):
        self.cred = cred
        self.tm = token_manager
        self.bucket = bucket
        self.timeout = timeout
        self.gateway = "https://gw.open.1688.com/openapi/param2/2"

    def execute(self, namespace: str, method: str, biz: dict):
        token = self.tm.get_valid_token()
        signer = Ali1688Signer(self.cred.app_secret)

        params = signer.public_params(self.cred.app_key, method, token)
        params.update(biz)
        params["sign"] = signer.sign(params)

        url = f"{self.gateway}/{namespace}/{method}/{self.cred.app_key}"
        self.bucket.acquire()

        try:
            resp = requests.post(url, data=params, timeout=self.timeout)
        except requests.RequestException as e:
            raise RetriableError(str(e))

        body = resp.json()
        if "error_response" in body:
            err = body["error_response"].get("code", "")
            if err in ("401", "access_token expired"):
                # 强制失效，下次前置刷新
                self.tm.invalidate()
                raise RetriableError(err)
            if "FLOW_CONTROL" in str(body) or err == "429":
                raise RetriableError("rate limited")
            raise FatalError(body["error_response"])
        return body

八、权限申请坑

• 交易类（下单/查单/退款）：要应用认证 + 接口订购 + 用户授权

• 跨境/分销/严选：单独权限包，不是“企业认证就有”

• 密文面单：白名单 + 下游平台资质

• 消息订阅：Webhook 域名 HTTPS + 验签

• 权限“已申请”≠“已生效”，控制台状态要看“已授权/已订购/沙箱通过”

九、生产检查表

• [ ] Secret 不在代码/前端/日志

• [ ] 多业务多 Key 隔离

• [ ] token 存 Redis，不存单机内存当真理

• [ ] refresh_token 失效走重新 OAuth，不人工贴 token

• [ ] 空值过滤、ASCII 排序、secret 首尾、MD5 大写

• [ ] timestamp 毫秒

• [ ] 改参数必重新签名

• [ ] 单 Key 留余量（10→8，20→16）

• [ ] 多实例用 Redis 滑动窗口

• [ ] 图搜/批量爬取单独限频

• [ ] 429 长退避，不 Sleep(0) 狂重试

• [ ] 非核心同步放凌晨，白天降速

• [ ] 网络/限流可重试，业务错误不重试

• [ ] token 401 最多刷 1 次再试

• [ ] 死信队列接 403 / 下架 / SKU 失效

十、一句话收口

AppKey/Secret 保身份，access_token 保授权，refresh_token 保续命，QPS 保你不被封。

1688 生产客户端的成熟度 =

密钥不落地 + token 前置刷新 + 按接口分级限流 + 429 长退避 + 401 只刷一次 + 业务错误进死信。

把“能调通”当成“能上线”，是 1688 ERP 最贵的幻觉。