1688 开放平台榜单接口实战：多类型榜单切换 + HMAC 签名 + 趋势数据结构化 + 选品分析（附 Python 代码）

import requests
import hmac
import hashlib
import base64
import time
import json
from datetime import datetime

class Ali1688TopListAPI:
    """1688榜单接口：双接口兼容+9大榜单+签名+批发解析+趋势计算"""
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        # 双接口地址
        self.api_url_new = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/product.topList.query/"
        self.api_url_old = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.item/alibaba.item.search.best/"
        self.timeout = 15

    def make_hmac_sign(self, params):
        """1688标准HMAC-SHA1签名：ASCII排序+Base64编码"""
        sorted_items = sorted(params.items())
        sign_str = "".join(f"{k}{v}" for k, v in sorted_items)
        hmac_obj = hmac.new(
            self.app_secret.encode("utf-8"),
            sign_str.encode("utf-8"),
            hashlib.sha1
        )
        return base64.b64encode(hmac_obj.digest()).decode("utf-8")

    def calc_trend_score(self, sale_count, price):
        """计算趋势热度分：销量权重70%+价格优势30%，辅助爆款预判"""
        sale_score = min(sale_count / 1000, 70)
        price_score = max(0, 30 - (price / 10))
        return round(sale_score + price_score, 2)

    def get_top_list(self, rank_type, category_id="", page=1, page_size=20, use_new_api=True):
        """
        获取榜单列表
        :param rank_type: 榜单类型（complex/hot/goodPrice/anchorHot等）
        :param category_id: 类目ID（1-3级，空=全类目）
        :param page: 页码
        :param page_size: 每页数量
        :param use_new_api: True=新类目榜单，False=经典榜单
        :return: 结构化榜单（含批发+趋势）
        """
        timestamp = str(int(time.time() * 1000))
        params = {
            "app_key": self.app_key,
            "timestamp": timestamp,
            "rankType": rank_type,
            "categoryId": category_id,
            "page": str(page),
            "pageSize": str(page_size)
        }
        # 兼容旧接口参数
        if not use_new_api:
            params["sort_type"] = params.pop("rankType")
            params["cat"] = params.pop("categoryId")

        params["sign"] = self.make_hmac_sign(params)
        api_url = self.api_url_new if use_new_api else self.api_url_old

        try:
            resp = requests.post(api_url, data=params, timeout=self.timeout)
            res = resp.json()

            if res.get("error_response"):
                return {"code":-1,"msg":res["error_response"].get("msg","接口调用失败")}

            # 解析新/旧接口数据
            if use_new_api:
                data = res.get("alibaba_product_top_list_query_response",{}).get("result",{})
            else:
                data = res.get("alibaba_item_search_best_response",{}).get("result",{})
            goods_list = data.get("itemList",[])
            total = data.get("total",0)

            items = []
            for idx, g in enumerate(goods_list, 1):
                price = float(g.get("price",0))
                sale_count = int(g.get("saleCount",0))
                items.append({
                    "rank": (page-1)*page_size + idx,  # 榜单排名
                    "offer_id": g.get("offerId"),
                    "title": g.get("subject"),
                    "price": price,
                    "min_order": g.get("minOrder",1),  # B2B起批量
                    "sale_count": sale_count,  # 销量
                    "trend_score": self.calc_trend_score(sale_count, price),  # 趋势分
                    "supplier": g.get("sellerNick"),
                    "is_gold": g.get("isGoldSupplier"),  # 实力商家
                    "main_img": g.get("imageUrl"),
                    "detail_url": g.get("detailUrl")
                })

            time.sleep(1.5)  # 防风控间隔
            return {"code":200,"msg":"成功","total":total,"items":items}

        except Exception as e:
            return {"code":500,"msg":f"请求异常：{str(e)}"}

# 调用示例
if __name__ == "__main__":
    API_KEY = "你的app_key"
    API_SECRET = "你的app_secret"
    # 类目ID：如女装1级=100001，可在1688开放平台查询
    CATEGORY_ID = "100001"

    api = Ali1688TopListAPI(API_KEY, API_SECRET)
    # 新接口-热销榜（hot），经典接口传use_new_api=False
    result = api.get_top_list(rank_type="hot", category_id=CATEGORY_ID)
    print(json.dumps(result, ensure_ascii=False, indent=2))