阿里巴巴 1688 关键字搜索接口实战：B2B 批发筛选 + 百川 V2 签名 + 风控熔断（Python 合规版）

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

class AlibabaKeywordSearch:
    def __init__(self, app_key, app_secret, access_token=None):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token  # 公开搜索可空
        self.base_url = "https://gw.open.1688.com/openapi/param2/2/alibaba.offer.search/2.0"
        self.session = requests.Session()
        # 令牌缓存（公开搜索优化）
        self.token_expire = datetime.utcnow() + timedelta(days=30) if access_token else None

    def _generate_sign(self, params):
        """百川V2标准签名（nonce+排序+MD5大写）"""
        # 1. 新增随机串（防重放，网上教程常遗漏）
        params["nonce"] = str(random.randint(100000, 999999))
        # 2. 按ASCII升序排序所有参数（不含sign）
        sorted_items = sorted(params.items(), key=lambda x: x[0])
        # 3. 拼接key+value（无分隔符）
        sign_str = "".join(f"{k}{v}" for k, v in sorted_items)
        # 4. 追加app_secret并MD5加密转大写
        sign_str += self.app_secret
        sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
        return sign, params["nonce"]

    def search(self, keyword, page_no=1, page_size=20, price_start=None, 
               price_end=None, min_order=None, sort_type="total", category_id=None):
        """
        B2B关键字搜索（含批发筛选）
        :param keyword: 搜索关键词
        :param page_no: 页码（≥1）
        :param page_size: 每页数量（≤40）
        :param price_start: 最低批发价
        :param price_end: 最高批发价
        :param min_order: 最小起订量
        :param sort_type: 排序（total销量/price_asc价格升/price_desc价格降）
        :param category_id: 类目ID
        :return: 结构化商品列表
        """
        # 基础公共参数
        params = {
            "app_key": self.app_key,
            "method": "alibaba.offer.search",
            "timestamp": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2.0",
            "format": "json"
        }
        # 业务参数（B2B批发核心）
        params["q"] = keyword.strip()
        params["pageNo"] = page_no
        params["pageSize"] = min(page_size, 40)  # 上限40
        if self.access_token:
            params["access_token"] = self.access_token
        if price_start:
            params["priceStart"] = price_start
        if price_end:
            params["priceEnd"] = price_end
        if min_order:
            params["minQuantity"] = min_order  # 起批量筛选
        if sort_type:
            params["sortType"] = sort_type
        if category_id:
            params["categoryId"] = category_id

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

        try:
            # 风控限速（关键防封）
            time.sleep(0.2)
            resp = self.session.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", {})
            item_list = data.get("itemList", [])
            # 结构化解析（仅保留核心批发字段）
            product_list = []
            for item in item_list:
                product_list.append({
                    "offer_id": item.get("offerId"),
                    "title": item.get("subject"),
                    "main_image": item.get("imageUrl"),
                    "price": item.get("price"),  # 批发价
                    "min_order": item.get("minOrderQuantity"),  # 起批量
                    "sales": item.get("saleQuantity"),  # 销量
                    "supplier": item.get("company", {}).get("companyName"),
                    "category_id": item.get("categoryId")
                })

            return {
                "code": 200,
                "msg": "success",
                "data": {
                    "total": data.get("total", 0),
                    "page_no": page_no,
                    "page_size": page_size,
                    "product_list": product_list
                }
            }
        except Exception as e:
            return {"code": 500, "msg": f"异常：{str(e)}"}

# ———— 调用示例 ————
if __name__ == "__main__":
    api = AlibabaKeywordSearch(
        app_key="你的AppKey",
        app_secret="你的AppSecret",
        access_token="你的AccessToken"  # 公开搜索可注释
    )
    # 搜索"女装T恤"，销量排序，价格10-50，起批量≥2
    res = api.search(
        keyword="女装T恤",
        page_no=1,
        page_size=20,
        price_start=10,
        price_end=50,
        min_order=2,
        sort_type="total"
    )
    print(json.dumps(res, ensure_ascii=False, indent=2))