引言：为什么需要掌握eBay API？
在当今的电商生态系统中，eBay作为全球最大的在线交易平台之一，每天产生数以亿计的商品数据、交易记录和用户行为信息。对于开发者、数据分析师、电商运营者而言，掌握eBay API调用技能已成为获取这些宝贵数据的关键途径。
通过eBay API，你可以实现：
自动化商品管理
：批量上传、修改、下架商品
实时监控竞争对手
：获取竞品价格、库存变化
数据分析与洞察
：分析市场趋势、用户偏好
订单处理自动化
：同步订单状态、自动发货通知
个性化推荐系统
：基于用户行为构建推荐算法
然而，eBay API的调用过程并非一帆风顺。许多开发者在初次接触时会遇到认证失败、参数错误、数据解析异常等难题。本文将从零开始，系统讲解eBay API的调用流程、技巧以及常见问题的解决方案。
一、eBay API基础概念
1.1 eBay API的类型
eBay提供了多种API接口，主要分为以下几类：
1. RESTful API（推荐）
Trading API
：核心交易功能，商品管理、订单处理
Browse API
：浏览商品、搜索商品
Order API
：订单管理
Fulfillment API
：物流与履约
Analytics API
：数据分析
2. SOAP API（传统）
适用于遗留系统，但eBay已逐渐停止维护
3. Data API
提供数据查询功能，如获取销售数据、市场趋势
1.2 认证机制
eBay API使用OAuth 2.0认证流程，需要以下凭证：
Client ID
：应用标识符
Client Secret
：应用密钥
Redirect URI
：回调地址
User Access Token
：用户访问令牌（用于代表用户操作）
App Access Token
：应用访问令牌（用于应用级操作）
二、准备工作：注册与配置
2.1 注册eBay开发者账号
访问
eBay Developers Program
点击”Join the Developers Program”
选择账户类型（推荐”Production”环境）
填写个人信息并同意协议
支付会员费（基础会员免费，高级功能需要付费）
2.2 创建应用
登录开发者控制台
点击”Create a Keyset”
填写应用名称和描述
选择API权限（根据需求选择，如Browse API、Trading API）
设置回调地址（Redirect URI），例如：
https://your-app.com/callback
2.3 获取凭证
创建成功后，你将获得：
Client ID
（App ID）
Client Secret
（Cert ID）
Dev ID
（开发者ID）
重要提示
：请妥善保管这些凭证，切勿泄露到公共代码库。
三、认证流程详解
3.1 获取User Access Token（代表用户操作）
这是最复杂的一步，需要用户授权。流程如下：
import requests
import base64
import json
# 你的凭证
CLIENT_ID = “你的ClientID”
CLIENT_SECRET = “你的ClientSecret”
REDIRECT_URI = “你的RedirectURI”
# 1. 生成授权URL（让用户访问）
def generate_auth_url():
auth_url = “https://auth.ebay.com/oauth2/authorize”
params = {
“client_id”: CLIENT_ID,
“response_type”: “code”,
“redirect_uri”: REDIRECT_URI,
“scope”: “https://api.ebay.com/oauth/api_scope”
}
return f”{auth_url}?{ ‘&’.join([f'{k}={v}’ for k,v in params.items()]) }”
# 2. 交换Token（用户授权后，用code换取token）
def get_access_token(auth_code):
token_url = “https://api.ebay.com/identity/v1/oauth2/token”
# 生成Basic Auth Header
credentials = f”{CLIENT_ID}:{CLIENT_SECRET}”
encoded_credentials = base64.b64encode(credentials.encode()).decode()
headers = {
“Authorization”: f”Basic {encoded_credentials}”,
“Content-Type”: “application/x-www-form-urlencoded”
}
data = {
“grant_type”: “authorization_code”,
“code”: auth_code,
“redirect_uri”: REDIRECT_URI
}
response = requests.post(token_url, headers=headers, data=data)
return response.json()
# 示例使用
if __name__ == “__main__”:
# 第一步：打印授权URL，让用户访问并获取code
print(“请访问此URL进行授权：”)
print(generate_auth_url())
# 第二步：用户授权后，会跳转到你的redirect_uri并携带code
# 例如：https://your-app.com/callback?code=abcdef123456
auth_code = “用户返回的code”
# 第三步：获取token
token_response = get_access_token(auth_code)
print(“Token响应：”, json.dumps(token_response, indent=2))
# 保存token
access_token = token_response[“access_token”]
refresh_token = token_response[“refresh_token”]
3.2 获取App Access Token（应用级操作）
对于不需要用户授权的操作（如搜索商品），可以使用App Access Token：
def get_app_access_token():
token_url = “https://api.ebay.com/identity/v1/oauth2/token”
credentials = f”{CLIENT_ID}:{CLIENT_SECRET}”
encoded_credentials = base64.b64encode(credentials.encode()).decode()
headers = {
“Authorization”: f”Basic {encoded_credentials}”,
“Content-Type”: “application/x-www-form-urlencoded”
}
data = {
“grant_type”: “client_credentials”,
“scope”: “https://api.ebay.com/oauth/api_scope”
}
response = requests.post(token_url, headers=headers, data=data)
return response.json()
# 使用示例
token_data = get_app_access_token()
app_access_token = token_data[“access_token”]
3.3 Token刷新机制
User Access Token有效期为2小时，Refresh Token有效期为2天。你需要实现自动刷新：
def refresh_user_token(refresh_token):
token_url = “https://api.ebay.com/identity/v1/oauth2/token”
credentials = f”{CLIENT_ID}:{CLIENT_SECRET}”
encoded_credentials = base64.b64encode(credentials.encode()).decode()
headers = {
“Authorization”: f”Basic {encoded_credentials}”,
“Content-Type”: “application/x-www-form-urlencoded”
}
data = {
“grant_type”: “refresh_token”,
“refresh_token”: refresh_token
}
response = requests.post(token_url, headers=headers, data=data)
return response.json()
四、核心API调用实战
4.1 Browse API：搜索商品
这是最常用的API之一，用于搜索eBay平台上的商品。
import requests
import json
class EbayAPI:
def __init__(self, access_token):
self.access_token = access_token
self.base_url = “https://api.ebay.com/buy/browse/v1”
self.headers = {
“Authorization”: f”Bearer {access_token}”,
“Content-Type”: “application/json”,
“X-EBAY-C-MARKETPLACE-ID”: “EBAY_US” # 美国站
}
def search_items(self, keyword, limit=10, filters=None):
“””
搜索商品
:param keyword: 搜索关键词
:param limit: 返回数量限制
:param filters: 过滤条件，如”price:[10..100]”
“””
endpoint = f”{self.base_url}/item_summary/search”
params = {
“q”: keyword,
“limit”: limit
}
if filters:
params[“filter”] = filters
response = requests.get(endpoint, headers=self.headers, params=params)
if response.status_code == 200:
return response.json()
else:
raise Exception(f”API调用失败: {response.status_code} – {response.text}”)
def get_item_details(self, item_id):
“””获取商品详情”””
endpoint = f”{self.base_url}/item/{item_id}”
response = requests.get(endpoint, headers=self.headers)
if response.status_code == 200:
return response.json()
else:
raise Exception(f”API调用失败: {response.status_code} – {response.text}”)
# 使用示例
if __name__ == “__main__”:
# 假设你已经有access_token
api = EbayAPI(“你的access_token”)
# 搜索iPhone，价格在100-500美元之间
try:
results = api.search_items(
keyword=”iPhone”,
limit=5,
filters=”price:[100..500],condition:{New}”
)
print(“搜索结果：”)
for item in results.get(“itemSummaries”, []):
print(f”商品ID: {item[‘itemId’]}”)
print(f”标题: {item[‘title’]}”)
print(f”价格: {item[‘price’][‘value’]} {item[‘price’][‘currency’]}”)
print(f”卖家: {item[‘seller’][‘username’]}”)
print(“-” * 50)
except Exception as e:
print(f”错误: {e}”)
4.2 Trading API：商品管理
Trading API功能更强大，但需要User Access Token。
import requests
import json
from xml.etree import ElementTree as ET
class EbayTradingAPI:
def __init__(self, access_token):
self.access_token = access_token
self.endpoint = “https://api.ebay.com/ws/api.dll”
self.headers = {
“X-EBAY-API-CALL-NAME”: “GetMyeBaySelling”,
“X-EBAY-API-APP-ID”: “你的ClientID”,
“X-EBAY-API-VERSION”: “1131”,
“X-EBAY-API-SITE-ID”: “0”, # 美国站
“Content-Type”: “text/xml”
}
def get_selling_items(self):
“””获取在售商品”””
xml_request = f”””
“””
response = requests.post(self.endpoint, headers=self.headers, data=xml_request)
if response.status_code == 200:
# 解析XML响应
root = ET.fromstring(response.content)
# 处理响应数据…
return root
else:
raise Exception(f”API调用失败: {response.status_code}”)
# 使用示例
trading_api = EbayTradingAPI(“你的UserAccessToken”)
selling_items = trading_api.get_selling_items()
4.3 Order API：订单管理
def get_orders(api, limit=50):
“””获取订单列表”””
endpoint = “https://api.ebay.com/sell/order/v2/order”
headers = {
“Authorization”: f”Bearer {api.access_token}”,
“Content-Type”: “application/json”,
“X-EBAY-C-MARKETPLACE-ID”: “EBAY_US”
}
params = {
“limit”: limit,
“filter”: “orderStatus:{PAID,SHIPPED}”
}
response = requests.get(endpoint, headers=headers, params=params)
if response.status_code == 200:
return response.json()
else:
raise Exception(f”API调用失败: {response.status_code} – {response.text}”)
五、常见报错与解决方案
5.1 认证相关错误
错误1：Invalid Scope
error: invalid_scope
原因
：请求的scope不在应用权限范围内。
解决方案
：
检查开发者控制台中申请的API权限
确保授权URL中的scope参数正确
重新创建Keyset并申请所需权限
错误2：Invalid Token
error: invalid_token
原因
：Token过期或被撤销。
解决方案
：
实现Token自动刷新机制
检查Token有效期（User Token 2小时）
重新获取Token
错误3：Redirect URI mismatch
error: redirect_uri_mismatch
原因
：授权URL中的redirect_uri与注册时不一致。
解决方案
：
确保授权URL和Token交换时使用相同的redirect_uri
检查开发者控制台中的redirect_uri配置
5.2 API调用错误
错误4：Rate Limit Exceeded
error: 429 Too Many Requests
原因
：请求频率超过限制。
解决方案
：
实现请求队列和延迟机制
了解API限流规则：
Browse API：每秒5次请求
Trading API：每小时5000次调用
使用指数退避策略重试
import time
import random
def call_api_with_retry(api_func, max_retries=3):
“””带重试机制的API调用”””
for attempt in range(max_retries):
try:
return api_func()
except Exception as e:
if “429” in str(e) or “rate limit” in str(e).lower():
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f”触发限流，等待 {wait_time:.2f} 秒…”)
time.sleep(wait_time)
else:
raise e
raise Exception(“达到最大重试次数”)
错误5：Item not found
error: 404 Item not found
原因
：商品ID错误或商品已下架。
解决方案
：
验证商品ID格式（通常是12位数字）
检查商品状态
使用Browse API的item_summary接口验证商品是否存在
错误6：Invalid parameter
error: 400 Bad Request – Invalid filter parameter
原因
：过滤条件格式错误。
解决方案
：
查阅官方文档的参数格式
使用URL编码特殊字符
示例正确格式：
price:[10..100]
（价格范围）
condition:{New,Used}
（多个条件）
sellingStatus:{EndedNotSold}
（销售状态）
5.3 数据解析错误
错误7：JSON解析失败
# 错误示例
try:
data = response.json()
except json.JSONDecodeError:
print(“JSON解析失败”)
原因
：
响应不是有效的JSON
响应被截断
服务器返回HTML错误页面
解决方案
：
def safe_parse_json(response):
“””安全解析JSON响应”””
try:
return response.json()
except json.JSONDecodeError:
# 记录原始响应用于调试
print(f”响应状态码: {response.status_code}”)
print(f”响应头: {response.headers}”)
print(f”响应内容: {response.text[:500]}”) # 只打印前500字符
raise
错误8：字段缺失
# 错误示例
price = item[‘price’][‘value’] # 可能KeyError
原因
：API响应结构变化或字段不存在。
解决方案
：
# 使用get方法安全访问
price = item.get(‘price’, {}).get(‘value’, 0)
# 或使用try-except
try:
price = item[‘price’][‘value’]
except KeyError:
price = 0
六、数据解析与处理技巧
6.1 解析商品搜索结果
def parse_item_summary(item):
“””解析商品摘要信息”””
return {
“item_id”: item.get(“itemId”),
“title”: item.get(“title”),
“price”: item.get(“price”, {}).get(“value”),
“currency”: item.get(“price”, {}).get(“currency”),
“condition”: item.get(“condition”),
“seller”: item.get(“seller”, {}).get(“username”),
“feedback_score”: item.get(“seller”, {}).get(“feedbackScore”),
“listing_marketplace_id”: item.get(“listingMarketplaceId”),
“image_url”: item.get(“image”, {}).get(“imageUrl”),
“item_url”: item.get(“itemWebUrl”),
“availability”: item.get(“availability”, {}).get(“status”),
“seller_location”: item.get(“seller”, {}).get(“sellerAccountType”)
}
# 批量处理搜索结果
def process_search_results(api, keyword, max_items=100):
“””批量搜索并解析商品”””
all_items = []
offset = 0
limit = 200 # API最大限制
while len(all_items) < max_items:
try:
results = api.search_items(
keyword=keyword,
limit=min(limit, max_items - len(all_items)),
filters="price:[1..1000]"
)
items = results.get("itemSummaries", [])
if not items:
break
for item in items:
parsed = parse_item_summary(item)
all_items.append(parsed)
# 检查是否有更多结果
if not results.get("next"):
break
offset += limit
except Exception as e:
print(f"处理批次失败: {e}")
break
return all_items[:max_items]
6.2 解析订单数据
def parse_order(order):
"""解析订单信息"""
return {
"order_id": order.get("orderId"),
"order_status": order.get("orderStatus"),
"creation_date": order.get("creationDate"),
"last_modified_date": order.get("lastModifiedDate"),
"total": order.get("amount", {}).get("value"),
"currency": order.get("amount", {}).get("currency"),
"buyer": order.get("buyer", {}).get("username"),
"payment_status": order.get("paymentStatus"),
"fulfillment_status": order.get("fulfillmentStatus"),
"line_items": [
{
"item_id": line_item.get("legacyItemId"),
"title": line_item.get("title"),
"quantity": line_item.get("quantity"),
"unit_price": line_item.get("unitPrice", {}).get("value")
}
for line_item in order.get("lineItems", [])
]
}
def get_all_orders(api, days=30):
"""获取指定天数内的所有订单"""
from datetime import datetime, timedelta
start_date = (datetime.now() - timedelta(days=days)).isoformat()
endpoint = "https://api.ebay.com/sell/order/v2/order"
headers = {
"Authorization": f"Bearer {api.access_token}",
"Content-Type": "1.1",
"X-EBAY-C-MARKETPLACE-ID": "EBAY_US"
}
params = {
"limit": 200,
"filter": f"creationDate:[{start_date}..]"
}
all_orders = []
while True:
response = requests.get(endpoint, headers=headers, params=params)
if response.status_code != 200:
raise Exception(f"API调用失败: {response.status_code}")
data = response.json()
orders = data.get("orders", [])
for order in orders:
all_orders.append(parse_order(order))
# 检查是否有下一页
if "next" not in data.get("links", {}):
break
# 更新offset参数（假设API支持）
params["offset"] = data.get("next")
return all_orders
6.3 数据清洗与验证
import re
def clean_text(text):
"""清洗文本数据"""
if not text:
return ""
# 移除HTML标签
text = re.sub(r'<[^>]+>’, ”, text)
# 移除特殊字符
text = re.sub(r'[^\w\s\-]’, ”, text)
# 标准化空格
text = ” “.join(text.split())
return text
def validate_item_data(item):
“””验证商品数据完整性”””
required_fields = [‘itemId’, ‘title’, ‘price’]
missing_fields = [f for f in required_fields if f not in item]
if missing_fields:
raise ValueError(f”缺失必填字段: {missing_fields}”)
# 验证价格格式
try:
price = float(item[‘price’][‘value’])
if price < 0:
raise ValueError("价格不能为负数")
except (ValueError, KeyError):
raise ValueError("价格格式错误")
# 验证商品ID格式
item_id = str(item['itemId'])
if not re.match(r'^\d{12}$', item_id):
raise ValueError(f"商品ID格式错误: {item_id}")
return True
6.4 数据存储
import sqlite3
import pandas as pd
def save_to_sqlite(data, db_path="ebay_data.db"):
"""保存数据到SQLite"""
conn = sqlite3.connect(db_path)
# 转换为DataFrame
df = pd.DataFrame(data)
# 保存
df.to_sql('ebay_items', conn, if_exists='append', index=False)
conn.close()
print(f"保存 {len(data)} 条记录到 {db_path}")
def save_to_csv(data, filename="ebay_data.csv"):
"""保存数据到CSV"""
df = pd.DataFrame(data)
df.to_csv(filename, index=False, encoding='utf-8')
print(f"保存 {len(data)} 条记录到 {filename}")
七、高级技巧与最佳实践
7.1 实现请求队列与限流
import queue
import threading
import time
class EbayAPIQueue:
def __init__(self, access_token, rate_limit=5):
self.access_token = access_token
self.rate_limit = rate_limit # 每秒请求数
self.request_queue = queue.Queue()
self.worker_thread = None
self.running = False
def start(self):
"""启动工作线程"""
self.running = True
self.worker_thread = threading.Thread(target=self._process_queue)
self.worker_thread.daemon = True
self.worker_thread.start()
def stop(self):
"""停止工作线程"""
self.running = False
if self.worker_thread:
self.worker_thread.join()
def add_request(self, method, url, params=None, callback=None):
"""添加请求到队列"""
self.request_queue.put({
"method": method,
"url": url,
"params": params,
"callback": callback
})
def _process_queue(self):
"""处理请求队列"""
last_request_time = 0
while self.running:
try:
# 检查限流
current_time = time.time()
time_since_last = current_time - last_request_time
if time_since_last < (1.0 / self.rate_limit):
sleep_time = (1.0 / self.rate_limit) - time_since_last
time.sleep(sleep_time)
# 获取请求
request = self.request_queue.get(timeout=1)
# 执行请求
headers = {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json"
}
if request["method"] == "GET":
response = requests.get(request["url"], headers=headers, params=request["params"])
elif request["method"] == "POST":
response = requests.post(request["url"], headers=headers, json=request["params"])
# 回调处理
if request["callback"]:
request["callback"](response)
last_request_time = time.time()
self.request_queue.task_done()
except queue.Empty:
continue
except Exception as e:
print(f"请求处理错误: {e}")
# 使用示例
api_queue = EbayAPIQueue("你的access_token", rate_limit=5)
api_queue.start()
# 添加多个请求
for i in range(10):
api_queue.add_request(
method="GET",
url="https://api.ebay.com/buy/browse/v1/item_summary/search",
params={"q": "iPhone", "limit": 1},
callback=lambda r: print(f"请求完成: {r.status_code}")
)
# 等待所有请求完成
api_queue.request_queue.join()
api_queue.stop()
7.2 批量操作优化
def batch_get_item_details(api, item_ids, batch_size=10):
"""批量获取商品详情"""
all_details = []
for i in range(0, len(item_ids), batch_size):
batch = item_ids[i:i+batch_size]
# 并行请求（注意：eBay API不支持真正的批量请求，这里模拟）
for item_id in batch:
try:
details = api.get_item_details(item_id)
all_details.append(details)
time.sleep(0.2) # 避免过快
except Exception as e:
print(f"获取商品 {item_id} 失败: {e}")
continue
return all_details
7.3 错误监控与日志
import logging
from datetime import datetime
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(f"ebay_api_{datetime.now().strftime('%Y%m%d')}.log"),
logging.StreamHandler()
]
)
def log_api_call(endpoint, status_code, response_time, error=None):
"""记录API调用日志"""
logger = logging.getLogger(__name__)
log_data = {
"timestamp": datetime.now().isoformat(),
"endpoint": endpoint,
"status_code": status_code,
"response_time": response_time,
"error": str(error) if error else None
}
if error:
logger.error(f"API调用失败: {json.dumps(log_data)}")
else:
logger.info(f"API调用成功: {json.dumps(log_data)}")
# 在API调用中使用
def monitored_api_call(api_func, *args, **kwargs):
"""包装API调用以添加监控"""
start_time = time.time()
try:
result = api_func(*args, **kwargs)
response_time = time.time() - start_time
log_api_call(api_func.__name__, 200, response_time)
return result
except Exception as e:
response_time = time.time() - start_time
log_api_call(api_func.__name__, 0, response_time, error=e)
raise
7.4 使用环境变量管理凭证
import os
from dotenv import load_dotenv
# 加载.env文件
load_dotenv()
class EbayConfig:
"""配置管理类"""
def __init__(self):
self.client_id = os.getenv("EBAY_CLIENT_ID")
self.client_secret = os.getenv("EBAY_CLIENT_SECRET")
self.redirect_uri = os.getenv("EBAY_REDIRECT_URI")
# 验证配置
if not all([self.client_id, self.client_secret, self.redirect_uri]):
raise ValueError("缺少必要的环境变量")
def get_auth_url(self, scope="https://api.ebay.com/oauth/api_scope"):
"""生成授权URL"""
auth_url = "https://auth.ebay.com/oauth2/authorize"
params = {
"client_id": self.client_id,
"response_type": "code",
"redirect_uri": self.redirect_uri,
"scope": scope
}
return f"{auth_url}?{ '&'.join([f'{k}={v}' for k,v in params.items()]) }"
# .env文件示例
"""
EBAY_CLIENT_ID=你的ClientID
EBAY_CLIENT_SECRET=你的ClientSecret
EBAY_REDIRECT_URI=https://your-app.com/callback
"""
八、完整项目示例：竞品监控系统
8.1 项目结构
ebay_monitor/
├── config/
│ ├── __init__.py
│ └── settings.py
├── api/
│ ├── __init__.py
│ ├── client.py
│ └── auth.py
├── utils/
│ ├── __init__.py
│ ├── logger.py
│ └── parser.py
├── data/
│ └── storage.py
├── main.py
└── requirements.txt
8.2 核心代码实现
config/settings.py
import os
from dataclasses import dataclass
@dataclass
class EbaySettings:
client_id: str = os.getenv("EBAY_CLIENT_ID")
client_secret: str = os.getenv("EBAY_CLIENT_SECRET")
redirect_uri: str = os.getenv("EBAY_REDIRECT_URI")
rate_limit: int = 5 # 每秒请求数
max_retries: int = 3
timeout: int = 30
api/auth.py
import requests
import base64
import json
from datetime import datetime, timedelta
class EbayAuth:
def __init__(self, settings):
self.settings = settings
self.token_data = None
def get_user_token(self, auth_code):
"""获取用户Token"""
token_url = "https://api.ebay.com/identity/v1/oauth2/token"
credentials = f"{self.settings.client_id}:{self.settings.client_secret}"
encoded_credentials = base64.b64encode(credentials.encode()).decode()
headers = {
"Authorization": f"Basic {encoded_credentials}",
"Content-Type": "application/x-www-form-urlencoded"
}
data = {
"grant_type": "authorization_code",
"code": auth_code,
"redirect_uri": self.settings.redirect_uri
}
response = requests.post(token_url, headers=headers, data=data)
response.raise_for_status()
self.token_data = response.json()
self.token_data["expires_at"] = datetime.now() + timedelta(seconds=self.token_data["expires_in"])
return self.token_data
def get_app_token(self):
"""获取应用Token"""
token_url = "https://api.ebay.com/identity/v1/oauth2/token"
credentials = f"{self.settings.client_id}:{self.settings.client_secret}"
encoded_credentials = base64.b64encode(credentials.encode()).decode()
headers = {
"Authorization": f"Basic {encoded_credentials}",
"Content-Type": "application/x-www-form-urlencoded"
}
data = {
"grant_type": "client_credentials",
"scope": "https://api.ebay.com/oauth/api_scope"
}
response = requests.post(token_url, headers=headers, data=data)
response.raise_for_status()
return response.json()
def is_token_expired(self):
"""检查Token是否过期"""
if not self.token_data or "expires_at" not in self.token_data:
return True
return datetime.now() >= self.token_data[“expires_at”]
def refresh_user_token(self):
“””刷新用户Token”””
if not self.token_data or “refresh_token” not in self.token_data:
raise ValueError(“没有可用的refresh_token”)
token_url = “https://api.ebay.com/identity/v1/oauth2/token”
credentials = f”{self.settings.client_id}:{self.settings.client_secret}”
encoded_credentials = base64.b64encode(credentials.encode()).decode()
headers = {
“Authorization”: f”Basic {encoded_credentials}”,
“Content-Type”: “application/x-www-form-urlencoded”
}
data = {
“grant_type”: “refresh_token”,
“refresh_token”: self.token_data[“refresh_token”]
}
response = requests.post(token_url, headers=headers, data=data)
response.raise_for_status()
self.token_data = response.json()
self.token_data[“expires_at”] = datetime.now() + timedelta(seconds=self.token_data[“expires_in”])
return self.token_data
api/client.py
import requests
import time
import json
from typing import Optional, Dict, Any
from datetime import datetime
class EbayAPIClient:
def __init__(self, auth, settings):
self.auth = auth
self.settings = settings
self.last_request_time = 0
def _ensure_token(self):
“””确保Token有效”””
if self.auth.is_token_expired():
try:
self.auth.refresh_user_token()
except:
# 如果刷新失败，获取新的App Token
token_data = self.auth.get_app_token()
self.auth.token_data = {
“access_token”: token_data[“access_token”],
“expires_at”: datetime.now() + timedelta(seconds=token_data[“expires_in”])
}
def _rate_limit_wait(self):
“””限流等待”””
current_time = time.time()
time_since_last = current_time – self.last_request_time
if time_since_last < (1.0 / self.settings.rate_limit):
sleep_time = (1.0 / self.settings.rate_limit) - time_since_last
time.sleep(sleep_time)
self.last_request_time = time.time()
def _make_request(self, method, endpoint, params=None, data=None, max_retries=None):
"""基础请求方法"""
if max_retries is None:
max_retries = self.settings.max_retries
self._ensure_token()
self._rate_limit_wait()
url = f"https://api.ebay.com{endpoint}"
headers = {
"Authorization": f"Bearer {self.auth.token_data['access_token']}",
"Content-Type": "application/json",
"X-EBAY-C-MARKETPLACE-ID": "EBAY_US"
}
for attempt in range(max_retries):
try:
if method.upper() == "GET":
response = requests.get(url, headers=headers, params=params, timeout=self.settings.timeout)
elif method.upper() == "POST":
response = requests.post(url, headers=headers, json=data, timeout=self.settings.timeout)
else:
raise ValueError(f"不支持的HTTP方法: {method}")
# 记录日志
self._log_request(method, endpoint, response.status_code, attempt)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# 触发限流，指数退避
wait_time = (2 ** attempt) + 0.5
print(f"触发限流，等待 {wait_time} 秒...")
time.sleep(wait_time)
continue
elif response.status_code == 401:
# Token问题，刷新后重试
self._ensure_token()
headers["Authorization"] = f"Bearer {self.auth.token_data['access_token']}"
continue
else:
response.raise_for_status()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise e
time.sleep(2 ** attempt)
raise Exception(f"达到最大重试次数，请求失败")
def _log_request(self, method, endpoint, status_code, attempt):
"""记录请求日志"""
logger = logging.getLogger(__name__)
log_data = {
"timestamp": datetime.now().isoformat(),
"method": method,
"endpoint": endpoint,
"status_code": status_code,
"attempt": attempt
}
logger.info(json.dumps(log_data))
def search_items(self, keyword, limit=10, filters=None):
"""搜索商品"""
endpoint = "/buy/browse/v1/item_summary/search"
params = {"q": keyword, "limit": limit}
if filters:
params["filter"] = filters
return self._make_request("GET", endpoint, params=params)
def get_item_details(self, item_id):
"""获取商品详情"""
endpoint = f"/buy/browse/v1/item/{item_id}"
return self._make_request("GET", endpoint)
def get_orders(self, limit=50, days=30):
"""获取订单"""
from datetime import datetime, timedelta
start_date = (datetime.now() - timedelta(days=days)).isoformat()
endpoint = "/sell/order/v2/order"
params = {
"limit": limit,
"filter": f"creationDate:[{start_date}..]"
}
return self._make_request("GET", endpoint, params=params)
utils/parser.py
from typing import List, Dict, Any
class EbayDataParser:
@staticmethod
def parse_item_summary(item: Dict[str, Any]) -> Dict[str, Any]:
“””解析商品摘要”””
return {
“item_id”: item.get(“itemId”),
“title”: item.get(“title”),
“price”: item.get(“price”, {}).get(“value”),
“currency”: item.get(“price”, {}).get(“currency”),
“condition”: item.get(“condition”),
“seller”: item.get(“seller”, {}).get(“username”),
“feedback_score”: item.get(“seller”, {}).get(“feedbackScore”),
“image_url”: item.get(“image”, {}).get(“imageUrl”),
“item_url”: item.get(“itemWebUrl”),
“availability”: item.get(“availability”, {}).get(“status”)
}
@staticmethod
def parse_order(order: Dict[str, Any]) -> Dict[str, Any]:
“””解析订单”””
return {
“order_id”: order.get(“orderId”),
“order_status”: order.get(“orderStatus”),
“creation_date”: order.get(“creationDate”),
“total”: order.get(“amount”, {}).get(“value”),
“currency”: order.get(“amount”, {}).get(“currency”),
“buyer”: order.get(“buyer”, {}).get(“username”),
“line_items_count”: len(order.get(“lineItems”, []))
}
@staticmethod
def batch_parse(items: List[Dict[str, Any]], parser_func) -> List[Dict[str, Any]]:
“””批量解析”””
return [parser_func(item) for item in items]
data/storage.py
import sqlite3
import pandas as pd
from datetime import datetime
from typing import List, Dict, Any
class DataStorage:
def __init__(self, db_path=”ebay_data.db”):
self.db_path = db_path
self._init_db()
def _init_db(self):
“””初始化数据库”””
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 创建商品表
cursor.execute(“””
CREATE TABLE IF NOT EXISTS items (
item_id TEXT PRIMARY KEY,
title TEXT,
price REAL,
currency TEXT,
condition TEXT,
seller TEXT,
feedback_score INTEGER,
image_url TEXT,
item_url TEXT,
availability TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
“””)
# 创建订单表
cursor.execute(“””
CREATE TABLE IF NOT EXISTS orders (
order_id TEXT PRIMARY KEY,
order_status TEXT,
creation_date TEXT,
total REAL,
currency TEXT,
buyer TEXT,
line_items_count INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
“””)
conn.commit()
conn.close()
def save_items(self, items: List[Dict[str, Any]]):
“””保存商品数据”””
if not items:
return
conn = sqlite3.connect(self.db_path)
df = pd.DataFrame(items)
# 去重
existing_ids = pd.read_sql(“SELECT item_id FROM items”, conn)[“item_id”].tolist()
df = df[~df[“item_id”].isin(existing_ids)]
if not df.empty:
df.to_sql(‘items’, conn, if_exists=’append’, index=False)
print(f”保存 {len(df)} 条商品记录”)
conn.close()
def save_orders(self, orders: List[Dict[str, Any]]):
“””保存订单数据”””
if not orders:
return
conn = sqlite3.connect(self.db_path)
df = pd.DataFrame(orders)
# 去重
existing_ids = pd.read_sql(“SELECT order_id FROM orders”, conn)[“order_id”].tolist()
df = df[~df[“order_id”].isin(existing_ids)]
if not df.empty:
df.to_sql(‘orders’, conn, if_exists=’append’, index=False)
print(f”保存 {len(df)} 条订单记录”)
conn.close()
def get_items_by_seller(self, seller: str) -> pd.DataFrame:
“””获取指定卖家的商品”””
conn = sqlite3.connect(self.db_path)
query = “SELECT * FROM items WHERE seller = ?”
df = pd.read_sql(query, conn, params=[seller])
conn.close()
return df
def get_price_statistics(self) -> Dict[str, Any]:
“””获取价格统计”””
conn = sqlite3.connect(self.db_path)
query = “””
SELECT
AVG(price) as avg_price,
MIN(price) as min_price,
MAX(price) as max_price,
COUNT(*) as item_count
FROM items
WHERE price > 0
“””
df = pd.read_sql(query, conn)
conn.close()
return {
“avg_price”: df.iloc[0][“avg_price”],
“min_price”: df.iloc[0][“min_price”],
“max_price”: df.iloc[0][“max_price”],
“item_count”: df.iloc[0][“item_count”]
}
main.py
import os
import time
from datetime import datetime
from config.settings import EbaySettings
from api.auth import EbayAuth
from api.client import EbayAPIClient
from utils.parser import EbayDataParser
from data.storage import DataStorage
def main():
# 1. 初始化配置
settings = EbaySettings()
# 2. 初始化认证
auth = EbayAuth(settings)
# 3. 初始化API客户端
api = EbayAPIClient(auth, settings)
# 4. 初始化存储
storage = DataStorage()
# 5. 执行监控任务
print(f”开始竞品监控: {datetime.now()}”)
# 搜索iPhone商品
try:
results = api.search_items(
keyword=”iPhone 15 Pro”,
limit=50,
filters=”price:[500..1500],condition:{New,Used}”
)
items = results.get(“itemSummaries”, [])
print(f”找到 {len(items)} 个商品”)
# 解析数据
parsed_items = EbayDataParser.batch_parse(items, EbayDataParser.parse_item_summary)
# 保存数据
storage.save_items(parsed_items)
# 获取统计信息
stats = storage.get_price_statistics()
print(f”价格统计: {stats}”)
except Exception as e:
print(f”执行失败: {e}”)
raise
print(“监控任务完成”)
if __name__ == “__main__”:
main()
requirements.txt
requests>=2.28.0
pandas>=1.5.0
python-dotenv>=0.19.0
九、性能优化与扩展建议
9.1 使用异步请求
对于大量数据获取，可以使用
aiohttp
实现异步请求：
import aiohttp
import asyncio
from typing import List
class AsyncEbayAPI:
def __init__(self, access_token):
self.access_token = access_access_token
self.base_url = “https://api.ebay.com/buy/browse/v1”
async def search_items_async(self, keyword, limit=10):
“””异步搜索商品”””
url = f”{self.base_url}/item_summary/search”
headers = {
“Authorization”: f”Bearer {self.access_token}”,
“Content-Type”: “application/json”
}
params = {“q”: keyword, “limit”: limit}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, params=params) as response:
return await response.json()
async def batch_search(self, keywords: List[str], limit_per_keyword=10):
“””批量异步搜索”””
tasks = [self.search_items_async(keyword, limit_per_keyword) for keyword in keywords]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# 使用示例
async def main():
api = AsyncEbayAPI(“你的access_token”)
keywords = [“iPhone”, “Samsung”, “Pixel”]
results = await api.batch_search(keywords)
print(f”批量搜索完成: {len(results)} 个结果”)
# 运行
# asyncio.run(main())
9.2 使用缓存
from functools import lru_cache
import hashlib
class CachedEbayAPI:
def __init__(self, api_client):
self.api = api_client
@lru_cache(maxsize=128)
def search_items_cached(self, keyword, limit, filters):
“””带缓存的搜索”””
# 生成缓存键
key = hashlib.md5(f”{keyword}_{limit}_{filters}”.encode()).hexdigest()
# 检查缓存（可以使用Redis等外部缓存）
# 这里简化为使用lru_cache
return self.api.search_items(keyword, limit, filters)
9.3 分布式部署
对于大规模应用，考虑使用：
Celery
：任务队列
Redis
：缓存和消息队列
PostgreSQL
：替代SQLite
Docker
：容器化部署
十、总结与最佳实践
10.1 关键要点回顾
认证是基础
：正确配置OAuth 2.0流程，处理Token刷新
限流是必须
：实现请求队列和指数退避策略
错误处理
：全面捕获和处理各种API错误
数据验证
：确保数据完整性和准确性
日志监控
：记录所有API调用和错误
安全第一
：使用环境变量管理凭证，避免硬编码
10.2 推荐的开发流程
开发阶段
：使用Sandbox环境测试
测试阶段
：实现完整的单元测试和集成测试
部署阶段
：使用环境变量和配置管理
监控阶段
：实时监控API调用和错误率
优化阶段
：根据监控数据优化性能
10.3 常见陷阱与避免方法
陷阱
后果
解决方案
硬编码凭证
安全风险
使用环境变量或密钥管理服务
无重试机制
请求失败
实现指数退避重试
忽略限流
账户被封
实现请求队列和限流
无错误日志
问题难排查
完整的日志记录
同步处理大量数据
性能差
使用异步或批量处理
不验证数据
数据污染
实现数据验证层
10.4 进一步学习资源
官方文档
：
eBay Developer Portal
API参考
：
eBay API Documentation
社区支持
：
eBay Developer Forum
GitHub示例
：搜索 “eBay API Python”
结语
eBay API调用是一个系统工程，需要开发者在认证、请求、错误处理、数据解析等各个环节都做到严谨细致。通过本文的系统讲解和完整代码示例，相信你已经掌握了从零开始调用eBay API的核心技能。
记住，成功的API集成不仅仅是让代码跑起来，更重要的是构建一个健壮、可维护、可扩展的系统。在实际开发中，务必遵循最佳实践，做好错误处理和监控，这样才能在复杂的生产环境中游刃有余。
如果你在实践中遇到任何问题，欢迎参考本文的错误处理章节，或查阅eBay官方文档和社区支持。祝你在eBay API开发之路上取得成功！