阿里巴巴国际站关键字搜索 API 实战：3 步搞定多语言适配 + 限流破局，询盘量提升 40%

import time
import hashlib
import requests
import urllib.parse
from datetime import datetime
class AlibabaglobalSearchAPI:
    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 = datetime.min  # 限流控制
    def _generate_sign(self, params: dict) -> str:
        """生成跨境合规签名"""
        # 过滤空值并按ASCII排序
        valid_params = {k: v for k, v in params.items() if v is not None}
        sorted_params = sorted(valid_params.items(), key=lambda x: x[0])
        # 拼接签名字符串
        sign_str = self.app_secret
        for k, v in sorted_params:
            value_str = str(v).encode('utf-8').decode('utf-8')
            sign_str += f"{k}{value_str}"
        sign_str += self.app_secret
        # SHA256加密（国际站要求高于淘宝）
        return hashlib.sha256(sign_str.encode()).hexdigest().upper()
    def search_products(self, keywords: str, country: str = "US", page: int = 1) -> dict:
        """跨境合规搜索商品"""
        # 限流控制：QPS≤2
        interval = (datetime.now() - self.last_call).total_seconds()
        if interval < 0.8:
            time.sleep(0.8 - interval)
        # 多语言关键词编码
        encoded_keywords = urllib.parse.quote(keywords, encoding='utf-8')
        params = {
            "method": "alibaba.product.search",
            "app_key": self.app_key,
            "access_token": self.access_token,
            "keywords": encoded_keywords,
            "country": country,
            "currency": "USD",  # 统一货币单位
            "sort": "relevance_desc",
            "page": page,
            "page_size": 30,
            # 显式声明必要字段，避免过度获取
            "fields": "product_id,title,price,min_order,category_id,image_url,country_of_origin",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "1.0"
        }
        params["sign"] = self._generate_sign(params)
        resp = requests.get(self.api_url, params=params, timeout=20)  # 跨境延长超时
        self.last_call = datetime.now()
        return self._desensitize_data(resp.json())
    def _desensitize_data(self, raw_data: dict) -> dict:
        """跨境数据脱敏（适配GDPR）"""
        if "products" not in raw_data.get("result", {}):
            return raw_data
        for product in raw_data["result"]["products"]:
            # 移除可能的敏感信息
            product.pop("supplier_contact", None)
            product.pop("detailed_address", None)
            # 昵称脱敏
            if "supplier_name" in product:
                supplier = product["supplier_name"]
                if len(supplier) > 6:
                    product["supplier_name"] = supplier[:3] + "***" + supplier[-2:]
        return raw_data

{
  "result": {
    "products": [
      {
        "product_id": "628123456789",
        "title": "Wireless Bluetooth Speaker Waterproof Portable",
        "title_cn": "无线蓝牙音箱 防水便携",
        "price": "19.99",
        "currency": "USD",
        "min_order": "10",
        "category_id": "100003",
        "image_url": "xxx.jpg",
        "country_of_origin": "CN",
        "supplier_name": "Shenzhen Tech Co.,Ltd"
      }
    ],
    "total_results": 890,
    "request_id": "xxx"
  }
}

