在跨境電商數(shù)據(jù)采集場(chǎng)景中，通過(guò)關(guān)鍵字精準(zhǔn)搜索商品列表是基礎(chǔ)且核心的需求。本文將聚焦阿里巴巴國(guó)際站的關(guān)鍵字搜索接口，詳細(xì)介紹如何構(gòu)建多條件搜索請(qǐng)求、處理分頁(yè)數(shù)據(jù)、解析商品列表信息，并提供可直接復(fù)用的 Python 實(shí)現(xiàn)方案，幫助開(kāi)發(fā)者快速搭建穩(wěn)定高效的商品搜索功能。

一、阿里巴巴搜索 API 基礎(chǔ)信息

阿里巴巴國(guó)際站提供的alibaba.product.search接口是實(shí)現(xiàn)關(guān)鍵字搜索商品列表的核心接口，支持多維度篩選條件組合，滿足不同場(chǎng)景的搜索需求。

接口特點(diǎn)：

基于 TOP 開(kāi)放平臺(tái)架構(gòu)，采用統(tǒng)一的簽名認(rèn)證機(jī)制

支持復(fù)雜條件組合搜索（關(guān)鍵字、價(jià)格、銷量、評(píng)分等）

分頁(yè)加載數(shù)據(jù)，最大頁(yè)容量為 50 條

響應(yīng)包含商品基本信息、價(jià)格、賣家、銷量等核心數(shù)據(jù)

接口端點(diǎn)：https://gw.api.alibaba.com/openapi/param2/2.0/aliexpress.open/api.searchproduct

二、核心搜索參數(shù)詳解

1. 公共參數(shù)

app_key：應(yīng)用唯一標(biāo)識(shí)

method：接口名稱，固定為alibaba.product.search

timestamp：請(qǐng)求時(shí)間戳（yyyy-MM-dd HH:mm:ss）

format：響應(yīng)格式，默認(rèn) JSON

v：API 版本，固定為 2.0

sign：請(qǐng)求簽名

partner_id：合作伙伴 ID（可選）

2. 核心搜索參數(shù)

keywords：搜索關(guān)鍵字（必填）

page_no：頁(yè)碼，默認(rèn) 1

page_size：每頁(yè)條數(shù)（1-50）

min_price/max_price：價(jià)格區(qū)間篩選

sort_type：排序方式（price_asc/price_desc/sales_desc/rating_desc）

category_id：分類 ID 篩選

trade_assurance：是否僅保價(jià)商品（true/false）

shipping_country：目標(biāo)配送國(guó)家

3. 響應(yīng)數(shù)據(jù)結(jié)構(gòu)

total_results：總搜索結(jié)果數(shù)

page_no/page_size：分頁(yè)信息

products：商品列表數(shù)組

filters：可用篩選條件（用于前端篩選項(xiàng)展示）

三、完整代碼實(shí)現(xiàn)

以下是 Python 實(shí)現(xiàn)的阿里巴巴國(guó)際站關(guān)鍵字搜索功能，支持多條件篩選、分頁(yè)控制和結(jié)果結(jié)構(gòu)化處理：

阿里巴巴國(guó)際站關(guān)鍵字搜索接口實(shí)現(xiàn)

import requests
import time
import hashlib
import hmac
import urllib.parse
from typing import Dict, List, Optional, Any
from datetime import datetime

