实战解析：阿里巴巴国际站 item_search 关键字搜索商品API接口

pip install requests>=2.31.0
pip install urllib3>=2.0.0

params = {
    "q": "bluetooth speaker",  # 英文关键词
    "country": "US",           # 目标美国市场
    "min_price": 5.0,
    "max_price": 50.0,
    "sort": "sales_desc",
    "page": 1,
    "page_size": 30,
    "fields": "product_id,title,price,min_order,image_url,supplier_id,gold_supplier"}

import hashlibimport hmacimport timeimport jsonimport requestsfrom urllib.parse import quote, urlencodefrom typing import Dict, Optionalclass AlibabaSearchAPI:
    """阿里巴巴国际站搜索API客户端"""
    
    def __init__(self, app_key: str, app_secret: str, access_token: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.api_url = "https://api.alibaba.com/openapi/rest"
        self.last_call_time = 0  # 限流控制
        
    def _generate_sign(self, params: Dict[str, str]) -> str:
        """
        生成HMAC-SHA256签名（跨境合规版）
        规则：app_secret + 排序后的key1value1key2value2... + app_secret
        """
        # 1. 过滤空值和sign字段
        valid_params = {k: v for k, v in params.items() 
                       if v is not None and k != "sign"}
        
        # 2. 按键名ASCII码升序排序
        sorted_params = sorted(valid_params.items(), key=lambda x: x[0])
        
        # 3. 拼接签名字符串
        sign_content = self.app_secret        for key, value in sorted_params:
            # 值需要是字符串，且进行UTF-8编码
            value_str = str(value).encode('utf-8').decode('utf-8')
            sign_content += f"{key}{value_str}"
        sign_content += self.app_secret        
        # 4. HMAC-SHA256加密并转大写
        sign = hmac.new(
            self.app_secret.encode('utf-8'),
            sign_content.encode('utf-8'),
            hashlib.sha256        ).hexdigest().upper()
        
        return sign    
    def _rate_limit(self):
        """限流控制：QPS≤2（国际站要求）[^50^]"""
        import time
        current_time = time.time()
        interval = current_time - self.last_call_time        if interval < 0.8:  # 设置0.8秒间隔
            sleep_time = 0.8 - interval            print(f"⏳ 触发限流，等待 {sleep_time:.2f} 秒...")
            time.sleep(sleep_time)
        self.last_call_time = time.time()
    
    def search_products(
        self,
        keywords: str,
        country: str = "US",
        min_price: float = None,
        max_price: float = None,
        page: int = 1,
        page_size: int = 30,
        sort: str = "relevance_desc"
    ) -> Optional[Dict]:
        """
        关键字搜索商品（跨境合规版）
        """
        # 限流控制
        self._rate_limit()
        
        # 多语言关键词编码（关键步骤）[^50^]
        encoded_keywords = quote(keywords, encoding='utf-8')
        
        # 构造请求参数
        params = {
            "method": "alibaba.item_search",
            "app_key": self.app_key,
            "access_token": self.access_token,
            "keywords": encoded_keywords,
            "country": country,
            "currency": "USD",  # 统一货币单位
            "sort": sort,
            "page": page,
            "page_size": page_size,
            "fields": "product_id,title,price,min_order,image_url,supplier_id,"
                      "supplier_name,gold_supplier,verified,country,response_rate",
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "3.0"
        }
        
        # 可选参数
        if min_price is not None:
            params["min_price"] = min_price        if max_price is not None:
            params["max_price"] = max_price        
        # 生成签名
        params["sign"] = self._generate_sign(params)
        
        try:
            print(f"🚀 开始搜索: {keywords} (目标市场: {country}, 页码: {page})")
            response = requests.get(
                self.api_url,
                params=params,
                headers={"Content-Type": "application/json"},
                timeout=20  # 跨境请求超时设为20秒[^50^]
            )
            response.raise_for_status()
            
            result = response.json()
            
            # 错误处理
            if result.get("code") != 0:
                print(f"❌ API错误: {result.get('msg')} (code: {result.get('code')})")
                return None
            
            # 数据脱敏（GDPR合规）[^50^]
            return self._desensitize_data(result.get("result", {}))
            
        except Exception as e:
            print(f"❌ 请求异常: {e}")
            return None
    
    def _desensitize_data(self, data: Dict) -> Dict:
        """
        跨境数据脱敏处理（符合GDPR）
        1. 移除联系方式
        2. 昵称脱敏
        """
        products = data.get("products", [])
        for product in products:
            # 移除敏感字段
            product.pop("supplier_contact", None)
            product.pop("supplier_phone", None)
            
            # 昵称脱敏：前3字符 + *** + 后2字符
            supplier_name = product.get("supplier_name", "")
            if len(supplier_name) > 6:
                product["supplier_name"] = supplier_name[:3] + "***" + supplier_name[-2:]
        
        return data# 使用示例if __name__ == "__main__":
    # 配置凭证（需替换为真实值）
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    ACCESS_TOKEN = "your_access_token"
    
    api = AlibabaSearchAPI(APP_KEY, APP_SECRET, ACCESS_TOKEN)
    
    # 实战搜索：美国市场的蓝牙音箱
    result = api.search_products(
        keywords="bluetooth speaker",
        country="US",
        min_price=10.0,
        max_price=100.0,
        page=1,
        page_size=30,
        sort="sales_desc"
    )
    
    if result:
        print(f"\n✅ 搜索成功，返回 {len(result.get('products', []))} 个商品")
        # 打印第一个商品示例
        if result["products"]:
            first_item = result["products"][0]
            print(json.dumps(first_item, ensure_ascii=False, indent=2))

{
  "code": 0,  # 0表示成功
  "msg": "success",
  "request_id": "your_request_id",
  "result": {
    "total_count": 1250,  # 总商品数
    "page_size": 30,
    "current_page": 1,
    "total_pages": 42,
    "products": [
      {
        "product_id": "60869458489",
        "title": "China Factory Outdoor Driving Casual Shoes Men",
        "price": "$5.70-$12.50",  # FOB价格区间
        "min_order": 10,  # MOQ
        "image_url": "https://sc02.alicdn.com/...jpg_220x220.jpg",
        "supplier_id": "2206514639625",
        "supplier_name": "Yiwu Junbang Shoe***",  # 脱敏后
        "gold_supplier": True,
        "verified": True,
        "country": "CN",
        "response_rate": 95.5,  # 响应率
        "trade_info": {
          "30day_transactions": 500,
          "repeat_buyer_rate": 0.35
        }
      }
    ]
  }}

def crawl_all_pages(api: AlibabaSearchAPI, keywords: str, country: str, max_pages: int = 10):
    """
    分页抓取所有商品（自动停止）
    """
    all_products = []
    page = 1
    
    while page <= max_pages:
        result = api.search_products(
            keywords=keywords,
            country=country,
            page=page,
            page_size=30
        )
        
        if not result or not result.get("products"):
            print("⚠️ 无更多数据或请求失败")
            break
        
        products = result["products"]
        all_products.extend(products)
        
        print(f"📄 第 {page} 页抓取完成，累计 {len(all_products)} 个商品")
        
        # 判断是否为最后一页
        if page >= result.get("total_pages", 0):
            break
        
        page += 1
        time.sleep(1)  # 分页间隔
    
    return all_products# 使用示例all_data = crawl_all_pages(api, "wireless earbuds", "DE", max_pages=5)print(f"🎯 总计抓取 {len(all_data)} 个德国市场商品")

def multi_language_search(api: AlibabaSearchAPI, base_keyword: str, languages: list):
    """
    多语言并行搜索（如英语、西班牙语、阿拉伯语）
    """
    results = {}
    for lang in languages:
        # 使用翻译API转换关键词（示例简化）
        translated_keyword = f"{base_keyword} ({lang})"  # 实际应调用翻译服务
        result = api.search_products(
            keywords=translated_keyword,
            country="US",  # 目标市场固定
            page_size=20
        )
        results[lang] = result    return results

import scheduledef monitor_product(api: AlibabaSearchAPI, product_id: str, check_interval: int = 3600):
    """
    监控商品价格和库存变化
    """
    def job():
        # 调用item_get接口获取详情（需另实现）
        detail = api.get_item_detail(product_id)
        if detail:
            price = detail.get("price_info", {}).get("price_range")
            stock = detail.get("stock", 0)
            print(f"🔄 [{datetime.now()}] 价格: {price}, 库存: {stock}")
    
    # 每小时执行一次
    schedule.every(check_interval).seconds.do(job)
    
    while True:
        schedule.run_pending()
        time.sleep(1)# 启动监控# monitor_product(api, "60869458489")

from tenacity import retry, stop_after_attempt, wait_exponential@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))def robust_search(api: AlibabaSearchAPI, **kwargs):
    """带重试的搜索函数"""
    return api.search_products(**kwargs)# 使用result = robust_search(api, keywords="led light", country="UK")

