前言 多数网上教程只演示简单调用速卖通关键词搜索接口，直接拿返回结果做业务输出。但实际对接中经常遇到：页码递增但数据重复、部分类目返回字段缺失、接口限流随机报错、多语言站点参数适配混乱等问题。本文不从基础鉴权讲起，聚焦业务层封装、偏移校验、脏数据过滤，给出可直接用于小规模采集业务的 Python 实现，解决线上容易踩的隐性问题。 前置说明 调用速卖通开放平台搜索接口，需要开发者
AppKey、AppSecret，获取 access_token。接口支持关键词、类目、排序、分页参数。注意官方接口并非 pageNum
简单累加，部分场景会出现数据漂移，单纯循环 page 会出现重复商品或者漏数据。 核心思路：不直接使用接口原始返回，增加一层结果缓存去重、空字段兜底、请求间隔自适应，过滤广告占位商品，输出清洗后的标准化商品列表。 核心代码示例 代码关键点解析 seen_item_ids 集合做本地去重：速卖通搜索接口分页存在数据偏移，翻页会返回重复商品，不能完全依赖接口分页，内存集合做 ID 过滤，保证业务数据唯一性。 强制休眠间隔：官方 QPS 有限制，高频调用会返回权限异常，固定间隔可以降低报错概率。 字段兜底处理，直接 get 获取字段，防止某个商品缺失字段导致程序直接抛出异常中断整个任务。 项目落地常见坑 多语言站点：关键词必须传入对应站点语言，直接传中文搜索英文站点返回结果为空。 广告商品：接口返回会混入推广广告商品，业务场景需要自行根据字段过滤。 分页上限：接口并非可以无限翻页，一般搜索结果有效页数有限，不要设置过大 max_page。 Token 过期：access_token 存在有效期，生产环境要增加 token 自动刷新逻辑。
import time
import requests
class AliExpressSearchClient:
def __init__(self, app_key, access_token):
self.app_key = app_key
self.access_token = access_token
self.api_url = "https://gw.api.aliexpress.com/openapi"
self.seen_item_ids = set() # 本地去重集合
def search_by_keyword(self, keyword, page=1, page_size=20, sort="sale"):
params = {
"app_key": self.app_key,
"timestamp": str(int(time.time()*1000)),
"access_token": self.access_token,
"method": "aliexpress.affiliate.product.query",
"sign_method": "md5",
"keyword": keyword,
"page_no": page,
"page_size": page_size,
"sort": sort
}
resp = requests.get(self.api_url, params=params, timeout=15)
result = resp.json()
return result
def get_clean_goods_list(self, keyword, max_page=3):
clean_list = []
for page in range(1, max_page + 1):
# 自适应休眠，规避平台限流
time.sleep(1.2)
raw_data = self.search_by_keyword(keyword, page=page)
resp_body = raw_data.get("aliexpress_affiliate_product_query_response", {})
product_list = resp_body.get("result", {}).get("products", [])
if not product_list:
break
for item in product_list:
item_id = item.get("product_id")
# 本地去重，解决接口分页漂移重复数据
if not item_id or item_id in self.seen_item_ids:
continue
self.seen_item_ids.add(item_id)
goods = {
"product_id": item_id,
"title": item.get("product_title", ""),
"price": item.get("sale_price"),
"original_price": item.get("original_price"),
"sales": item.get("sales"),
"main_img": item.get("product_main_image_url")
}
clean_list.append(goods)
return clean_list
if __name__ == "__main__":
client = AliExpressSearchClient(app_key="your_appkey", access_token="your_token")
data = client.get_clean_goods_list(keyword="wireless mouse", max_page=2)
print(f"有效商品数量：{len(data)}")
for g in data[:5]:
print(g)