class AlibabaSearchAPI:
    def __init__(self, app_key: str, app_secret: str):
        """
        初始化阿里巴巴商品搜索API客戶端
        
        :param app_key: 應(yīng)用的App Key
        :param app_secret: 應(yīng)用的App Secret
        """
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://gw.api.alibaba.com/openapi/param2/2.0/aliexpress.open/api.searchproduct"
        self.max_page_size = 50  # 最大每頁(yè)條數(shù)
        
    def _generate_sign(self, params: Dict[str, str]) -> str:
        """
        生成API請(qǐng)求簽名
        
        :param params: 請(qǐng)求參數(shù)字典
        :return: 簽名字符串
        """
        # 按參數(shù)名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        
        # 拼接參數(shù)并URL編碼
        query_string = "&".join([
            f"{k}={urllib.parse.quote(str(v), safe='')}" 
            for k, v in sorted_params
        ])
        
        # 計(jì)算HMAC-SHA1簽名
        sign_str = self.app_secret + query_string + self.app_secret
        signature = hmac.new(
            self.app_secret.encode('utf-8'),
            sign_str.encode('utf-8'),
            hashlib.sha1
        ).hexdigest().upper()
        
        return signature
    
    def search_products(self,
                       keywords: str,
                       page_no: int = 1,
                       page_size: int = 20,
                       min_price: Optional[float] = None,
                       max_price: Optional[float] = None,
                       sort_type: Optional[str] = None,
                       category_id: Optional[str] = None,
                       trade_assurance: Optional[bool] = None,
                       shipping_country: Optional[str] = None) -> Dict[str, Any]:
        """
        搜索阿里巴巴國(guó)際站商品
        
        :param keywords: 搜索關(guān)鍵字
        :param page_no: 頁(yè)碼
        :param page_size: 每頁(yè)條數(shù)(1-50)
        :param min_price: 最低價(jià)格
        :param max_price: 最高價(jià)格
        :param sort_type: 排序方式
        :param category_id: 分類ID
        :param trade_assurance: 是否僅保價(jià)商品
        :param shipping_country: 配送國(guó)家
        :return: 搜索結(jié)果
        """
        # 限制最大頁(yè)容量
        page_size = min(page_size, self.max_page_size)
        
        # 公共參數(shù)
        params: Dict[str, str] = {
            "app_key": self.app_key,
            "method": "alibaba.product.search",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "keywords": keywords,
            "page_no": str(page_no),
            "page_size": str(page_size)
        }
        
        # 添加可選參數(shù)
        if min_price is not None:
            params["min_price"] = str(min_price)
        if max_price is not None:
            params["max_price"] = str(max_price)
        if sort_type:
            params["sort_type"] = sort_type
        if category_id:
            params["category_id"] = category_id
        if trade_assurance is not None:
            params["trade_assurance"] = "true" if trade_assurance else "false"
        if shipping_country:
            params["shipping_country"] = shipping_country
        
        # 生成簽名
        params["sign"] = self._generate_sign(params)
        
        try:
            # 發(fā)送請(qǐng)求
            response = requests.get(
                self.api_url,
                params=params,
                timeout=15
            )
            response.raise_for_status()
            
            # 解析響應(yīng)
            result = response.json()
            
            # 處理API錯(cuò)誤
            if "error_response" in result:
                error = result["error_response"]
                return {
                    "success": False,
                    "error_code": error.get("code"),
                    "error_msg": error.get("msg")
                }
                
            # 處理正常響應(yīng)
            search_result = result.get("aliexpress_open_api_searchproduct_response", {}).get("result", {})
            return self._parse_search_result(search_result)
            
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "error_msg": f"請(qǐng)求異常: {str(e)}"
            }
        except Exception as e:
            return {
                "success": False,
                "error_msg": f"處理響應(yīng)失敗: {str(e)}"
            }
    
    def _parse_search_result(self, raw_result: Dict[str, Any]) -> Dict[str, Any]:
        """
        解析原始搜索結(jié)果為結(jié)構(gòu)化數(shù)據(jù)
        
        :param raw_result: 原始API響應(yīng)數(shù)據(jù)
        :return: 結(jié)構(gòu)化的搜索結(jié)果
        """
        # 提取分頁(yè)信息
        pagination = {
            "total_results": raw_result.get("total_results", 0),
            "page_no": raw_result.get("page_no", 1),
            "page_size": raw_result.get("page_size", 20),
            "total_pages": (raw_result.get("total_results", 0) + 
                           raw_result.get("page_size", 20) - 1) // 
                           raw_result.get("page_size", 20)
        }
        
        # 解析商品列表
        products = []
        for item in raw_result.get("products", []):
            # 提取價(jià)格信息
            price_ranges = []
            if "price_ranges" in item:
                for pr in item["price_ranges"]:
                    price_ranges.append({
                        "min_qty": pr.get("min_qty"),
                        "max_qty": pr.get("max_qty"),
                        "price": pr.get("price")
                    })
            
            products.append({
                "product_id": item.get("product_id"),
                "title": item.get("title"),
                "main_image": item.get("main_image_url"),
                "price": {
                    "min_price": item.get("min_price"),
                    "max_price": item.get("max_price"),
                    "currency": item.get("currency_code"),
                    "ranges": price_ranges
                },
                "seller": {
                    "user_id": item.get("user_id"),
                    "company_name": item.get("company_name"),
                    "country": item.get("country"),
                    "response_rate": item.get("response_rate"),
                    "transaction_level": item.get("transaction_level")
                },
                "sales": {
                    "monthly_sales": item.get("monthly_sales"),
                    "feedback_count": item.get("feedback_count"),
                    "rating": item.get("rating")
                },
                "trade_assurance": item.get("trade_assurance") == "true",
                "category_id": item.get("category_id"),
                "category_name": item.get("category_name"),
                "url": item.get("product_detail_url")
            })
        
        return {
            "success": True,
            "pagination": pagination,
            "products": products,
            "available_filters": self._parse_filters(raw_result.get("filters", []))
        }
    
    def _parse_filters(self, raw_filters: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """解析可用篩選條件"""
        filters = []
        for flt in raw_filters:
            options = []
            for opt in flt.get("options", []):
                options.append({
                    "value": opt.get("value"),
                    "label": opt.get("label"),
                    "count": opt.get("count")
                })
            
            filters.append({
                "name": flt.get("name"),
                "label": flt.get("label"),
                "options": options
            })
        return filters
    
    def search_all_pages(self,
                        keywords: str,
                        max_pages: int = 5,
                        **kwargs) -> Dict[str, Any]:
        """
        批量獲取多頁(yè)搜索結(jié)果
        
        :param keywords: 搜索關(guān)鍵字
        :param max_pages: 最大獲取頁(yè)數(shù)
        :param**kwargs: 其他搜索參數(shù)
        :return: 合并的搜索結(jié)果
        """
        all_products = []
        current_page = 1
        total_pages = 1
        
        while current_page <= max_pages and current_page <= total_pages:
            # 搜索當(dāng)前頁(yè)
            result = self.search_products(
                keywords=keywords,
                page_no=current_page,
                **kwargs
            )
            
            if not result.get("success"):
                return result
                
            # 收集商品
            all_products.extend(result.get("products", []))
            
            # 更新分頁(yè)信息
            pagination = result.get("pagination", {})
            total_pages = pagination.get("total_pages", 1)
            
            # 準(zhǔn)備下一頁(yè)
            current_page += 1
            
            # 添加請(qǐng)求間隔，避免觸發(fā)頻率限制
            time.sleep(1)
        
        return {
            "success": True,
            "total_products": len(all_products),
            "products": all_products,
            "original_pagination": pagination
        }

# 使用示例
if __name__ == "__main__":
    # 替換為你的應(yīng)用憑證
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    
    # 初始化API客戶端
    search_api = AlibabaSearchAPI(APP_KEY, APP_SECRET)
    
    # 示例1：基礎(chǔ)搜索
    basic_result = search_api.search_products(
        keywords="smart watch",
        page_no=1,
        page_size=20,
        sort_type="sales_desc",
        min_price=10,
        max_price=100,
        trade_assurance=True
    )
    
    if basic_result["success"]:
        print(f"基礎(chǔ)搜索: 找到 {basic_result['pagination']['total_results']} 個(gè)商品")
        print(f"第一頁(yè)商品數(shù)量: {len(basic_result['products'])}")
        if basic_result["products"]:
            print(f"第一個(gè)商品: {basic_result['products'][0]['title']}")
    
    # 示例2：批量獲取多頁(yè)結(jié)果
    multi_page_result = search_api.search_all_pages(
        keywords="bluetooth earbuds",
        max_pages=3,
        page_size=50,
        shipping_country="US"
    )
    
    if multi_page_result["success"]:
        print(f"n多頁(yè)搜索: 共獲取 {multi_page_result['total_products']} 個(gè)商品")

四、代碼核心功能解析

1.** 多條件搜索實(shí)現(xiàn) **：

支持關(guān)鍵字、價(jià)格區(qū)間、排序方式等 10 + 篩選條件

自動(dòng)處理參數(shù)邊界（如頁(yè)容量限制在 1-50 之間）

提供清晰的參數(shù)類型定義，便于開(kāi)發(fā)使用

2.** 分頁(yè)策略優(yōu)化 **：

基礎(chǔ)方法search_products處理單頁(yè)搜索

高級(jí)方法search_all_pages自動(dòng)批量獲取多頁(yè)數(shù)據(jù)

內(nèi)置請(qǐng)求間隔控制，避免觸發(fā)頻率限制

3.** 數(shù)據(jù)解析增強(qiáng) **：

結(jié)構(gòu)化處理原始響應(yīng)，提取核心業(yè)務(wù)字段

分類組織商品信息（基礎(chǔ)信息、價(jià)格、賣家、銷量等）

解析可用篩選條件，便于前端實(shí)現(xiàn)篩選功能

4.** 錯(cuò)誤處理機(jī)制 **：

捕獲 HTTP 請(qǐng)求異常，提供詳細(xì)錯(cuò)誤信息

解析 API 返回的錯(cuò)誤碼和描述

統(tǒng)一返回格式，包含成功標(biāo)識(shí)和業(yè)務(wù)數(shù)據(jù)

五、實(shí)戰(zhàn)注意事項(xiàng)

1.** 接口權(quán)限與限制 **：

alibaba.product.search接口需要在開(kāi)放平臺(tái)申請(qǐng)使用權(quán)限

免費(fèi)開(kāi)發(fā)者賬號(hào)有調(diào)用頻率限制（通常 QPS=10）

部分篩選條件（如品牌篩選）需要額外權(quán)限

2.** 搜索策略優(yōu)化 **：

合理設(shè)置page_size，減少請(qǐng)求次數(shù)（最大 50 條 / 頁(yè)）

優(yōu)先使用篩選條件縮小范圍，再分頁(yè)獲取

熱門關(guān)鍵詞搜索建議添加緩存，緩存時(shí)間 15-30 分鐘

3.** 數(shù)據(jù)處理建議 **：

商品標(biāo)題可能包含 HTML 實(shí)體，需進(jìn)行轉(zhuǎn)義處理

價(jià)格數(shù)據(jù)需結(jié)合貨幣代碼進(jìn)行轉(zhuǎn)換和展示

批量獲取時(shí)實(shí)現(xiàn)斷點(diǎn)續(xù)傳，應(yīng)對(duì)網(wǎng)絡(luò)中斷

4.** 反爬與合規(guī) **：

嚴(yán)格遵守 API 調(diào)用頻率限制，避免 IP 被臨時(shí)封禁

數(shù)據(jù)使用需符合阿里巴巴國(guó)際站的開(kāi)發(fā)者協(xié)議

生產(chǎn)環(huán)境建議使用代理 IP 池分散請(qǐng)求來(lái)源

六、功能擴(kuò)展方向

實(shí)現(xiàn)搜索建議功能，集成阿里巴巴的關(guān)鍵詞推薦 API

開(kāi)發(fā)可視化搜索工具，支持條件組合與結(jié)果導(dǎo)出

添加搜索結(jié)果去重功能，基于商品 ID 過(guò)濾重復(fù)項(xiàng)

集成商品詳情接口，實(shí)現(xiàn) "搜索 - 查看詳情" 完整流程

通過(guò)本文提供的方案，開(kāi)發(fā)者可以快速構(gòu)建功能完善的阿里巴巴國(guó)際站商品搜索模塊，為跨境電商應(yīng)用提供精準(zhǔn)的商品數(shù)據(jù)支持。實(shí)際開(kāi)發(fā)中，建議根據(jù)業(yè)務(wù)需求合理配置篩選條件，平衡數(shù)據(jù)準(zhǔn)確性和接口性能。????

審核編輯 黃宇