def find_niche_products(api: AlibabaSearchAPI, category: str, max_competition: float):
    """
    挖掘低竞争高销量蓝海市场
    """
    result = api.search_products(
        keywords=category,
        sort="sales_desc",
        page_size=50
    )
    
    niche_products = []
    for product in result.get("products", []):
        # 计算竞争指数 = 销量 / 供应商数量
        competition = product.get("trade_info", {}).get("competition_index", 0)
        if competition < max_competition:
            niche_products.append(product)
    
    return niche_products

def analyze_suppliers(products: list):
    """
    分析供应商分布和响应能力
    """
    from collections import Counter
    
    countries = [p.get("country") for p in products]
    gold_rate = sum([p.get("gold_supplier") for p in products]) / len(products)
    avg_response = sum([p.get("response_rate", 0) for p in products]) / len(products)
    
    return {
        "country_distribution": Counter(countries),
        "gold_supplier_rate": f"{gold_rate:.2%}",
        "avg_response_rate": f"{avg_response:.2f}%"
    }

import pandas as pddef export_to_excel(products: list, filename: str):
    """
    导出商品数据到Excel
    """
    df_data = []
    for p in products:
        df_data.append({
            "商品ID": p.get("product_id"),
            "标题": p.get("title"),
            "价格区间": p.get("price"),
            "MOQ": p.get("min_order"),
            "供应商": p.get("supplier_name"),
            "金牌供应商": "是" if p.get("gold_supplier") else "否",
            "国家": p.get("country"),
            "30天销量": p.get("trade_info", {}).get("30day_transactions", 0)
        })
    
    df = pd.DataFrame(df_data)
    df.to_excel(filename, index=False)
    print(f"📊 数据已导出至 {filename}")