速卖通关键词搜索 API 实战：多币种适配 + 跨境物流筛选 + 智能分页 + 签名容错（Python 合规版）

import requests
import hmac
import hashlib
import time
import json
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class AliExpressSearch:
    def __init__(self, app_key, app_secret, access_token, region="sg"):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        # 多区域网关适配
        self.gateways = {
            "sg": "https://api-sg.aliexpress.com/sync",
            "cn": "https://gw-api.alibaba.com/router/rest",
            "us": "https://api-us.aliexpress.com/sync"
        }
        self.api_url = self.gateways.get(region, self.gateways["sg"])
        # 重试配置：429限流/5xx异常自动重试
        self.session = requests.Session()
        retry_cfg = Retry(total=3, backoff_factor=0.5, status_forcelist=[429,500,502])
        self.session.mount("https://", HTTPAdapter(max_retries=retry_cfg))

    def create_sign(self, params):
        """速卖通标准HMAC-SHA256签名：ASCII排序+密钥首尾拼接"""
        # 过滤空参数，避免签名失败
        sorted_items = sorted({k:v for k,v in params.items() if v}.items())
        sign_str = self.app_secret
        for k, v in sorted_items:
            sign_str += f"{k}{v}"
        sign_str += self.app_secret
        return hmac.new(self.app_secret.encode(), sign_str.encode(), hashlib.sha256).hexdigest().upper()

    def calc_score(self, sale30, rating, logistics_days, price_diff):
        """跨境选品打分：销量40%+好评率30%+物流20%+价格优势10%"""
        s_score = min(sale30/1000*40,40)
        r_score = min(rating*6,30)
        l_score = max(20 - logistics_days, 0)
        p_score = min(price_diff*2,10)
        return round(s_score+r_score+l_score+p_score,2)

    def search_item(self, keyword, page=1, page_size=20, min_price=None, max_price=None,
                     currency="USD", ship_from="CN", max_logistics_days=30):
        """
        跨境关键词搜索
        :param keyword: 搜索关键词（英文优先）
        :param page: 页码
        :param page_size: 每页条数（最大50）
        :param min_price: 最低价格
        :param max_price: 最高价格
        :param currency: 币种 USD/EUR/RUB
        :param ship_from: 发货地 CN(中国)/US(美国)/EU(欧洲)
        :param max_logistics_days: 最大物流时效
        """
        # 公共参数（严格按文档）
        params = {
            "method": "aliexpress.item.search",
            "app_key": self.app_key,
            "timestamp": str(int(time.time()*1000)),  # 13位毫秒级
            "format": "json",
            "v": "2.0",
            "sign_method": "hmac-sha256",
            "access_token": self.access_token,
            "keywords": keyword,
            "page_no": page,
            "page_size": page_size,
            "currency": currency,
            "ship_from": ship_from
        }
        # 可选参数动态添加
        if min_price: params["min_price"] = min_price
        if max_price: params["max_price"] = max_price

        # 生成签名
        params["sign"] = self.create_sign(params)

        try:
            resp = self.session.post(self.api_url, data=params, timeout=15)
            result = resp.json()

            # 错误处理：令牌过期提示重鉴权
            if result.get("code") != 200:
                return {"code": -1, "msg": result.get("msg", "调用失败"), "error_code": result.get("code")}

            # 解析数据
            res_data = result.get("result", {})
            goods_list = res_data.get("items", [])
            total = res_data.get("total_count", 0)
            res_items = []

            for item in goods_list:
                logistics_days = item.get("logistics_days", 99)
                if logistics_days > max_logistics_days:
                    continue  # 过滤超时效商品
                res_items.append({
                    "item_id": item["item_id"],
                    "title_en": item["title"],
                    "title_zh": item.get("translate_title", ""),
                    "price": item["price"],
                    "currency": currency,
                    "discount_rate": item.get("discount_rate", 0),
                    "sale30": item.get("sale_30d", 0),
                    "rating": item.get("rating", 0),
                    "ship_from": item["ship_from"],
                    "is_overseas_warehouse": item.get("is_overseas_warehouse", False),
                    "logistics_days": logistics_days,
                    "score": self.calc_score(item.get("sale_30d",0), item.get("rating",0),
                                               logistics_days, item.get("discount_rate",0))
                })

            time.sleep(0.5)  # 限流休眠
            return {"code": 200, "total": total, "data": res_items}

        except Exception as e:
            return {"code": -2, "msg": f"请求异常：{str(e)}"}

# ———— 调用示例 ————
if __name__ == "__main__":
    APP_KEY = "你的AppKey"
    APP_SECRET = "你的AppSecret"
    ACCESS_TOKEN = "你的AccessToken"

    # 初始化（区域选sg新加坡，适配全球）
    api = AliExpressSearch(APP_KEY, APP_SECRET, ACCESS_TOKEN, region="sg")
    # 搜索：无线耳机、20-50美元、美国发货、15天内物流
    res = api.search_item("wireless earphone", min_price=20, max_price=50,
                           currency="USD", ship_from="US", max_logistics_days=15)
    print(json.dumps(res, ensure_ascii=False, indent=2))