速卖通(AliExpress)开放平台的item_search接口是根据关键词搜索商品列表的核心工具,适用于跨境电商选品、市场分析、价格监控等场景。本文将全面讲解该接口的对接流程、参数配置、代码实现及最佳实践,帮助开发者系统掌握从入门到精通的全流程。
一、接口基础认知
二、对接前置准备
三、接口调用流程
四、代码实现示例(Python)
import requests
import hashlib
import time
import json
class AliexpressSearchApi:
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.url = "https://gw.api.alibaba.com/openapi/param2/2/portals.open/api.item.search"
self.token_expire_time = 0 # 令牌过期时间(时间戳,秒)
def generate_sign(self, params):
"""生成签名"""
# 1. 按参数名ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接为key=value&key=value格式
sign_str = "&".join([f"{k}={v}" for k, v in sorted_params])
# 3. 追加app_secret
sign_str += f"&app_secret={self.app_secret}"
# 4. MD5加密(32位大写)
sign = hashlib.md5(sign_str.encode()).hexdigest().upper()
return sign
def refresh_access_token(self):
"""刷新access_token"""
token_url = "https://gw.api.alibaba.com/openapi/http/1/system.oauth2/getToken"
params = {
"grant_type": "refresh_token",
"client_id": self.app_key,
"client_secret": self.app_secret,
"refresh_token": "your_refresh_token" # 替换为实际的refresh_token
}
try:
response = requests.post(token_url, data=params, timeout=10)
result = response.json()
if "error" in result:
raise Exception(f"令牌刷新失败: {result['error_description']}")
self.access_token = result["access_token"]
self.token_expire_time = time.time() + int(result["expires_in"])
return self.access_token
except Exception as e:
raise Exception(f"刷新token异常: {str(e)}")
def item_search(self, keyword, page=1, page_size=20, **kwargs):
"""
搜索商品列表
:param keyword: 搜索关键词(必填)
:param page: 页码,默认1
:param page_size: 每页条数,默认20,最大50
:param kwargs: 可选参数(priceStart, priceEnd, sort等)
:return: 搜索结果
"""
# 检查并刷新令牌
if not self.access_token or time.time() > self.token_expire_time - 60:
self.refresh_access_token()
# 1. 组装基础参数
params = {
"app_key": self.app_key,
"access_token": self.access_token,
"timestamp": int(time.time() * 1000), # 毫秒级时间戳
"format": "json",
"v": "1.0",
"keyword": keyword,
"page": page,
"pageSize": page_size
}
# 2. 合并可选参数
params.update(kwargs)
# 3. 生成签名
params["sign"] = self.generate_sign(params)
try:
# 4. 发送POST请求
response = requests.post(
url=self.url,
json=params,
headers={"Content-Type": "application/json"},
timeout=15
)
response.raise_for_status()
result = response.json()
# 5. 处理响应
if "error_response" in result:
error = result["error_response"]
return {
"success": False,
"error_code": error.get("code"),
"error_msg": error.get("msg")
}
search_data = result.get("item_search_response", {})
return {
"success": True,
"total_count": search_data.get("totalCount", 0),
"page": page,
"page_size": page_size,
"total_pages": search_data.get("totalPages", 0),
"items": search_data.get("items", {}).get("item", [])
}
except Exception as e:
return {
"success": False,
"error_msg": f"请求异常: {str(e)}"
}
# 使用示例
if __name__ == "__main__":
# 替换为实际参数
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
ACCESS_TOKEN = "your_access_token"
# 初始化API客户端
api = AliexpressSearchApi(APP_KEY, APP_SECRET, ACCESS_TOKEN)
# 搜索"wireless earbuds",第1页,20条/页,价格10-50美元,按销量降序
result = api.item_search(
keyword="wireless earbuds",
page=1,
page_size=20,
priceStart=10,
priceEnd=50,
sort="SALE_AMOUNT_DESC" # 排序方式:SALE_AMOUNT_DESC-销量降序,PRICE_ASC-价格升序
)
if result["success"]:
print(f"共搜索到 {result['total_count']} 个商品")
print(f"第 {result['page']} 页,共 {result['total_pages']} 页")
# 打印前5个商品信息
for i, item in enumerate(result["items"][:5]):
print(f"\n商品 {i+1}:")
print(f"ID: {item.get('productId')}")
print(f"标题: {item.get('title')}")
print(f"售价: {item.get('price')} {item.get('currency')}")
print(f"30天销量: {item.get('saleCount')}")
print(f"店铺: {item.get('storeName')}")
print(f"主图: {item.get('imageUrl')}")
else:
print(f"搜索失败: {result['error_msg']} (错误码: {result.get('error_code')})")