在跨境电商开发领域，eBay 作为全球最大的在线交易平台之一，其开放 API 为开发者提供了丰富的商品数据获取能力。本文将聚焦 eBay 关键字搜索商品列表接口的实现，涵盖 OAuth2.0 认证、高级搜索参数配置、分页策略及完整代码实现，帮助开发者快速构建稳定的 eBay 商品检索功能。
一、eBay 搜索 API 基础信息
eBay 提供的 Finding API 是获取商品列表的核心接口，支持通过关键字、分类、价格区间等多维度筛选商品。
核心特点：
基于 RESTful 架构设计，支持 JSON/XML 响应格式
采用 OAuth2.0 认证机制，安全性更高
提供丰富的筛选参数，支持精确搜索
单页最大返回 100 条数据，支持分页查询
接口端点：https://api.ebay.com/services/search/FindingService/v1
二、认证机制详解
eBay Finding API 使用 OAuth2.0 进行身份验证，获取访问令牌的步骤如下：
在 eBay 开发者平台创建应用，获取 Client ID 和 Client Secret
通过客户端凭证流程 (Client Credentials Flow) 获取访问令牌
令牌有效期为 7200 秒 (2 小时)，过期前需重新获取
三、核心搜索参数说明
基础参数：
keywords：搜索关键字（必填）
categoryId：商品分类 ID（可选）
itemFilter：过滤条件（价格区间、卖家类型等）
sortOrder：排序方式（BestMatch、PricePlusShippingLowest 等）
分页参数：
paginationInput.pageNumber：页码
paginationInput.entriesPerPage：每页条数 (1-100)
输出控制：
outputSelector：指定返回字段，减少数据传输量
四、完整代码实现
下面是使用 Python 实现的 eBay 关键字搜索商品列表功能，包含令牌管理和搜索逻辑：
eBay关键字搜索商品列表接口实现
import requests
import time
import json
from typing import Dict, List, Optional
class EbaySearchAPI:
def __init__(self, client_id: str, client_secret: str, site_id: str = '0'):
"""
初始化eBay搜索API客户端
:param client_id: 应用的Client ID
:param client_secret: 应用的Client Secret
:param site_id: 站点ID，0表示美国站
"""
self.client_id = client_id
self.client_secret = client_secret
self.site_id = site_id
self.base_url = "Service Unavailable"
self.oauth_url = "https://api.ebay.com/identity/v1/oauth2/token"
self.access_token = None
self.token_expiry = 0 # 令牌过期时间戳
def _get_access_token(self) -> Optional[str]:
"""获取或刷新访问令牌"""
# 检查令牌是否有效
if self.access_token and time.time() < self.token_expiry:
return self.access_token
# 准备请求参数
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Authorization": f"Basic {self._encode_credentials()}"
}
data = {
"grant_type": "client_credentials",
"scope": "https://api.ebay.com/oauth/api_scope"
}
try:
response = requests.post(
self.oauth_url,
headers=headers,
data=data,
timeout=10
)
response.raise_for_status()
token_data = response.json()
# 保存令牌及过期时间
self.access_token = token_data["access_token"]
self.token_expiry = time.time() + token_data["expires_in"] - 60 # 提前60秒刷新
return self.access_token
except requests.exceptions.RequestException as e:
print(f"获取令牌失败: {str(e)}")
return None
def _encode_credentials(self) -> str:
"""编码客户端凭证"""
import base64
credentials = f"{self.client_id}:{self.client_secret}".encode("utf-8")
return base64.b64encode(credentials).decode("utf-8")
def search_products(self,
keywords: str,
page: int = 1,
per_page: int = 20,
min_price: Optional[float] = None,
max_price: Optional[float] = None,
sort_order: str = "BestMatch") -> Dict:
"""
搜索eBay商品
:param keywords: 搜索关键字
:param page: 页码
:param per_page: 每页条数(1-100)
:param min_price: 最低价格
:param max_price: 最高价格
:param sort_order: 排序方式
:return: 搜索结果
"""
# 获取访问令牌
token = self._get_access_token()
if not token:
return {"error": "无法获取访问令牌"}
# 准备请求参数
params = {
"OPERATION-NAME": "findItemsAdvanced",
"SERVICE-VERSION": "1.13.0",
"SECURITY-APPNAME": self.client_id,
"GLOBAL-ID": f"EBAY-{self.site_id}",
"keywords": keywords,
"paginationInput.pageNumber": page,
"paginationInput.entriesPerPage": per_page,
"sortOrder": sort_order,
"response-data-format": "JSON"
}
# 添加价格过滤条件
filter_index = 0
if min_price is not None:
params[f"itemFilter({filter_index}).name"] = "MinPrice"
params[f"itemFilter({filter_index}).value"] = min_price
params[f"itemFilter({filter_index}).paramName"] = "Currency"
params[f"itemFilter({filter_index}).paramValue"] = "USD"
filter_index += 1
if max_price is not None:
params[f"itemFilter({filter_index}).name"] = "MaxPrice"
params[f"itemFilter({filter_index}).value"] = max_price
params[f"itemFilter({filter_index}).paramName"] = "Currency"
params[f"itemFilter({filter_index}).paramValue"] = "USD"
filter_index += 1
# 设置请求头
headers = {
"Authorization": f"Bearer {token}",
"X-EBAY-SOA-REQUEST-DATA-FORMAT": "JSON"
}
try:
response = requests.get(
self.base_url,
params=params,
headers=headers,
timeout=15
)
response.raise_for_status()
return self._parse_response(response.json())
except requests.exceptions.RequestException as e:
print(f"搜索请求失败: {str(e)}")
return {"error": str(e)}
def _parse_response(self, response: Dict) -> Dict:
"""解析API响应，提取有用信息"""
result = {
"total_items": 0,
"page": 0,
"per_page": 0,
"items": []
}
try:
search_result = response["findItemsAdvancedResponse"][0]
# 提取分页信息
pagination = search_result["paginationOutput"][0]
result["total_items"] = int(pagination["totalEntries"][0])
result["page"] = int(pagination["pageNumber"][0])
result["per_page"] = int(pagination["entriesPerPage"][0])
# 提取商品信息
if "searchResult" in search_result and len(search_result["searchResult"][0]["item"]) > 0:
for item in search_result["searchResult"][0]["item"]:
result["items"].append({
"item_id": item["itemId"][0],
"title": item["title"][0],
"price": {
"value": float(item["sellingStatus"][0]["currentPrice"][0]["__value__"]),
"currency": item["sellingStatus"][0]["currentPrice"][0]["@currencyId"]
},
"url": item["viewItemURL"][0],
"location": item.get("location", [""])[0],
"shipping_cost": float(item["shippingInfo"][0]["shippingServiceCost"][0]["__value__"]),
"is_top_rated": "topRatedListing" in item and item["topRatedListing"][0].lower() == "true"
})
except (KeyError, IndexError, ValueError) as e:
print(f"解析响应失败: {str(e)}")
result["error"] = f"解析响应失败: {str(e)}"
return result
# 使用示例
if __name__ == "__main__":
# 替换为你的eBay应用凭证
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
# 初始化API客户端（美国站）
ebay_api = EbaySearchAPI(CLIENT_ID, CLIENT_SECRET, site_id='0')
# 搜索商品
search_result = ebay_api.search_products(
keywords="wireless headphones",
page=1,
per_page=20,
min_price=20.0,
max_price=100.0,
sort_order="PricePlusShippingLowest"
)
# 处理搜索结果
if "error" not in search_result:
print(f"找到 {search_result['total_items']} 个商品")
print(f"当前第 {search_result['page']} 页，共 {search_result['per_page']} 条/页\n")
for idx, item in enumerate(search_result["items"], 1):
print(f"{idx}. {item['title']}")
print(f" 价格: {item['price']['currency']} {item['price']['value']}")
print(f" 运费: {item['price']['currency']} {item['shipping_cost']}")
print(f" 链接: {item['url']}\n")
else:
print(f"搜索失败: {search_result['error']}")
五、代码核心功能解析
令牌管理机制：
自动处理令牌的获取与刷新，无需手动干预
提前 60 秒刷新令牌，避免请求时令牌过期
使用 Base64 编码客户端凭证，符合 OAuth2.0 规范
搜索参数处理：
支持多维度筛选，包括价格区间、排序方式等
灵活处理可选参数，仅在提供时添加到请求中
严格遵循 eBay API 的参数命名规范
响应解析优化：
将原始 API 响应转换为更友好的字典格式
提取核心商品信息，去除冗余数据
包含错误处理，提高代码健壮性
多站点支持：
通过 site_id 参数支持不同国家 / 地区的 eBay 站点
默认为美国站 (0)，可根据需求切换为其他站点
六、实战注意事项
API 调用限制：
Finding API 有调用频率限制，默认每日 10,000 次
建议实现请求间隔控制，避免触发限流
站点选择：
不同站点的商品和价格存在差异
完整站点 ID 列表可参考 eBay 开发者文档
错误处理：
常见错误包括令牌过期、参数错误、频率超限
实际开发中应根据错误代码实现针对性处理
数据缓存：
对热门搜索词结果进行缓存，减少 API 调用
缓存时间建议设置为 15-30 分钟，保证数据新鲜度
七、功能扩展建议
实现批量搜索功能，支持多关键字同时查询
添加商品图片 URL 提取，丰富展示内容
集成价格趋势分析，提供历史价格数据
实现搜索结果的本地存储，支持离线查看
通过本文介绍的方法，开发者可以快速集成 eBay 的商品搜索功能，为跨境电商应用提供稳定可靠的数据源。在实际开发中，需遵守 eBay 开发者协议，合理使用 API 获取的数据。
