易贝（eBay）开放平台的item_search接口是通过关键词、分类等条件搜索商品列表的核心工具，适用于全球跨境电商选品、市场分析、价格监控等场景。作为全球第二大电商平台，其覆盖的商品品类和市场范围极具商业价值。本文将全面讲解该接口的对接流程、参数配置、代码实现及最佳实践，帮助开发者系统掌握从入门到精通的全流程。
一、接口基础认知
二、对接前置准备
三、接口调用流程
四、代码实现示例（Python）
import requests
import base64
import time
class EbaySearchApi:
def __init__(self, client_id, client_secret, marketplace_id="EBAY-US"):
self.client_id = client_id
self.client_secret = client_secret
self.marketplace_id = marketplace_id # 站点标识
self.base_url = "https://api.ebay.com/buy/browse/v1/item_summary/search"
self.token_url = "https://api.ebay.com/identity/v1/oauth2/token"
self.access_token = None
self.token_expire_time = 0 # 令牌过期时间（秒级时间戳）
def get_access_token(self):
"""获取Access Token（自动处理过期）"""
if self.access_token and time.time() < self.token_expire_time - 60:
return self.access_token
# 构建Basic认证
auth_str = f"{self.client_id}:{self.client_secret}"
auth_base64 = base64.b64encode(auth_str.encode()).decode()
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Authorization": f"Basic {auth_base64}"
}
data = {
"grant_type": "client_credentials",
"scope": "https://api.ebay.com/oauth/api_scope/buy.browse"
}
try:
response = requests.post(
self.token_url,
headers=headers,
data=data,
timeout=10
)
response.raise_for_status()
result = response.json()
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, limit=20, offset=0, filters=None, sort=None):
"""
搜索商品列表
:param keyword: 搜索关键词（必填）
:param limit: 每页条数（1-100）
:param offset: 偏移量（分页起始位置）
:param filters: 筛选条件字典（如{"price": "[10..100]", "category_id": "2035"}）
:param sort: 排序方式（如"price_asc"）
:return: 搜索结果
"""
# 获取令牌
self.get_access_token()
# 构建请求头
headers = {
"Authorization": f"Bearer {self.access_token}",
"X-EBAY-C-MARKETPLACE-ID": self.marketplace_id,
"Content-Type": "application/json"
}
# 构建查询参数
params = {
"q": keyword,
"limit": min(limit, 100), # 限制最大条数
"offset": offset
}
# 添加筛选条件
if filters:
filter_str = ",".join([f"{k}:{v}" for k, v in filters.items()])
params["filter"] = filter_str
# 添加排序条件
if sort:
params["sort"] = sort
try:
# 发送请求
response = requests.get(
self.base_url,
headers=headers,
params=params,
timeout=15
)
response.raise_for_status()
result = response.json()
# 处理响应
if "errors" in result:
return {
"success": False,
"errors": result["errors"]
}
# 提取分页信息
pagination = result.get("pagination", {})
return {
"success": True,
"total_items": pagination.get("totalItems", 0),
"limit": pagination.get("limit", limit),
"offset": pagination.get("offset", offset),
"items": result.get("itemSummaries", [])
}
except Exception as e:
return {
"success": False,
"error_msg": f"请求异常: {str(e)}"
}
# 使用示例
if __name__ == "__main__":
# 替换为实际参数
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
MARKETPLACE_ID = "EBAY-US" # 美国站点
# 初始化API客户端
api = EbaySearchApi(CLIENT_ID, CLIENT_SECRET, MARKETPLACE_ID)
# 搜索"wireless headphones"，价格10-100美元，电子分类，按价格升序
result = api.item_search(
keyword="wireless headphones",
limit=20,
offset=0,
filters={
"price": "[10..100]", # 价格区间
"category_id": "2035", # 电子分类ID
"item_condition": "NEW" # 全新商品
},
sort="price_asc" # 价格升序
)
if result["success"]:
print(f"搜索结果: 共 {result['total_items']} 个商品")
print(f"当前显示: 第 {result['offset']+1}-{result['offset']+len(result['items'])} 个\n")
# 打印前5个商品信息
for i, item in enumerate(result["items"][:5]):
print(f"商品 {i+1}:")
print(f"ID: {item.get('itemId')}")
print(f"标题: {item.get('title')[:50]}...") # 截断长标题
print(f"售价: {item.get('price', {}).get('value')} {item.get('price', {}).get('currency')}")
print(f"卖家好评率: {item.get('seller', {}).get('feedbackPercentage')}%")
print(f"主图: {item.get('image', {}).get('imageUrl')}")
print(f"商品URL: {item.get('itemWebUrl')}\n")
else:
print(f"搜索失败: {result.get('error_msg') or result.get('errors')}")