速卖通开放平台接口实战：跨境电商商品检索与详情解析全方案（附多语言处理 + 签名避坑代码）

import requestsimport hashlibimport timeimport osfrom urllib.parse import urlencodedef generate_aliexpress_sign(params, app_secret):    """生成速卖通签名：按ASCII升序排序+MD5加密（核心避坑点）"""    # 1. 排除sign字段，按参数名ASCII升序排序    sorted_params = sorted([(k, str(v)) for k, v in params.items() if k != "sign"])    # 2. 拼接参数（格式：key1=value1&key2=value2）    sign_str = urlencode(sorted_params)    # 3. 拼接App Secret并MD5加密（注意编码为UTF-8）    sign_str += app_secret    return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()def aliexpress_item_search(keyword, category_id, page=1, language="en"):    """速卖通商品检索接口实战：解决多语言乱码+签名失败"""    # 1. 获取环境变量中的密钥（避免硬编码）    app_key = os.getenv("ALIEXPRESS_APP_KEY")    app_secret = os.getenv("ALIEXPRESS_APP_SECRET")    if not app_key or not app_secret:        raise Exception("请先配置ALIEXPRESS_APP_KEY和ALIEXPRESS_APP_SECRET环境变量")        # 2. 构造基础参数    base_params = {        "method": "aliexpress.open.api.item.search",        "app_key": app_key,        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),        "format": "json",        "v": "1.0",        "keyword": keyword,        "category_id": category_id,        "page": page,        "page_size": 30,  # 每页30条，兼顾效率与稳定性        "sort": "sales_desc",  # 销量降序，选品更精准        "language": language    }        # 3. 生成签名    base_params["sign"] = generate_aliexpress_sign(base_params, app_secret)        try:        # 4. 发起请求（超时设5秒，避免卡请求）        response = requests.get(            url="https://api.aliexpress.com/openapi/param2/1/com.aliexpress.open.api",            params=base_params,            timeout=5,            verify=True        )        response.raise_for_status()  # 捕获4xx/5xx错误（如429频次超限）        result = response.json()                # 5. 处理多语言编码（避免乱码）        if "aliexpress_open_api_item_search_response" in result:            data = result["aliexpress_open_api_item_search_response"]["result"]            # 确保商品标题/描述为UTF-8编码            for item in data.get("items", []):                item["title"] = item["title"].encode("latin-1").decode("utf-8")                item["description"] = item["description"].encode("latin-1").decode("utf-8")            return data        else:            raise Exception(f"接口返回异常：{result.get('error_response', {}).get('msg', '未知错误')}")        except requests.exceptions.Timeout:        raise Exception("接口请求超时，建议10秒后重试")    except requests.exceptions.HTTPError as e:        if "429" in str(e):            raise Exception("调用频次超限，建议次日再试或申请提升配额")        else:            raise Exception(f"HTTP错误：{str(e)}")    except Exception as e:        raise Exception(f"检索接口调用失败：{str(e)}")# 调用示例（面向英语市场，检索“wireless earbuds”，类目ID为100003481）if __name__ == "__main__":    try:        search_result = aliexpress_item_search(            keyword="wireless earbuds",            category_id="100003481",            page=1,            language="en"        )        print(f"共检索到{search_result['total_count']}个商品")        for item in search_result["items"][:3]:  # 打印前3个商品            print(f"商品ID：{item['product_id']}，标题：{item['title']}，销量：{item['sales_count']}")    except Exception as e:        print(f"实战提示：{str(e)}")

def aliexpress_item_get(product_id, language="en"):    """速卖通商品详情接口实战：解析SKU+物流信息"""    app_key = os.getenv("ALIEXPRESS_APP_KEY")    app_secret = os.getenv("ALIEXPRESS_APP_SECRET")    if not app_key or not app_secret:        raise Exception("请先配置密钥环境变量")        base_params = {        "method": "aliexpress.open.api.item.get",        "app_key": app_key,        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),        "format": "json",        "v": "1.0",        "product_id": product_id,        "language": language    }    base_params["sign"] = generate_aliexpress_sign(base_params, app_secret)        try:        response = requests.get(            url="https://api.aliexpress.com/openapi/param2/1/com.aliexpress.open.api",            params=base_params,            timeout=8,  # 详情数据多，超时设8秒            verify=True        )        response.raise_for_status()        result = response.json()        item = result["aliexpress_open_api_item_get_response"]["result"]                # 1. 解析SKU：规格组合→库存/价格映射        sku_mapping = {}        for sku in item.get("sku_infos", []):            # 拼接规格（例：“颜色:黑色;尺寸:M”）            spec_combination = ";".join([f"{s['attr_name']}:{s['attr_value']}" for s in sku["spec_attrs"]])            sku_mapping[spec_combination] = {                "price": sku["price"],                "stock": sku["stock"],                "sku_id": sku["sku_id"]            }        item["sku_mapping"] = sku_mapping  # 新增整理后的SKU映射                # 2. 整理物流信息：筛选支持的物流方式        logistics_list = []        for logistics in item.get("logistics_info", []):            if logistics["is_support"]:  # 只保留支持的物流                logistics_list.append({                    "logistics_name": logistics["logistics_name"],                    "shipping_time": logistics["shipping_time"],  # 发货时间                    "fee_type": logistics["fee_type"]  # 运费类型（免费/收费）                })        item["support_logistics"] = logistics_list  # 新增整理后的物流列表                # 3. 多语言字段编码处理        item["title"] = item["title"].encode("latin-1").decode("utf-8")        item["description"] = item["description"].encode("latin-1").decode("utf-8")                return item    except Exception as e:        raise Exception(f"详情接口调用失败：{str(e)}")# 调用示例（获取商品ID为123456789的详情）if __name__ == "__main__":    try:        item_detail = aliexpress_item_get(product_id="123456789", language="en")        print(f"商品标题：{item_detail['title']}")        print(f"好评率：{item_detail['rating_info']['positive_rate']}%")        print(f"支持的物流：{[l['logistics_name'] for l in item_detail['support_logistics']]}")        print(f"SKU映射（前2个）：{dict(list(item_detail['sku_mapping'].items())[:2])}")    except Exception as e:        print(f"实战提示：{str(e)}")