利用速卖通API 实现商品详情实时数据采集（开发接入示例）

import requests
import hashlib
import time
import json
from urllib.parse import urlencode

class AliexpressAPI:
    def __init__(self, app_key, app_secret, access_token=None):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.gateway_url = "https://gw.api.alibaba.com/openapi/param2/1/com.alibaba.product/aliexpress.open.api.product.get"
    
    def generate_signature(self, params):
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = self.app_secret + ''.join([f"{k}{v}" for k, v in sorted_params])
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def get_product_detail(self, product_id):
        timestamp = str(int(time.time() * 1000))
        
        common_params = {
            "method": "aliexpress.open.api.product.get",
            "appKey": self.app_key,
            "timestamp": timestamp,
            "format": "json",
            "v": "2.0",
            "signMethod": "md5"
        }
        
        if self.access_token:
            common_params["session"] = self.access_token
            
        api_params = {
            "productId": product_id,
            "fields": "product_id,title,category_id,price,original_price,short_description,description,images"
        }
        
        all_params = {**common_params, **api_params}
        all_params["sign"] = self.generate_signature(all_params)
        
        try:
            response = requests.post(self.gateway_url, data=all_params)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"API请求异常: {e}")
            return None

def save_to_file(data, filename):
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

if __name__ == "__main__":
    # 请替换为您的实际应用凭证
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    ACCESS_TOKEN = "your_access_token"  # 如果需要
    
    api_client = AliexpressAPI(APP_KEY, APP_SECRET, ACCESS_TOKEN)
    product_id = "1005001234567890"  # 替换为实际商品ID
    
    product_data = api_client.get_product_detail(product_id)
    if product_data:
        save_to_file(product_data, f"product_{product_id}.json")
        print(f"商品数据已保存至 product_{product_id}.json")    

一、速卖通 API 接入概述

在跨境电商数据分析场景中，实时获取商品详情数据是一项核心需求。速卖通作为全球知名的跨境电商平台，提供了完善的 API 接口体系，允许开发者通过程序化方式获取商品、订单、物流等多维度数据。本文将详细介绍如何接入速卖通官方 API 实现商品详情数据的实时采集。

二、开发前准备

1. 注册开发者账号

   • 访问速卖通平台
   • 完成账号注册

2. 创建应用

   • 创建需求应用
   • 获取 ApiKey 和 ApiSecret（用于 API 调用身份验证）
   • 提交应用审核，获取相应 API 权限

3. API 文档阅读

   • 熟悉 API 调用规范
   • 重点关注商品详情获取接口 (aliexpress.open.api.product.get)

三、API 接入核心流程

速卖通 API 采用 RESTful 风格设计，主要通过 HTTP POST 方式调用。接入流程包括：

1. 身份认证机制

   • 采用 AppKey+AppSecret 进行签名认证
   • 部分接口需要 OAuth2.0 授权（需提前获取 AccessToken）

2. 签名生成规则

   • 将所有请求参数（包括公共参数和业务参数）按参数名 ASCII 码从小到大排序
   • 拼接参数名和参数值（不包含特殊符号）
   • 在拼接字符串前加上 AppSecret
   • 对拼接后的字符串进行 MD5 加密并转为大写

3. 请求频率限制

   • 不同 API 接口有不同的 QPS 限制
   • 需合理控制请求频率，避免触发限流机制

四、代码实现详解

代码部分包含两个核心模块：

1. API 客户端实现(aliexpress_api_client.py)

   • 封装 API 调用基础功能
   • 实现签名生成算法
   • 处理 API 请求和响应

2. 批量采集脚本(batch_product_collector.py)

   • 实现多商品 ID 的批量采集
   • 加入请求间隔控制
   • 完善的错误处理和日志记录

五、数据存储与后续处理

采集到的商品数据以 JSON 格式存储，方便后续处理。典型的数据处理流程包括：

1. 数据清洗：去除冗余字段，处理缺失值
2. 数据存储：存入关系型数据库或 NoSQL 数据库
3. 数据分析：进行价格趋势分析、竞品对比等
4. 数据可视化：生成图表展示商品关键指标

六、注意事项与最佳实践

1. API 调用注意事项

   • 妥善保管 AppKey 和 AppSecret
   • 注意 API 版本变化，及时调整代码
   • 处理好网络异常和 API 限流情况

2. 性能优化建议

   • 使用异步请求提高采集效率
   • 实现数据缓存机制，避免重复请求
   • 采用分布式架构提升采集规模

3. 合规使用 API

   • 遵守速卖通 API 使用条款
   • 控制数据使用范围，避免数据滥用
   • 定期审核数据采集和使用流程

通过以上步骤，开发者可以实现一个稳定、高效的速卖通商品数据采集系统，为跨境电商业务决策提供有力支持。