在全球化采购场景中，高效获取多平台商品搜索结果是选品决策、价格对比、市场分析的核心基础。item_search接口作为跨平台商品检索的核心工具，需适配 Amazon、eBay、Lazada、速卖通等主流跨境电商平台，实现关键词搜索、多条件筛选、数据整合与标准化。本文将从业务场景出发，详解多平台搜索接口的共性与差异，提供标准化对接方案，帮助开发者构建覆盖全球市场的采购搜索系统。
一、业务场景与核心价值
二、通用对接框架（跨平台共性流程）
三、主流平台搜索接口差异对比
四、代码实现：多平台统一搜索框架
import requests
import time
import json
from abc import ABC, abstractmethod
import base64
import hashlib
import xmltodict
from googletrans import Translator # 多语言翻译
from typing import Dict, List, Optional
# 实时汇率服务（示例，实际需对接API）
class ExchangeRateService:
@staticmethod
def get_rate(currency: str, target: str = "CNY") -> float:
"""获取货币汇率（目标货币默认人民币）"""
rates = {
"USD": 7.2, "EUR": 8.0, "GBP": 9.2,
"SGD": 5.3, "MYR": 1.7, "JPY": 0.05
}
return rates.get(currency, 1.0)
class BaseSearchApi(ABC):
"""搜索接口抽象基类，定义统一规范"""
def __init__(self, platform: str, credentials: Dict, region: str):
self.platform = platform # 平台名称
self.credentials = credentials # 认证凭证
self.region = region # 区域（如us/gb/sg）
self.translator = Translator() # 多语言翻译器
self.rate_service = ExchangeRateService() # 汇率服务
self.token = None # 访问令牌
self.token_expire = 0 # 令牌过期时间（时间戳）
@abstractmethod
def _get_auth_headers(self) -> Dict:
"""获取认证请求头（各平台实现）"""
pass
@abstractmethod
def _map_params(self, search_params: Dict) -> Dict:
"""将通用参数映射为平台特有参数"""
pass
@abstractmethod
def _parse_response(self, raw_response) -> List[Dict]:
"""解析平台原始响应，提取商品列表"""
pass
def _translate_keyword(self, keyword: str) -> str:
"""将关键词翻译为目标平台语言（基于区域）"""
lang_map = {"us": "en", "de": "de", "fr": "fr", "sg": "en", "my": "ms"}
target_lang = lang_map.get(self.region, "en")
try:
return self.translator.translate(keyword, dest=target_lang).text
except:
return keyword # 翻译失败返回原文
def standardize_item(self, raw_item: Dict) -> Dict:
"""标准化单条商品数据"""
# 1. 基础信息
standardized = {
"platform": self.platform,
"item_id": self._extract(raw_item, "item_id"),
"title": self._extract(raw_item, "title"),
"title_original": self._extract(raw_item, "title"), # 保留原文
"main_image": self._extract(raw_item, "main_image"),
"url": self._extract(raw_item, "url")
}
# 2. 价格信息（统一转换为人民币）
price = self._extract(raw_item, "price")
currency = self._extract(raw_item, "currency")
standardized["price"] = {
"original": price,
"currency": currency,
"cny": round(float(price) * self.rate_service.get_rate(currency), 2)
}
# 3. 核心指标
standardized["metrics"] = {
"sales": self._extract(raw_item, "sales"), # 销量
"rating": self._extract(raw_item, "rating"), # 评分
"review_count": self._extract(raw_item, "review_count"), # 评论数
"stock": self._extract(raw_item, "stock") # 库存
}
# 4. 卖家信息
standardized["seller"] = {
"id": self._extract(raw_item, "seller_id"),
"name": self._extract(raw_item, "seller_name"),
"rating": self._extract(raw_item, "seller_rating") # 卖家好评率
}
return standardized
def _extract(self, raw_item: Dict, field: str) -> Optional[str]:
"""提取字段（各平台需重写）"""
raise NotImplementedError
def search(self, keyword: str, **kwargs) -> Dict:
"""
统一搜索接口
:param keyword: 搜索关键词（中文/英文）
:param kwargs: 可选参数：
- price_min: 最低价格
- price_max: 最高价格
- category_id: 分类ID
- sort: 排序方式（price_asc/price_desc/sales）
- page: 页码
- page_size: 每页条数
:return: 标准化搜索结果
"""
try:
# 1. 处理关键词（翻译为平台语言）
platform_keyword = self._translate_keyword(keyword)
# 2. 组装通用搜索参数
search_params = {
"keyword": platform_keyword,
"price_min": kwargs.get("price_min"),
"price_max": kwargs.get("price_max"),
"category_id": kwargs.get("category_id"),
"sort": kwargs.get("sort", "default"),
"page": kwargs.get("page", 1),
"page_size": kwargs.get("page_size", 20)
}
# 3. 映射为平台特有参数
platform_params = self._map_params(search_params)
# 4. 获取认证头
headers = self._get_auth_headers()
# 5. 发送请求
response = self._send_request(platform_params, headers)
if not response:
return {"success": False, "msg": f"{self.platform}搜索无结果"}
# 6. 解析响应
raw_items = self._parse_response(response)
if not raw_items:
return {"success": False, "msg": f"{self.platform}无匹配商品"}
# 7. 标准化数据
standardized_items = [self.standardize_item(item) for item in raw_items]
# 8. 返回结果
return {
"success": True,
"platform": self.platform,
"total": len(standardized_items),
"page": search_params["page"],
"items": standardized_items
}
except Exception as e:
return {"success": False, "msg": f"{self.platform}搜索失败: {str(e)}"}
def _send_request(self, params: Dict, headers: Dict) -> Optional[Dict]:
"""发送请求（带重试机制）"""
retry_count = 0
max_retries = 3
while retry_count < max_retries:
try:
response = requests.get(
url=self.endpoint,
params=params,
headers=headers,
timeout=15
)
response.raise_for_status()
# 处理XML响应（如Amazon）
if "xml" in response.headers.get("Content-Type", ""):
return xmltodict.parse(response.text, dict_constructor=dict)
return response.json()
except requests.exceptions.RequestException as e:
retry_count += 1
if retry_count >= max_retries:
raise Exception(f"请求失败（{retry_count}次重试）: {str(e)}")
time.sleep(2 **retry_count) # 指数退避
return None
class EbaySearchApi(BaseSearchApi):
"""eBay搜索接口实现"""
def __init__(self, credentials: Dict, region: str = "us"):
super().__init__("ebay", credentials, region)
self.endpoint = "https://api.ebay.com/buy/browse/v1/item_summary/search"
self.marketplace_id = self._get_marketplace_id(region)
def _get_marketplace_id(self, region: str) -> str:
"""映射区域到eBay站点ID"""
return {"us": "EBAY-US", "uk": "EBAY-GB", "de": "EBAY-DE"}.get(region, "EBAY-US")
def _get_auth_headers(self) -> Dict:
"""获取eBay认证头（OAuth 2.0）"""
if not self.token or time.time() > self.token_expire - 60:
# 刷新Token
auth_str = f"{self.credentials['client_id']}:{self.credentials['client_secret']}"
auth_b64 = base64.b64encode(auth_str.encode()).decode()
resp = requests.post(
"https://api.ebay.com/identity/v1/oauth2/token",
headers={"Authorization": f"Basic {auth_b64}"},
data={"grant_type": "client_credentials", "scope": "https://api.ebay.com/oauth/api_scope/buy.browse"}
)
resp_data = resp.json()
self.token = resp_data["access_token"]
self.token_expire = time.time() + resp_data["expires_in"]
return {
"Authorization": f"Bearer {self.token}",
"X-EBAY-C-MARKETPLACE-ID": self.marketplace_id
}
def _map_params(self, search_params: Dict) -> Dict:
"""映射通用参数到eBay参数"""
params = {
"q": search_params["keyword"],
"limit": search_params["page_size"],
"offset": (search_params["page"] - 1) * search_params["page_size"]
}
# 价格筛选
if search_params["price_min"] and search_params["price_max"]:
params["filter"] = f"price:[{search_params['price_min']}..{search_params['price_max']}]"
# 分类筛选
if search_params["category_id"]:
params["filter"] = f"{params.get('filter', '')},category_id:{search_params['category_id']}"
# 排序
sort_map = {
"price_asc": "price_asc",
"price_desc": "price_desc",
"sales": "popularity"
}
if search_params["sort"] in sort_map:
params["sort"] = sort_map[search_params["sort"]]
return params
def _parse_response(self, raw_response: Dict) -> List[Dict]:
"""解析eBay响应"""
return raw_response.get("itemSummaries", [])
def _extract(self, raw_item: Dict, field: str) -> Optional[str]:
"""提取eBay字段"""
mappings = {
"item_id": raw_item.get("itemId"),
"title": raw_item.get("title"),
"main_image": raw_item.get("image", {}).get("imageUrl"),
"url": raw_item.get("itemWebUrl"),
"price": raw_item.get("price", {}).get("value"),
"currency": raw_item.get("price", {}).get("currency"),
"sales": raw_item.get("soldQuantity"),
"rating": raw_item.get("averageRating"),
"review_count": raw_item.get("reviewCount"),
"stock": raw_item.get("inventory", {}).get("availableQuantity"),
"seller_id": raw_item.get("seller", {}).get("sellerId"),
"seller_name": raw_item.get("seller", {}).get("username"),
"seller_rating": raw_item.get("seller", {}).get("feedbackPercentage")
}
return mappings.get(field)
class AmazonSearchApi(BaseSearchApi):
"""Amazon搜索接口实现（简化版）"""
def __init__(self, credentials: Dict, region: str = "us"):
super().__init__("amazon", credentials, region)
self.domain = {"us": "com", "uk": "co.uk", "de": "de"}.get(region, "com")
self.endpoint = f"https://webservices.amazon.{self.domain}/onca/xml"
def _get_auth_headers(self) -> Dict:
"""Amazon无需请求头，依赖URL签名"""
return {}
def _map_params(self, search_params: Dict) -> Dict:
"""映射通用参数到Amazon参数"""
params = {
"Service": "AWSECommerceService",
"AWSAccessKeyId": self.credentials["access_key"],
"AssociateTag": self.credentials["associate_tag"],
"Operation": "ItemSearch",
"Keywords": search_params["keyword"],
"ItemPage": search_params["page"],
"ResponseGroup": "ItemAttributes,Offers,Images,SalesRank",
"Timestamp": time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
}
# 价格筛选
if search_params["price_min"]:
params["LowPrice"] = search_params["price_min"]
if search_params["price_max"]:
params["HighPrice"] = search_params["price_max"]
# 分类筛选
if search_params["category_id"]:
params["SearchIndex"] = search_params["category_id"]
# 排序
sort_map = {
"price_asc": "price",
"price_desc": "price-desc-rank",
"sales": "salesrank"
}
if search_params["sort"] in sort_map:
params["Sort"] = sort_map[search_params["sort"]]
# 生成签名
params["Signature"] = self._generate_signature(params)
return params
def _generate_signature(self, params: Dict) -> str:
"""生成Amazon SigV4签名（简化版）"""
sorted_params = sorted(params.items(), key=lambda x: x[0])
query_str = "&".join([f"{k}={urllib.parse.quote_plus(str(v))}" for k, v in sorted_params])
sign_str = f"GET\nwebservices.amazon.{self.domain}\n/onca/xml\n{query_str}"
signature = hmac.new(
self.credentials["secret_key"].encode(),
sign_str.encode(),
hashlib.sha256
).digest()
return base64.b64encode(signature).decode()
def _parse_response(self, raw_response: Dict) -> List[Dict]:
"""解析Amazon XML响应"""
return raw_response.get("ItemSearchResponse", {}).get("Items", {}).get("Item", [])
def _extract(self, raw_item: Dict, field: str) -> Optional[str]:
"""提取Amazon字段"""
mappings = {
"item_id": raw_item.get("ASIN"),
"title": raw_item.get("ItemAttributes", {}).get("Title"),
"main_image": raw_item.get("LargeImage", {}).get("URL"),
"url": f"https://www.amazon.{self.domain}/dp/{raw_item.get('ASIN')}",
"price": raw_item.get("Offers", {}).get("Offer", {}).get("OfferListing", {}).get("Price", {}).get("Amount"),
"currency": raw_item.get("Offers", {}).get("Offer", {}).get("OfferListing", {}).get("Price", {}).get("CurrencyCode"),
"sales": None, # Amazon不直接返回销量，需通过SalesRank推断
"rating": raw_item.get("CustomerReviews", {}).get("AverageRating"),
"review_count": raw_item.get("CustomerReviews", {}).get("TotalReviews"),
"stock": raw_item.get("Offers", {}).get("Offer", {}).get("OfferListing", {}).get("Availability"),
"seller_id": "Amazon" if raw_item.get("IsEligibleForPrime") else None,
"seller_name": "Amazon" if raw_item.get("IsEligibleForPrime") else None,
"seller_rating": None
}
return mappings.get(field)
# 使用示例
if __name__ == "__main__":
# 1. 配置平台凭证
ebay_creds = {
"client_id": "your_ebay_client_id",
"client_secret": "your_ebay_client_secret"
}
amazon_creds = {
"access_key": "your_amazon_access_key",
"secret_key": "your_amazon_secret_key",
"associate_tag": "your_associate_tag"
}
# 2. 初始化多平台API客户端
ebay_api = EbaySearchApi(ebay_creds, region="us") # 美国站
amazon_api = AmazonSearchApi(amazon_creds, region="us") # 美国站
# 3. 统一搜索（关键词"无线耳机"，价格10-100美元，按销量排序）
ebay_result = ebay_api.search(
keyword="无线耳机",
price_min=10,
price_max=100,
sort="sales",
page=1,
page_size=10
)
amazon_result = amazon_api.search(
keyword="无线耳机",
price_min=10,
price_max=100,
sort="sales",
page=1,
page_size=10
)
# 4. 输出整合结果
print(f"eBay搜索结果：{json.dumps(ebay_result, indent=2, ensure_ascii=False)}")
print(f"Amazon搜索结果：{json.dumps(amazon_result, indent=2, ensure_ascii=False)}")
# 5. 横向对比（示例：提取最低价格商品）
if ebay_result["success"] and amazon_result["success"]:
all_items = ebay_result["items"] + amazon_result["items"]
cheapest_item = min(all_items, key=lambda x: x["price"]["cny"])
print(f"\n最低价商品：{cheapest_item['title']}，{cheapest_item['price']['cny']}元（{cheapest_item['platform']}）")