from dataclasses import dataclass
from typing import List, Optional
from datetime import date
@dataclass
class GlobalStandardProduct:
    """国际站商品结构化模型"""
    product_id: str
    title_en: str  # 英文标题
    title_cn: str  # 中文标题
    price_usd: float  # 美元价格
    min_order: int  # 最小起订量
    category_id: str
    image_url: str
    origin_country: str
    supplier_name: str  # 脱敏后供应商名
    keywords_en: List[str]  # 英文关键词
    keywords_cn: List[str]  # 中文关键词
    is_valid: bool  # 是否有效商品
    @classmethod
    def from_raw(cls, raw_product: dict, category_avg_price: float) -> "GlobalStandardProduct":
        """原始数据转结构化模型"""
        # 多语言标题处理
        title_en = raw_product.get("title", "").strip()
        title_cn = raw_product.get("title_cn", "").strip()
        # 价格与起订量标准化
        price_usd = cls._standardize_price(raw_product["price"])
        min_order = cls._standardize_min_order(raw_product["min_order"])
        # 关键词提取
        keywords_en = cls._extract_keywords(title_en, lang="en")
        keywords_cn = cls._extract_keywords(title_cn, lang="cn")
        # 有效性判断
        is_valid = (price_usd >= category_avg_price * 0.5 
                   and min_order <= 1000 
                   and cls._check_image(raw_product.get("image_url")))
        return cls(
            product_id=raw_product["product_id"],
            title_en=title_en,
            title_cn=title_cn,
            price_usd=price_usd,
            min_order=min_order,
            category_id=str(raw_product["category_id"]),
            image_url=raw_product.get("image_url", ""),
            origin_country=raw_product.get("country_of_origin", "CN"),
            supplier_name=cls._desensitize_supplier(raw_product["supplier_name"]),
            keywords_en=keywords_en,
            keywords_cn=keywords_cn,
            is_valid=is_valid
        )
    @staticmethod
    def _standardize_price(price_str: str) -> float:
        """价格标准化（处理范围价）"""
        if "-" in price_str:
            return float(price_str.split("-")[0])  # 取低价端
        return float(price_str)
    @staticmethod
    def _standardize_min_order(order_str: str) -> int:
        """起订量标准化"""
        if not order_str.isdigit():
            return 1  # 默认1件
        return int(order_str)
    @staticmethod
    def _extract_keywords(title: str, lang: str) -> List[str]:
        """多语言关键词提取"""
        if not title:
            return []
        if lang == "cn":
            import jieba.analyse
            return jieba.analyse.extract_tags(title, topK=3, stop_words=True)
        else:
            from nltk.tokenize import word_tokenize
            from nltk.corpus import stopwords
            tokens = word_tokenize(title.lower())
            stop_words = set(stopwords.words('english'))
            keywords = [t for t in tokens if t.isalpha() and t not in stop_words]
            return keywords[:3]
    @staticmethod
    def _check_image(url: str) -> bool:
        """图片有效性校验"""
        if not url:
            return False
        try:
            resp = requests.head(url, timeout=5, allow_redirects=True)
            return resp.status_code == 200 and "image" in resp.headers.get("Content-Type", "")
        except:
            return False
    @staticmethod
    def _desensitize_supplier(name: str) -> str:
        """供应商名脱敏"""
        if len(name) <= 5:
            return name[:2] + "***"
        return name[:4] + "***" + name[-3:]

def optimize_region_search(params: dict, target_market: str) -> dict:
    """区域化搜索参数优化"""
    region_config = {
        "EU": {"filters": "certification:CE,FCC", "sort": "relevance_desc"},
        "SEA": {"filters": "min_order:1-50", "sort": "price_asc"},
        "ME": {"filters": "payment_terms:LC", "sort": "sales_desc"}
    }
    if target_market in region_config:
        params.update(region_config[target_market])
    return params

┌─────────────┐  热点品类数据  ┌─────────────┐  全量商品数据  ┌─────────────┐
│ 本地缓存    │◄────────────►│ Redis缓存   │◄────────────►│ 数据库缓存  │
│（内存，30min）│              │（4小时）     │              │（7天）      │
└─────────────┘              └─────────────┘              └─────────────┘

import redis
from functools import lru_cache
class GlobalProductCache:
    def __init__(self):
        self.redis_client = redis.Redis(host="localhost", port=6379, db=2, decode_responses=True)
        self.local_cache = lru_cache(maxsize=100)  # 热点品类
    def get_cached_products(self, keywords: str, country: str) -> Optional[List[dict]]:
        """缓存查询"""
        cache_key = f"global:search:{country}:{keywords}"
        # 先查本地缓存
        try:
            return self.local_cache[cache_key]
        except KeyError:
            pass
        # 再查Redis
        cached = self.redis_client.get(cache_key)
        if not cached:
            return None
        import json
        return json.loads(cached)
    def set_cached_products(self, keywords: str, country: str, products: List[GlobalStandardProduct]):
        """缓存写入"""
        cache_key = f"global:search:{country}:{keywords}"
        # 序列化
        product_dicts = [
            {k: v for k, v in product.__dict__.items()} for product in products if product.is_valid
        ]
        import json
        serialized = json.dumps(product_dicts)
        # 本地缓存（热点）
        self.local_cache[cache_key] = product_dicts
        # Redis缓存（4小时）
        self.redis_client.setex(cache_key, 14400, serialized)

def cross_border_retry(api_call_func, max_retries: int = 4) -> dict:
    """跨境重试：适配网络波动与限流"""
    retries = 0
    retry_intervals = [1, 3, 5, 10]  # 递增间隔（跨境延迟高）
    while retries < max_retries:
        try:
            result = api_call_func()
            if "error_response" not in result:
                return result
            err_code = result["error_response"]["code"]
            # 限流或网络错误重试，权限错误直接返回
            if err_code not in [10024, 10030]:
                return result
            time.sleep(retry_intervals[retries])
            retries += 1
        except requests.exceptions.Timeout:
            print("跨境超时，重试中...")
            time.sleep(retry_intervals[retries])
            retries += 1
    return {"error": "跨境调用失败"}