1688 开放平台关键词搜索商品接口实战：B2B 批发筛选 + HMAC 签名 + 分页稳定获取（附 Python 代码）

import requests
import hmac
import hashlib
import base64
import time
import json

class Ali1688SearchAPI:
    """1688关键词搜索商品列表（官方API·生产级封装）"""
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.uniform.search.goods.query/"
        self.item_ids = set()

    def _make_sign(self, params):
        """1688官方HMAC-SHA1签名（标准正确实现）"""
        sorted_items = sorted(params.items())
        sign_str = "".join([f"{k}{v}" for k, v in sorted_items])
        hmac_digest = hmac.new(
            self.app_secret.encode("utf-8"),
            sign_str.encode("utf-8"),
            hashlib.sha1
        ).digest()
        return base64.b64encode(hmac_digest).decode("utf-8")

    def search_items(self, keyword, page=1, page_size=20):
        """关键词搜索商品（自动去重+B2B结构化）"""
        timestamp = str(int(time.time() * 1000))
        
        # 公共参数
        params = {
            "app_key": self.app_key,
            "timestamp": timestamp,
            "keyword": keyword.strip().replace(" ", ""),
            "page": page,
            "pageSize": page_size
        }
        
        # 生成签名
        params["sign"] = self._make_sign(params)

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

            # 错误判断
            if "error_response" in result:
                return {"code": -1, "msg": "接口调用异常"}

            # 解析商品
            data = result.get("alibaba_uniform_search_goods_query_response", {})
            goods_list = data.get("result", {}).get("items", [])
            total = data.get("result", {}).get("total", 0)

            # 结构化+去重
            items = []
            for g in goods_list:
                offer_id = g.get("offerId")
                if offer_id in self.item_ids:
                    continue
                self.item_ids.add(offer_id)

                items.append({
                    "offer_id": offer_id,
                    "title": g.get("subject"),
                    "price": g.get("price"),
                    "min_order": g.get("minOrderCount"),  # 起批量（B2B核心）
                    "seller_nick": g.get("sellerNick"),
                    "is_trade": g.get("isTrade"),
                    "main_img": g.get("imageUrl"),
                    "detail_url": g.get("detailUrl")
                })

            time.sleep(0.8)
            return {
                "code": 200,
                "msg": "success",
                "total": total,
                "items": items
            }

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

# ———— 调用示例 ————
if __name__ == "__main__":
    APP_KEY = "你的app_key"
    APP_SECRET = "你的app_secret"

    api = Ali1688SearchAPI(APP_KEY, APP_SECRET)
    res = api.search_items(keyword="纯棉卫衣", page=1, page_size=20)
    print(json.dumps(res, ensure_ascii=False, indent=2))