阿里巴巴 1688 商品详情接口实战：B2B 批发数据全解析 + 签名优化 + 异常熔断（Python 合规版）

import requests
import hashlib
import time
import urllib.parse
from datetime import datetime, timedelta

class AlibabaProductDetail:
    def __init__(self, app_key, app_secret, refresh_token=None):
        self.app_key = app_key
        self.app_secret = app_secret
        self.refresh_token = refresh_token
        self.access_token = None
        self.token_expire = None
        self.base_url = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.product.get"
        # 初始化token
        if refresh_token:
            self._refresh_token()

    def _refresh_token(self):
        """AccessToken自动刷新（有效期2小时）"""
        url = "https://gw.open.1688.com/auth/token"
        data = {
            "grant_type": "refresh_token",
            "refresh_token": self.refresh_token,
            "app_key": self.app_key,
            "app_secret": self.app_secret
        }
        try:
            resp = requests.post(url, data=data, timeout=10)
            token_data = resp.json()
            self.access_token = token_data.get("access_token")
            self.token_expire = datetime.utcnow() + timedelta(seconds=token_data.get("expires_in", 7200) - 60)
        except Exception as e:
            raise Exception(f"Token刷新失败：{str(e)}")

    def _generate_sign(self, params):
        """HMAC-SHA256标准签名（官方2.0规范）"""
        # 参数按ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 拼接参数并URL编码
        sign_str = "&".join(f"{k}={urllib.parse.quote_plus(str(v))}" for k, v in sorted_params)
        # 追加app_secret并加密
        sign_str += f"&secret={self.app_secret}"
        sign = hashlib.sha256(sign_str.encode("utf-8")).hexdigest().upper()
        return sign

    def get_detail(self, product_id, fields=None):
        """
        获取商品详情（含批发价、SKU、规格）
        :param product_id: 商品ID（URL提取，如619899292404）
        :param fields: 按需返回字段，默认全量
        :return: 结构化商品数据
        """
        # Token过期检查
        if self.refresh_token and datetime.utcnow() >= self.token_expire:
            self._refresh_token()

        # 基础参数
        params = {
            "app_key": self.app_key,
            "method": "alibaba.product.get",
            "timestamp": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2.0",
            "format": "json",
            "access_token": self.access_token
        }
        # 业务参数
        params["productId"] = product_id
        if fields:
            params["fields"] = fields

        # 生成签名
        params["sign"] = self._generate_sign(params)

        try:
            resp = requests.post(self.base_url, data=params, timeout=15)
            result = resp.json()

            if not result.get("success"):
                return {"code": -1, "msg": f"接口错误：{result.get('errorMsg')}", "error": result}

            data = result.get("result", {})
            # 结构化解析B2B核心数据
            product_info = {
                "product_id": data.get("productId"),
                "title": data.get("subject"),
                "brand": data.get("brand", {}).get("name", ""),
                "main_image": data.get("imageUrl", ""),
                "price_ranges": data.get("priceRanges", []),  # 批发阶梯价
                "moq": data.get("minOrderQuantity", 0),  # 最小起订量
                "sku_list": data.get("skuInfo", {}).get("skuList", []),  # SKU规格库存
                "supplier": {
                    "company_name": data.get("company", {}).get("companyName", ""),
                    "credit_rating": data.get("company", {}).get("creditRating", "")
                }
            }
            time.sleep(0.3)
            return {"code": 200, "msg": "success", "data": product_info}
        except Exception as e:
            return {"code": 500, "msg": f"异常：{str(e)}"}

# ———— 调用示例 ————
if __name__ == "__main__":
    api = AlibabaProductDetail(
        app_key="你的AppKey",
        app_secret="你的AppSecret",
        refresh_token="你的RefreshToken"
    )
    # 按需指定字段，减少数据体积
    fields = "subject,imageUrl,priceRanges,minOrderQuantity,skuInfo,company"
    res = api.get_detail(product_id="619899292404", fields=fields)
    print(json.dumps(res, ensure_ascii=False, indent=2))