阿里巴巴开放平台关键字搜索商品接口实战详解：OAuth2.0 认证落地 + 检索效率优化（附避坑代码）

import requests
import time
import json
import os
from datetime import datetime
class AlibabaAuthHandler:
    """阿里巴巴OAuth2.0认证处理器（自动授权+令牌刷新）"""
    def __init__(self, app_key, app_secret, redirect_uri, cache_dir="./ali_auth_cache"):
        self.app_key = app_key
        self.app_secret = app_secret
        self.redirect_uri = redirect_uri
        self.auth_url = "https://oauth.1688.com/authorize"
        self.token_url = "https://oauth.1688.com/token"
        self.cache_dir = cache_dir
        self.token_file = os.path.join(cache_dir, "ali_access_token.json")
        self.token_info = None  # 存储令牌信息（access_token/expires_time等）
        # 初始化：创建缓存目录+加载已有令牌
        os.makedirs(cache_dir, exist_ok=True)
        self._load_token()
    def _load_token(self):
        """加载已保存的令牌，判断是否过期"""
        if os.path.exists(self.token_file):
            try:
                with open(self.token_file, "r", encoding="utf-8") as f:
                    self.token_info = json.load(f)
                # 令牌有效期提前300秒刷新（避免临界点过期）
                if int(time.time()) < self.token_info["expires_time"] - 300:
                    print("✅ 加载有效令牌，剩余有效期：{}分钟".format(
                        (self.token_info["expires_time"] - time.time())//60
                    ))
                    return True
                else:
                    print("⌛ 令牌已过期，尝试自动刷新")
                    return self._refresh_token()  # 自动刷新令牌
            except Exception as e:
                print(f"❌ 加载令牌失败：{str(e)}，需重新授权")
        return False
    def _refresh_token(self):
        """自动刷新令牌（无需用户干预）"""
        if not self.token_info or "refresh_token" not in self.token_info:
            return False
        params = {
            "grant_type": "refresh_token",
            "client_id": self.app_key,
            "client_secret": self.app_secret,
            "refresh_token": self.token_info["refresh_token"]
        }
        try:
            response = requests.post(self.token_url, data=params, timeout=15)
            new_token = response.json()
            if "error" in new_token:
                raise Exception(f"刷新失败：{new_token['error_description']}")
            # 更新令牌信息
            self.token_info = {
                "access_token": new_token["access_token"],
                "expires_in": new_token["expires_in"],
                "refresh_token": new_token["refresh_token"],
                "expires_time": int(time.time()) + new_token["expires_in"]
            }
            self._save_token()  # 保存新令牌
            print("✅ 令牌刷新成功，新有效期：{}小时".format(new_token["expires_in"]//3600))
            return True
        except Exception as e:
            print(f"❌ 令牌刷新失败：{str(e)}，需重新授权")
            return False
    def _save_token(self):
        """保存令牌到本地缓存（避免重复授权）"""
        with open(self.token_file, "w", encoding="utf-8") as f:
            json.dump(self.token_info, f, ensure_ascii=False, indent=2)
    def get_auth_url(self):
        """生成授权URL，引导用户在浏览器完成授权"""
        params = {
            "response_type": "code",
            "client_id": self.app_key,
            "redirect_uri": self.redirect_uri,
            "state": "ali_search_api_auth"  # 自定义状态值，防CSRF
        }
        auth_url = f"{self.auth_url}?{requests.compat.urlencode(params)}"
        print(f"\n🔗 请在浏览器打开以下URL完成授权：\n{auth_url}")
        print(f"📌 授权后会跳转至：{self.redirect_uri}?code=XXX&state=XXX，请复制URL中的'code'参数")
        return auth_url
    def authorize_with_code(self, auth_code):
        """用授权码获取令牌（首次授权时调用）"""
        params = {
            "grant_type": "authorization_code",
            "client_id": self.app_key,
            "client_secret": self.app_secret,
            "code": auth_code,
            "redirect_uri": self.redirect_uri
        }
        try:
            response = requests.post(self.token_url, data=params, timeout=15)
            token_data = response.json()
            if "error" in token_data:
                raise Exception(f"授权失败：{token_data['error_description']}")
            # 存储令牌信息
            self.token_info = {
                "access_token": token_data["access_token"],
                "expires_in": token_data["expires_in"],
                "refresh_token": token_data["refresh_token"],
                "expires_time": int(time.time()) + token_data["expires_in"]
            }
            self._save_token()
            print(f"✅ 授权成功！令牌有效期：{token_data['expires_in']//3600}小时")
            return True
        except Exception as e:
            print(f"❌ 授权失败：{str(e)}")
            return False
    def get_valid_token(self):
        """获取有效令牌（自动处理加载/刷新）"""
        if not self._load_token():
            # 需手动授权
            self.get_auth_url()
            auth_code = input("请输入授权后获取的'code'：").strip()
            if not self.authorize_with_code(auth_code):
                raise Exception("❌ 未获取有效令牌，无法调用接口")
        return self.token_info["access_token"]

import hmac
import hashlib
import urllib.parse
from collections import OrderedDict
class AlibabaSignGenerator:
    """阿里巴巴接口签名生成器（严格遵循阿里规范）"""
    @staticmethod
    def generate_sign(params, app_secret):
        """
        生成签名：参数排序→URL编码→HMAC-SHA256加密
        :param params: 参数字典（含公共参数+业务参数）
        :param app_secret: 应用Secret
        :return: 签名字符串（大写）
        """
        # 1. 移除已存在的sign参数（避免重复计算）
        if "sign" in params:
            del params["sign"]
        # 2. 按参数名ASCII升序排序（关键！阿里签名对顺序敏感）
        # 用OrderedDict确保排序后顺序不变
        sorted_params = OrderedDict(sorted(params.items(), key=lambda x: x[0]))
        # 3. 拼接为"key=URL编码值"格式（URL编码需保留大写字母，如%3D而非%3d）
        sign_str = "&".join([
            f"{k}={urllib.parse.quote(str(v), safe='', encoding='utf-8')}" 
            for k, v in sorted_params.items()
        ])
        # 4. HMAC-SHA256加密（密钥为app_secret，结果转大写）
        hmac_obj = hmac.new(
            app_secret.encode("utf-8"),
            sign_str.encode("utf-8"),
            hashlib.sha256
        )
        return hmac_obj.hexdigest().upper()

import requests
import time
from datetime import datetime
class AlibabaKeywordSearchClient:
    """阿里巴巴关键字搜索商品接口客户端（高效+容错）"""
    def __init__(self, app_key, app_secret, redirect_uri):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://api.1688.com/router/json"
        self.method = "alibaba.aliindex.search"
        self.version = "1.0"
        self.max_page_size = 50  # 阿里限制每页最大50条
        self.request_interval = 1.2  # 请求间隔（秒），避免429频率超限（≤1次/秒）
        # 初始化认证处理器
        self.auth_handler = AlibabaAuthHandler(app_key, app_secret, redirect_uri)
        # 获取有效令牌
        self.access_token = self.auth_handler.get_valid_token()
    def _get_common_params(self):
        """生成接口公共参数（每次请求必传）"""
        return {
            "app_key": self.app_key,
            "method": self.method,
            "format": "json",
            "sign_method": "hmac-sha256",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),  # 严格格式
            "version": self.version,
            "access_token": self.access_token
        }
    def search_single_page(self, keywords, **kwargs):
        """
        单页搜索：获取指定页码的商品数据
        :param keywords: 搜索关键词
        :param kwargs: 筛选参数（page/page_size/price_start等）
        :return: 清洗后的单页数据
        """
        # 1. 拼接公共参数+业务参数
        params = self._get_common_params()
        params["keywords"] = keywords  # 必传关键词
        # 2. 处理筛选参数（校验合法性）
        valid_filters = ["page", "page_size", "price_start", "price_end", 
                        "category_id", "sort", "is_tmall", "is_taobao", "location"]
        for filter_key in valid_filters:
            if filter_key in kwargs and kwargs[filter_key] is not None:
                if filter_key == "page_size":
                    # 限制page_size≤50
                    params[filter_key] = min(kwargs[filter_key], self.max_page_size)
                elif filter_key == "page":
                    # 页码≥1
                    params[filter_key] = max(int(kwargs[filter_key]), 1)
                else:
                    params[filter_key] = kwargs[filter_key]
        # 3. 生成签名
        params["sign"] = AlibabaSignGenerator.generate_sign(params, self.app_secret)
        # 4. 发送请求（处理异常）
        try:
            response = requests.post(
                self.api_url,
                data=params,
                headers={
                    "Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
                },
                timeout=20  # 超时时间（避免卡请求）
            )
            response.raise_for_status()  # 触发4xx/5xx错误
            result = response.json()
            # 5. 处理阿里错误响应
            if "error_response" in result:
                error = result["error_response"]
                raise Exception(f"接口错误[{error['code']}]：{error['msg']}")
            # 6. 返回原始结果（后续统一清洗）
            return result.get("alibaba_aliindex_search_response", {})
        except requests.exceptions.RequestException as e:
            raise Exception(f"网络错误：{str(e)}")
        except Exception as e:
            raise Exception(f"搜索失败：{str(e)}")
    def search_with_pagination(self, keywords, max_pages=None, **kwargs):
        """
        分页搜索：自动获取多页数据（避免手动翻页）
        :param keywords: 搜索关键词
        :param max_pages: 最大获取页数（None=获取全部）
        :param kwargs: 筛选参数
        :return: 整合后的所有商品数据+统计信息
        """
        all_products = []
        current_page = 1
        total_pages = 1
        total_count = 0
        try:
            # 1. 先获取第一页，拿到总页数和总数量
            first_page_result = self.search_single_page(keywords, page=current_page, **kwargs)
            if not first_page_result or "result" not in first_page_result:
                raise Exception("未获取到搜索结果")
            first_page_data = first_page_result["result"]
            total_count = first_page_data.get("total_results", 0)
            total_pages = first_page_data.get("total_pages", 1)
            page_size = first_page_data.get("page_size", 20)
            # 2. 处理最大页数限制
            if max_pages and total_pages > max_pages:
                total_pages = max_pages
            print(f"🔍 关键词「{keywords}」搜索结果：共{total_count}个商品，{total_pages}页（每页{page_size}条）")
            # 3. 清洗第一页商品数据
            first_page_products = self._clean_product_data(first_page_data.get("products", []))
            all_products.extend(first_page_products)
            print(f"✅ 已获取第{current_page}页，累计{len(all_products)}个商品")
            # 4. 循环获取后续页面
            current_page += 1
            while current_page <= total_pages:
                # 控制请求频率（避免429）
                time.sleep(self.request_interval)
                try:
                    page_result = self.search_single_page(
                        keywords, page=current_page, page_size=page_size, **kwargs
                    )
                    if not page_result or "result" not in page_result:
                        print(f"⚠️  第{current_page}页获取失败，重试1次")
                        time.sleep(self.request_interval)
                        # 重试1次
                        page_result = self.search_single_page(
                            keywords, page=current_page, page_size=page_size, **kwargs
                        )
                        if not page_result:
                            print(f"❌ 第{current_page}页重试失败，跳过")
                            current_page += 1
                            continue
                    # 清洗当前页数据
                    page_data = page_result["result"]
                    page_products = self._clean_product_data(page_data.get("products", []))
                    all_products.extend(page_products)
                    print(f"✅ 已获取第{current_page}页，累计{len(all_products)}个商品")
                except Exception as e:
                    print(f"⚠️  第{current_page}页处理异常：{str(e)}，跳过")
                current_page += 1
            # 5. 返回整合结果
            return {
                "keyword": keywords,
                "total_count": total_count,
                "collected_count": len(all_products),
                "page_size": page_size,
                "total_pages": total_pages,
                "products": all_products,
                "search_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }
        except Exception as e:
            print(f"❌ 分页搜索异常：{str(e)}")
            # 即使异常，返回已获取的商品数据
            return {
                "keyword": keywords,
                "collected_count": len(all_products),
                "products": all_products,
                "error_msg": str(e)
            }
    def _clean_product_data(self, raw_products):
        """清洗商品数据（格式统一+异常值处理）"""
        cleaned_products = []
        for product in raw_products:
            # 处理数值类型（避免字符串导致的计算错误）
            price = float(product.get("price", 0)) if product.get("price") else 0.0
            sales = int(product.get("sales", 0)) if product.get("sales") else 0
            # 处理图片URL（补全HTTPS）
            img_url = product.get("image_url", "")
            if img_url.startswith("//"):
                img_url = "https:" + img_url
            # 提取价格区间（批发价）
            price_range = product.get("price_range", "")
            min_price = max_price = 0.0
            if "-" in price_range:
                try:
                    min_price_str, max_price_str = price_range.split("-")
                    min_price = float(min_price_str.strip())
                    max_price = float(max_price_str.strip())
                except:
                    pass  # 价格区间格式异常，不处理
            # 整理清洗后的商品数据
            cleaned_products.append({
                "product_id": product.get("product_id", ""),  # 商品唯一ID
                "title": product.get("title", "").strip(),   # 商品标题（去空格）
                "price": price,                              # 单价
                "price_range": price_range,                  # 批发价区间
                "min_price": min_price,                      # 最低批发价
                "max_price": max_price,                      # 最高批发价
                "sales": sales,                              # 30天销量
                "company_name": product.get("company_name", ""),  # 供应商名称
                "company_location": product.get("company_location", ""),  # 供应商地区
                "credit": product.get("credit", ""),         # 供应商信用等级
                "image_url": img_url,                        # 商品主图
                "link": product.get("link", ""),             # 商品详情页链接
                "is_tmall": product.get("is_tmall", False)   # 是否天猫商品
            })
        return cleaned_products

def alibaba_search_demo():
    # --------------------------
    # 1. 替换为你的应用信息（必改）
    # --------------------------
    APP_KEY = "你的App Key"
    APP_SECRET = "你的App Secret"
    REDIRECT_URI = "你的授权回调地址（HTTPS，与开放平台一致）"
    # 2. 初始化搜索客户端
    search_client = AlibabaKeywordSearchClient(APP_KEY, APP_SECRET, REDIRECT_URI)
    # 3. 定义搜索参数（按需求调整）
    KEYWORD = "蓝牙耳机 批发"  # 搜索关键词
    SEARCH_FILTERS = {
        "sort": "volume_desc",    # 按销量降序（选品优先）
        "price_start": 50,        # 最低50元（排除低价劣质品）
        "price_end": 200,         # 最高200元（控制采购成本）
        "location": "广东",       # 广东供应商（就近采购）
        "page_size": 50,          # 每页50条（最大）
    }
    MAX_PAGES = 3  # 最多获取3页（按需调整）
    # 4. 执行分页搜索
    print("📢 开始搜索关键词：{}".format(KEYWORD))
    search_result = search_client.search_with_pagination(
        keywords=KEYWORD,
        max_pages=MAX_PAGES,
        **SEARCH_FILTERS
    )
    # 5. 处理搜索结果（导出CSV示例）
    if search_result and search_result["products"]:
        print(f"\n🎉 搜索完成！共获取{len(search_result['products'])}个商品")
        # 导出到CSV
        import csv
        csv_filename = f"ali_search_{KEYWORD.replace(' ', '_')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        with open(csv_filename, "w", encoding="utf-8-sig", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=[
                "product_id", "title", "price", "price_range", "min_price", 
                "max_price", "sales", "company_name", "company_location", "credit"
            ])
            writer.writeheader()
            writer.writerows(search_result["products"])
        print(f"📊 结果已导出至：{csv_filename}")
        # 简单分析：销量Top5商品
        top5_products = sorted(search_result["products"], key=lambda x: x["sales"], reverse=True)[:5]
        print("\n🏆 销量Top5商品：")
        for i, product in enumerate(top5_products, 1):
            print(f"{i}. 标题：{product['title']} | 销量：{product['sales']}件 | 价格：{product['price']}元 | 供应商：{product['company_name']}")
    else:
        print(f"\n❌ 未获取到有效商品数据：{search_result.get('error_msg', '未知错误')}")
if __name__ == "__main__":
    alibaba_search_demo()