from aliexpress_api import AliexpressApi# 初始化 API 客户端api = AliexpressApi(
    app_key="你的App Key",
    app_secret="你的App Secret",
    access_token="你的Access Token",
    language="en_US")def get_product_detail(product_id: str) -> dict:
    """
    查询速卖通商品详情
    :param product_id: 速卖通商品ID（数字串，如1005005808863025）
    :return: 商品详情字典
    """
    try:
        response = api.execute(
            method="aliexpress.solution.product.detail.get",
            params={
                "product_id": product_id,
                "language": "en",
                "currency": "USD"
            }
        )
        return response    except Exception as e:
        print(f"查询商品详情失败：{e}")
        return {}# 测试调用if __name__ == "__main__":
    test_product_id = "1005005808863025"
    detail = get_product_detail(test_product_id)
    if detail and detail.get("code") == 200:
        product_info = detail.get("data", {})
        print("商品标题：", product_info.get("product_title"))
        print("商品价格：", product_info.get("sale_price"))
        print("商品主图：", product_info.get("main_image_url"))
        print("库存数量：", product_info.get("stock_quantity"))
        print("商品描述：", product_info.get("product_description"))
    else:
        print("获取商品详情失败，响应：", detail)

import timeimport hashlibimport requestsfrom urllib.parse import urlencode, quote_plusdef generate_sign(params: dict, app_secret: str) -> str:
    """
    生成速卖通API签名（MD5）
    :param params: 请求参数（不含sign）
    :param app_secret: 应用Secret
    :return: 签名字符串
    """
    # 1. 参数按ASCII升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 2. 拼接为key=value格式，无分隔符
    sign_str = app_secret    for k, v in sorted_params:
        if v is not None and v != "":
            sign_str += f"{k}{v}"
    sign_str += app_secret    # 3. MD5加密并转大写
    sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    return signdef ali_api_request(method: str, params: dict, app_key: str, app_secret: str, access_token: str, gateway: str) -> dict:
    """
    原生发送速卖通API请求
    :param method: 接口方法名
    :param params: 业务参数
    :param app_key: App Key
    :param app_secret: App Secret
    :param access_token: access_token
    :param gateway: API网关地址
    :return: 接口响应
    """
    # 1. 构造公共参数
    common_params = {
        "app_key": app_key,
        "method": method,
        "format": "json",
        "v": "2.0",
        "timestamp": str(int(time.time() * 1000)),  # 毫秒级时间戳
        "sign_method": "md5",
        "access_token": access_token    }
    # 2. 合并公共参数和业务参数
    all_params = {**common_params, **params}
    # 3. 生成签名
    all_params["sign"] = generate_sign(all_params, app_secret)
    # 4. 发送GET请求
    try:
        response = requests.get(
            url=gateway,
            params=all_params,
            timeout=15
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败：{e}")
        return {}# 测试原生调用（商品详情）if __name__ == "__main__":
    APP_KEY = "你的App Key"
    APP_SECRET = "你的App Secret"
    ACCESS_TOKEN = "你的access_token"
    GATEWAY = "https://api-sg.aliexpress.com/sync"  # 新加坡节点，国内可用

    result = ali_api_request(
        method="aliexpress.solution.product.detail.get",
        params={
            "product_id": "1005005808863025",
            "language": "en",
            "currency": "USD"
        },
        app_key=APP_KEY,
        app_secret=APP_SECRET,
        access_token=ACCESS_TOKEN,
        gateway=GATEWAY    )
    print("原生请求响应：", result)

import requestsimport timeimport hashlibfrom requests.adapters import HTTPAdapterfrom urllib3.util.retry import Retry# 自行替换开放平台密钥APP_KEY = "你的APP_KEY"APP_SECRET = "你的APP_SECRET"ACCESS_TOKEN = "你的ACCESS_TOKEN"API_URL = "https://api-sg.aliexpress.com/sync"  # 新加坡节点，国内可用class AliExpressItemDetailApi:
    def __init__(self, app_key, app_secret, access_token):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.session = self._build_session()
        self.last_request_time = 0  # 频率控制

    def _build_session(self):
        # 自动重试机制，提升接口稳定性
        retry = Retry(total=3, backoff_factor=0.5, status_forcelist=[429, 500, 503])
        session = requests.Session()
        session.mount("https://", HTTPAdapter(max_retries=retry))
        return session    def _make_sign(self, params):
        # 速卖通官方签名规则（网上90%写错）
        sorted_items = sorted(params.items(), key=lambda x: x[0])
        plain = self.app_secret        for k, v in sorted_items:
            if v:
                plain += f"{k}{v}"
        plain += self.app_secret        return hashlib.md5(plain.encode('utf-8')).hexdigest().upper()

    def get_item_detail(self, item_id, language="en"):
        # 频率控制：免费版QPS=2，间隔至少0.5秒
        current_time = time.time()
        if current_time - self.last_request_time < 0.5:
            time.sleep(0.5)
        self.last_request_time = current_time
        
        timestamp = str(int(time.time()))
        
        # 组装请求参数
        params = {
            "method": "aliexpress.item.get",
            "app_key": self.app_key,
            "access_token": self.access_token,
            "timestamp": timestamp,
            "format": "json",
            "v": "2.0",
            "item_id": item_id,
            "language": language,
            # 全字段获取，覆盖跨境电商核心需求
            "fields": "title,price,original_price,image_url,sku_property_list,logistics_info,seller_info,evaluation_info,promotion_info"
        }
        
        # 生成签名
        params["sign"] = self._make_sign(params)
        
        try:
            resp = self.session.get(API_URL, params=params, timeout=15)
            result = resp.json()
            
            # 错误判断
            if result.get("code") != 0:
                return {"success": False, "msg": result.get("msg", "接口异常")}
            
            # 核心数据解析与清洗
            data = result.get("result", {})
            cleaned_data = {
                "商品ID": data.get("item_id"),
                "多语言标题": data.get("title"),
                "售价": data.get("price"),
                "原价": data.get("original_price"),
                "主图链接": data.get("image_url"),
                "SKU规格": data.get("sku_property_list", []),
                "物流信息": data.get("logistics_info", {}),
                "卖家信息": data.get("seller_info", {}),
                "评价统计": data.get("evaluation_info", {}),
                "促销信息": data.get("promotion_info", {}),
                "商品链接": f"https://www.aliexpress.com/item/{item_id}.html"
            }
            return {"success": True, "data": cleaned_data}
            
        except Exception as e:
            return {"success": False, "msg": f"请求异常：{str(e)}"}# 调用示例if __name__ == "__main__":
    api = AliExpressItemDetailApi(APP_KEY, APP_SECRET, ACCESS_TOKEN)
    # 替换为真实商品ID
    res = api.get_item_detail("1005005586923234", language="en")
    
    if res["success"]:
        print("✅ 商品详情获取成功")
        print(f"商品标题：{res['data']['多语言标题']}")
        print(f"售价：{res['data']['售价']}")
        print(f"物流信息：{res['data']['物流信息']}")
    else:
        print(f"❌ {res['msg']}")

from tenacity import retry, stop_after_attempt, wait_fixed@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))def get_product_detail_with_retry(product_id: str):
    return get_product_detail(product_id)

def refresh_access_token(refresh_token: str, app_key: str, app_secret: str) -> dict:
    url = "https://api.aliexpress.com/system/oauth2/token"
    params = {
        "grant_type": "refresh_token",
        "client_id": app_key,
        "client_secret": app_secret,
        "refresh_token": refresh_token    }
    response = requests.post(url, params=params)
    return response.json()