1688 item_search 接口对接全攻略：从入门到精通

import requestsimport hashlibimport timeimport jsonclass AlibabaSearchApi:
    def __init__(self, app_key, app_secret):
        """初始化API客户端"""
        self.app_key = app_key
        self.app_secret = app_secret        # 接口地址
        self.token_url = "https://gw.open.1688.com/openapi/http/1/system.oauth2/getToken"  # 令牌接口
        self.search_url = "https://gw.open.1688.com/openapi/param2/2/portals.open.api.item.search"  # 搜索接口
        # 令牌缓存（避免频繁请求令牌）
        self.access_token = None
        self.token_expire_at = 0  # 令牌过期时间（时间戳，秒）

    def generate_sign(self, params):
        """生成签名（核心方法）"""
        # 1. 按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接参数（key+value）
        sign_str = ""
        for key, value in sorted_params:
            # 注意：参数值需转为字符串，避免类型问题
            sign_str += f"{key}{str(value)}"
        # 3. 追加app_secret并MD5加密
        sign_str += self.app_secret
        sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
        return sign    def get_access_token(self):
        """获取/刷新access_token（自动判断有效期）"""
        # 检查令牌是否有效（提前60秒刷新，避免过期）
        current_time = time.time()
        if self.access_token and current_time < self.token_expire_at - 60:
            return self.access_token        # 客户端模式参数
        token_params = {
            "grant_type": "client_credentials",
            "client_id": self.app_key,
            "client_secret": self.app_secret        }

        try:
            response = requests.post(
                url=self.token_url,
                data=token_params,
                timeout=10,
                headers={"Content-Type": "application/x-www-form-urlencoded"}
            )
            response.raise_for_status()  # 触发HTTP错误（如404、500）
            result = response.json()

            # 处理令牌请求错误
            if "error" in result:
                raise Exception(f"令牌请求失败：{result['error_description']}")

            # 缓存令牌和过期时间
            self.access_token = result["access_token"]
            self.token_expire_at = current_time + int(result["expires_in"])
            print(f"令牌获取成功，有效期至：{time.ctime(self.token_expire_at)}")
            return self.access_token        except Exception as e:
            raise Exception(f"获取access_token异常：{str(e)}")

    def item_search(self, keywords, page=1, page_size=20, **kwargs):
        """
        按关键字搜索商品（核心业务方法）
        :param keywords: 搜索关键字（必填）
        :param page: 页码（默认1）
        :param page_size: 每页条数（默认20，最大50）
        :param kwargs: 可选参数（如categoryId、priceStart、sort等）
        :return: 搜索结果（字典）
        """
        # 1. 获取有效令牌
        token = self.get_access_token()

        # 2. 组装公共参数
        common_params = {
            "app_key": self.app_key,
            "access_token": token,
            "method": "portals.open.api.item.search",
            "format": "json",
            "v": "1.0",
            "timestamp": str(int(current_time * 1000)),  # 毫秒级时间戳
            "keywords": keywords,
            "page": page,
            "pageSize": page_size        }

        # 3. 合并可选参数（kwargs）
        all_params = {**common_params, **kwargs}
        # 移除值为None的可选参数（避免无效参数）
        all_params = {k: v for k, v in all_params.items() if v is not None}

        # 4. 生成签名
        all_params["sign"] = self.generate_sign(all_params)

        # 5. 发送搜索请求
        try:
            response = requests.post(
                url=self.search_url,
                data=all_params,
                timeout=15,
                headers={"Content-Type": "application/x-www-form-urlencoded"}
            )
            response.raise_for_status()
            result = response.json()

            # 6. 处理响应
            if "error_response" in result:
                error = result["error_response"]
                return {
                    "success": False,
                    "error_code": error.get("code"),
                    "error_msg": error.get("msg"),
                    "request_id": error.get("requestId")  # 用于1688技术支持排查问题
                }

            # 提取商品数据
            search_data = result["item_search_response"]
            total_items = search_data.get("totalCount", 0)  # 总商品数
            item_list = search_data.get("items", {}).get("item", [])  # 商品列表

            return {
                "success": True,
                "total_count": total_items,
                "page": page,
                "page_size": page_size,
                "total_pages": (total_items + page_size - 1) // page_size,  # 总页数
                "items": item_list            }

        except Exception as e:
            return {
                "success": False,
                "error_msg": f"搜索请求异常：{str(e)}"
            }# ------------------------------# 使用示例# ------------------------------if __name__ == "__main__":
    # 替换为自己的app_key和app_secret
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"

    # 初始化API客户端
    api = AlibabaSearchApi(APP_KEY, APP_SECRET)

    # 搜索“纯棉T恤”，第1页，20条/页，价格10-50元，按销量降序
    search_result = api.item_search(
        keywords="纯棉T恤",
        page=1,
        page