速卖通商品采集API全解析：从接口调用到跨境电商数据流搭建
一、为什么需要速卖通商品采集API？
速卖通（AliExpress）作为全球最大的跨境电商平台之一，拥有超过 1.5 亿活跃买家和数以亿计的 SKU。对于跨境电商从业者、反向海淘独立站运营者、选品工具开发者而言，高效获取速卖通商品数据是业务运转的核心基础设施。
传统的人工采集方式存在三大痛点：
| 痛点 | 具体表现 | 影响 |
|---|---|---|
| 效率瓶颈 | 人工复制粘贴，单日最多采集数百条 | 无法规模化 |
| 数据滞后 | 价格、库存变动无法实时同步 | 丢单、客诉 |
| 合规风险 | 爬虫方案容易被反爬机制封禁 | 数据中断、IP被封 |
通过速卖通商品采集API，可以实现毫秒级数据获取、批量采集、实时同步，彻底解决以上问题。
二、速卖通商品采集API核心能力一览
一套完整的速卖通商品采集API通常包含以下接口模块：
2.1 商品搜索接口
根据关键词、类目、价格区间等条件，搜索速卖通平台上的商品列表。
典型请求参数：
GET /api/aliexpress/search
参数说明：
keyword string 搜索关键词，如 "wireless earbuds"
page int 页码，默认 1
page_size int 每页条数，默认 20，最大 50
sort string 排序方式：price_asc / price_desc / orders_desc / rating_desc
min_price float 最低价格（美元）
max_price float 最高价格（美元）
ship_to string 目的国家代码，如 "US"、"RU"
language string 返回语言，如 "en"、"ru"、"pt"返回数据结构：
{
"code"
:
200
,
"message"
:
"success"
,
"data"
:
{
"total"
:
12580
,
"page"
:
1
,
"page_size"
:
20
,
"products"
:
[
{
"product_id"
:
"1005006218491234"
,
"title"
:
"Wireless Earbuds Bluetooth 5.3 TWS"
,
"image_url"
:
"https://ae-pic-a1.aliexpress-media.com/..."
,
"price"
:
3.45
,
"original_price"
:
8.99
,
"discount"
:
"62% OFF"
,
"orders"
:
15234
,
"rating"
:
4.8
,
"store_name"
:
"TechWorld Official Store"
,
"store_id"
:
"12345678"
,
"ship_to_country"
:
"US"
,
"shipping_fee"
:
0.00
,
"product_url"
:
"https://www.aliexpress.com/item/1005006218491234.html"
}
]
}
}2.2 商品详情接口
根据商品ID获取完整的商品详情，包括 SKU 规格属性、多图、描述、物流信息等。
GET /api/aliexpress/product/detail
参数说明：
product_id string 速卖通商品ID
ship_to string 目的国家代码（影响物流费计算）
language string 返回语言返回数据结构：
{
"code"
:
200
,
"message"
:
"success"
,
"data"
:
{
"product_id"
:
"1005006218491234"
,
"title"
:
"Wireless Earbuds Bluetooth 5.3 TWS Earphone"
,
"description"
:
"Product description HTML content..."
,
"main_images"
:
[
"https://ae-pic-a1.aliexpress-media.com/kf/xxx.jpg"
,
"https://ae-pic-a1.aliexpress-media.com/kf/yyy.jpg"
,
"https://ae-pic-a1.aliexpress-media.com/kf/zzz.jpg"
]
,
"sku_props"
:
[
{
"prop_name"
:
"Color"
,
"prop_values"
:
[
{
"name"
:
"Black"
,
"image"
:
"https://..."
}
,
{
"name"
:
"White"
,
"image"
:
"https://..."
}
,
{
"name"
:
"Blue"
,
"image"
:
"https://..."
}
]
}
,
{
"prop_name"
:
"Plug Type"
,
"prop_values"
:
[
{
"name"
:
"US"
,
"image"
:
""
}
,
{
"name"
:
"EU"
,
"image"
:
""
}
]
}
]
,
"skus"
:
[
{
"sku_id"
:
"1005006218491234_001"
,
"prop_path"
:
"Color:Black;Plug Type:US"
,
"price"
:
3.45
,
"stock"
:
9999
,
"sku_image"
:
"https://..."
}
,
{
"sku_id"
:
"1005006218491234_002"
,
"prop_path"
:
"Color:White;Plug Type:EU"
,
"price"
:
3.45
,
"stock"
:
5678
,
"sku_image"
:
"https://..."
}
]
,
"price"
:
3.45
,
"original_price"
:
8.99
,
"orders"
:
15234
,
"rating"
:
4.8
,
"reviews_count"
:
3456
,
"store"
:
{
"store_id"
:
"12345678"
,
"store_name"
:
"TechWorld Official Store"
,
"store_rating"
:
4.9
,
"store_followers"
:
234567
}
,
"shipping"
:
{
"ship_to"
:
"US"
,
"shipping_fee"
:
0.00
,
"shipping_method"
:
"AliExpress Standard Shipping"
,
"estimated_delivery"
:
"15-30 days"
}
,
"product_url"
:
"https://www.aliexpress.com/item/1005006218491234.html"
}
}2.3 店铺商品批量采集接口
根据店铺ID批量获取该店铺下所有商品列表，适用于竞品监控和全店采集场景。
GET /api/aliexpress/store/products
参数说明：
store_id string 速卖通店铺ID
page int 页码
page_size int 每页条数
sort string 排序方式2.4 物流运费查询接口
GET /api/aliexpress/shipping/fee
参数说明：
product_id string 商品ID
sku_id string SKU ID（可选）
ship_to string 目的国家代码
ship_from string 发货国家代码（可选）三、代码实战：用 Python 调用速卖通采集API
3.1 基础封装
import requests
import json
from typing import Optional, List, Dict
class AliExpressAPI:
"""速卖通商品采集API客户端"""
BASE_URL = "https://api.your-domain.com/api/aliexpress"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"User-Agent": "AliExpressAPI-Python/1.0"
})
def search_products(
self,
keyword: str,
page: int = 1,
page_size: int = 20,
sort: str = "orders_desc",
min_price: Optional[float] = None,
max_price: Optional[float] = None,
ship_to: str = "US",
language: str = "en"
) -> Dict:
"""搜索速卖通商品"""
params = {
"keyword": keyword,
"page": page,
"page_size": page_size,
"sort": sort,
"ship_to": ship_to,
"language": language
}
if min_price is not None:
params["min_price"] = min_price
if max_price is not None:
params["max_price"] = max_price
resp = self.session.get(f"{self.BASE_URL}/search", params=params, timeout=30)
resp.raise_for_status()
return resp.json()
def get_product_detail(
self,
product_id: str,
ship_to: str = "US",
language: str = "en"
) -> Dict:
"""获取商品详情"""
params = {
"product_id": product_id,
"ship_to": ship_to,
"language": language
}
resp = self.session.get(f"{self.BASE_URL}/product/detail", params=params, timeout=30)
resp.raise_for_status()
return resp.json()
def get_store_products(
self,
store_id: str,
page: int = 1,
page_size: int = 20,
sort: str = "orders_desc"
) -> Dict:
"""批量获取店铺商品"""
params = {
"store_id": store_id,
"page": page,
"page_size": page_size,
"sort": sort
}
resp = self.session.get(f"{self.BASE_URL}/store/products", params=params, timeout=30)
resp.raise_for_status()
return resp.json()
def get_shipping_fee(
self,
product_id: str,
ship_to: str,
sku_id: Optional[str] = None
) -> Dict:
"""查询物流运费"""
params = {
"product_id": product_id,
"ship_to": ship_to
}
if sku_id:
params["sku_id"] = sku_id
resp = self.session.get(f"{self.BASE_URL}/shipping/fee", params=params, timeout=30)
resp.raise_for_status()
return resp.json()3.2 批量采集实战
import time
from dataclasses import dataclass
@dataclass
class CollectedProduct:
"""采集商品数据模型"""
product_id: str
title: str
price: float
image_url: str
orders: int
rating: float
store_name: str
product_url: str
def batch_collect_by_keyword(
api: AliExpressAPI,
keyword: str,
max_pages: int = 10,
page_size: int = 50,
ship_to: str = "US",
delay: float = 0.5
) -> List[CollectedProduct]:
"""
按关键词批量采集速卖通商品
- max_pages: 最大采集页数
- page_size: 每页条数
- delay: 每页间隔（秒），避免触发限流
"""
all_products = []
for page in range(1, max_pages + 1):
print(f"[采集] 关键词={keyword}, 第 {page}/{max_pages} 页...")
result = api.search_products(
keyword=keyword,
page=page,
page_size=page_size,
ship_to=ship_to
)
if result.get("code") != 200:
print(f"[错误] API返回异常: {result.get('message')}")
break
products = result["data"]["products"]
if not products:
print("[完成] 没有更多商品了")
break
for p in products:
all_products.append(CollectedProduct(
product_id=p["product_id"],
title=p["title"],
price=p["price"],
image_url=p["image_url"],
orders=p.get("orders", 0),
rating=p.get("rating", 0),
store_name=p.get("store_name", ""),
product_url=p["product_url"]
))
print(f" -> 本页采集 {len(products)} 条，累计 {len(all_products)} 条")
time.sleep(delay)
return all_products
def collect_with_sku_details(
api: AliExpressAPI,
product_ids: List[str],
ship_to: str = "US",
delay: float = 0.3
) -> List[Dict]:
"""
批量获取商品详情（含SKU规格）
适用于需要采集完整规格属性的场景
"""
details = []
for i, pid in enumerate(product_ids, 1):
print(f"[详情] 获取商品 {i}/{len(product_ids)}: {pid}")
result = api.get_product_detail(pid, ship_to=ship_to)
if result.get("code") == 200:
details.append(result["data"])
else:
print(f" -> 失败: {result.get('message')}")
time.sleep(delay)
return details3.3 数据存储
import sqlite3
import csv
from pathlib import Path
def save_to_csv(products: List[CollectedProduct], filepath: str):
"""将采集结果保存为CSV"""
with open(filepath, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f)
writer.writerow([
"商品ID", "标题", "价格(USD)", "图片URL",
"销量", "评分", "店铺名称", "商品链接"
])
for p in products:
writer.writerow([
p.product_id, p.title, p.price, p.image_url,
p.orders, p.rating, p.store_name, p.product_url
])
print(f"[保存] CSV文件已生成: {filepath}（共 {len(products)} 条）")
def save_to_sqlite(products: List[CollectedProduct], db_path: str):
"""将采集结果保存到SQLite数据库"""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS aliexpress_products (
product_id TEXT PRIMARY KEY,
title TEXT,
price REAL,
image_url TEXT,
orders INTEGER,
rating REAL,
store_name TEXT,
product_url TEXT,
collected_at TEXT DEFAULT (datetime('now', 'localtime'))
)
""")
for p in products:
cursor.execute("""
INSERT OR REPLACE INTO aliexpress_products
(product_id, title, price, image_url, orders, rating, store_name, product_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
p.product_id, p.title, p.price, p.image_url,
p.orders, p.rating, p.store_name, p.product_url
))
conn.commit()
conn.close()
print(f"[保存] SQLite数据库已更新: {db_path}（共 {len(products)} 条）")3.4 完整调用示例
if __name__ == "__main__":
# 初始化API客户端
api = AliExpressAPI(api_key="YOUR_API_KEY")
# 1. 按关键词批量采集
products = batch_collect_by_keyword(
api=api,
keyword="wireless earbuds",
max_pages=5,
page_size=50,
ship_to="US"
)
# 2. 保存采集结果
save_to_csv(products, "aliexpress_earbuds.csv")
save_to_sqlite(products, "aliexpress_products.db")
# 3. 获取Top 10热销商品的详情（含SKU）
top_10 = sorted(products, key=lambda x: x.orders, reverse=True)[:10]
details = collect_with_sku_details(
api=api,
product_ids=[p.product_id for p in top_10]
)
print(f"\n=== 采集完成 ===")
print(f"商品列表: {len(products)} 条")
print(f"商品详情: {len(details)} 条")
print(f"Top 1 热销: {top_10[0].title} (销量: {top_10[0].orders})")四、典型业务场景
场景一：反向海淘独立站选品
反向海淘（从国内电商平台采购，发货到海外）是近年来快速增长的跨境模式。通过速卖通采集API可以实现：
热销选品：按类目 + 销量排序，快速发现爆款
价格监控：定时采集目标商品价格，自动调价
库存同步：实时获取SKU库存状态，避免超卖
一键上架：采集商品标题、图片、描述，自动同步到独立站
# 反向海淘选品示例：采集3C类目月销>1000的热销品
products = batch_collect_by_keyword(
api=api,
keyword="phone case",
max_pages=20,
page_size=50,
sort="orders_desc",
ship_to="US"
)
# 筛选月销>1000的高潜力商品
hot_products = [p for p in products if p.orders > 1000]场景二：竞品价格监控
def monitor_competitor_prices(api, product_ids, interval=3600):
"""
定时监控竞品价格变化
- interval: 采集间隔（秒），默认1小时
"""
while True:
for pid in product_ids:
detail = api.get_product_detail(pid)
if detail["code"] == 200:
data = detail["data"]
print(f"[监控] {data['title'][:30]}... "
f"当前价格: ${data['price']} "
f"原价: ${data['original_price']} "
f"销量: {data['orders']}")
# 价格低于阈值时告警
if data["price"] < 2.0:
print(f" >>> 价格预警！商品 {pid} 降至 ${data['price']}")
time.sleep(interval)场景三：数据驱动的选品分析
import statistics
def analyze_market(api, keyword, max_pages=10):
"""市场分析：采集数据并输出市场洞察"""
products = batch_collect_by_keyword(api, keyword, max_pages=max_pages)
prices = [p.price for p in products]
orders = [p.orders for p in products]
ratings = [p.rating for p in products]
report = {
"关键词": keyword,
"采集数量": len(products),
"价格区间": f"${min(prices):.2f} - ${max(prices):.2f}",
"平均价格": f"${statistics.mean(prices):.2f}",
"价格中位数": f"${statistics.median(prices):.2f}",
"平均销量": f"{statistics.mean(orders):.0f}",
"最高销量": f"{max(orders)}",
"平均评分": f"{statistics.mean(ratings):.2f}",
}
print("\n=== 市场分析报告 ===")
for k, v in report.items():
print(f" {k}: {v}")
return report五、API调用最佳实践
5.1 限流与重试策略
import time
from functools import wraps
def retry_with_backoff(max_retries=3, base_delay=1.0):
"""
指数退避重试装饰器
当API返回 429（限流）或 5xx 错误时自动重试
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries + 1):
try:
result = func(*args, **kwargs)
# 检查是否被限流
if isinstance(result, dict) and result.get("code") == 429:
delay = base_delay * (2 ** attempt)
print(f"[限流] 第{attempt+1}次重试，等待{delay}秒...")
time.sleep(delay)
continue
return result
except requests.exceptions.RequestException as e:
if attempt < max_retries:
delay = base_delay * (2 ** attempt)
print(f"[网络异常] {e}，{delay}秒后重试...")
time.sleep(delay)
else:
raise
return None
return wrapper
return decorator
# 使用方式
class ResilientAliExpressAPI(AliExpressAPI):
@retry_with_backoff(max_retries=3, base_delay=1.0)
def search_products(self, *args, **kwargs):
return super().search_products(*args, **kwargs)
@retry_with_backoff(max_retries=3, base_delay=1.0)
def get_product_detail(self, *args, **kwargs):
return super().get_product_detail(*args, **kwargs)5.2 异步并发采集
import asyncio
import aiohttp
class AsyncAliExpressAPI:
"""异步速卖通API客户端，适用于大规模采集场景"""
BASE_URL = "https://api.your-domain.com/api/aliexpress"
def __init__(self, api_key: str, max_concurrency: int = 5):
self.api_key = api_key
self.semaphore = asyncio.Semaphore(max_concurrency)
async def get_product_detail(
self,
session: aiohttp.ClientSession,
product_id: str,
ship_to: str = "US"
) -> Dict:
headers = {"Authorization": f"Bearer {self.api_key}"}
params = {"product_id": product_id, "ship_to": ship_to}
async with self.semaphore:
async with session.get(
f"{self.BASE_URL}/product/detail",
headers=headers,
params=params,
timeout=aiohttp.ClientTimeout(total=30)
) as resp:
return await resp.json()
async def batch_get_details(
self,
product_ids: List[str],
ship_to: str = "US"
) -> List[Dict]:
"""并发获取多个商品详情"""
async with aiohttp.ClientSession() as session:
tasks = [
self.get_product_detail(session, pid, ship_to)
for pid in product_ids
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 过滤异常结果
valid_results = [
r for r in results
if isinstance(r, dict) and r.get("code") == 200
]
return valid_results
# 使用示例
async def main():
api = AsyncAliExpressAPI(api_key="YOUR_API_KEY", max_concurrency=5)
product_ids = ["1005006218491234", "1005006218495678", "1005006218499012"]
details = await api.batch_get_details(product_ids)
print(f"成功获取 {len(details)}/{len(product_ids)} 条商品详情")
asyncio.run(main())5.3 数据缓存策略
import hashlib
import json
from datetime import datetime, timedelta
class APICache:
"""简易内存缓存，减少重复API调用"""
def __init__(self, ttl_seconds=3600):
self.cache = {}
self.ttl = ttl_seconds
def _make_key(self, endpoint: str, params: dict) -> str:
raw = f"{endpoint}:{json.dumps(params, sort_keys=True)}"
return hashlib.md5(raw.encode()).hexdigest()
def get(self, endpoint: str, params: dict):
key = self._make_key(endpoint, params)
if key in self.cache:
data, timestamp = self.cache[key]
if datetime.now() - timestamp < timedelta(seconds=self.ttl):
print(f"[缓存命中] {endpoint}")
return data
return None
def set(self, endpoint: str, params: dict, data: dict):
key = self._make_key(endpoint, params)
self.cache[key] = (data, datetime.now())
# 使用带缓存的API
class CachedAliExpressAPI(AliExpressAPI):
def __init__(self, api_key: str, cache_ttl=3600):
super().__init__(api_key)
self.cache = APICache(cache_ttl)
def get_product_detail(self, product_id: str, **kwargs):
params = {"product_id": product_id, **kwargs}
# 先查缓存
cached = self.cache.get("product/detail", params)
if cached:
return cached
# 缓存未命中，调用API
result = super().get_product_detail(product_id, **kwargs)
if result.get("code") == 200:
self.cache.set("product/detail", params, result)
return result六、技术架构设计：从API到数据流
对于有大规模采集需求的企业用户，推荐以下架构设计：
┌─────────────────────────────────────────────────────────┐
│ 业务应用层 │
│ (选品工具 / 价格监控 / 独立站同步 / 数据分析看板) │
└────────────────────────┬────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────┐
│ API SDK / 客户端 │
│ (Python / Java / Node.js / PHP 多语言SDK) │
└────────────────────────┬────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────┐
│ 速卖通采集API 服务层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 商品搜索 │ │ 商品详情 │ │ 店铺采集 │ │ 物流查询 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 限流管控 │ │ 数据清洗 │ │ 缓存加速 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└────────────────────────┬────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────┐
│ 数据存储层 │
│ (MySQL / MongoDB / Elasticsearch / Redis) │
└──────────────────────────────────────────────────────────┘架构要点：
| 层级 | 核心组件 | 说明 |
|---|---|---|
| 业务应用层 | 选品工具、价格监控、独立站同步 | 直接服务于业务运营 |
| SDK层 | 多语言客户端封装 | 降低接入门槛，统一调用方式 |
| API服务层 | 接口服务 + 限流 + 缓存 | 保证稳定性和性能 |
| 存储层 | 关系库 + 缓存 + 搜索引擎 | 满足不同查询场景 |
七、选型建议：自建爬虫 vs API服务
| 对比维度 | 自建爬虫 | 采集API服务 |
|---|---|---|
| 开发成本 | 高（需维护反爬对抗、解析逻辑） | 低（直接调用接口） |
| 稳定性 | 低（页面改版即失效） | 高（接口稳定，SLA保障） |
| 数据覆盖 | 受反爬限制 | 全品类覆盖 |
| 维护成本 | 持续投入 | 零维护 |
| 合规风险 | 高 | 低（合规数据源） |
| 实时性 | 低 | 高（秒级获取） |
| 扩展性 | 差 | 好（按需调用更多接口） |
建议：对于日均采集量超过 500 条的业务场景，直接使用采集API服务的综合成本远低于自建爬虫。
八、常见问题（FAQ）
Q1：API返回的数据是否实时？
A：商品搜索接口数据延迟通常在 1-5 分钟以内，商品详情接口为实时获取。价格、库存等关键字段建议以详情接口为准。
Q2：API有调用频率限制吗？
A：通常按API Key维度限流，常见限制为每秒 10-50 次请求，每日上限根据套餐不同从 1 万到 100 万次不等。建议在客户端做好限流和缓存。
Q3：可以采集到SKU级别的规格和库存吗？
A：可以。商品详情接口返回完整的 SKU 属性（颜色、尺寸等）及对应的SKU价格、库存数量。
Q4：支持哪些国家/语言？
A：速卖通覆盖 200+ 国家和地区，支持 20+ 语言。通过 ship_to 和 language 参数控制返回的目标国家和语言。
Q5：图片可以下载到本地吗？
A：API返回的图片URL为速卖通CDN地址，可直接下载。建议对采集到的图片做本地化存储，避免依赖第三方CDN。
九、总结
速卖通商品采集API是跨境电商数据基础设施的核心组件。本文从接口能力、代码实现、业务场景、架构设计四个维度进行了系统讲解：
核心接口：商品搜索、详情、店铺批量采集、物流查询
代码实战：Python SDK封装 + 批量采集 + 异步并发 + 数据存储
业务场景：反向海淘选品、竞品监控、市场分析
工程实践：限流重试、异步并发、数据缓存
架构设计：从SDK到存储的四层架构
对于跨境电商从业者和独立站运营者而言，选对数据接口服务商比自建爬虫更高效、更稳定。如果你正在搭建反向海淘系统或需要速卖通数据接口，欢迎联系我们获取API文档和技术支持。
