在全球化贸易中，跨国采购依赖高效获取多平台商品详情的能力。item_get接口作为核心工具，需适配 Amazon、eBay、Lazada、速卖通等主流跨境平台，实现商品信息的统一采集与分析。本文将从业务场景出发，详解多平台接口的共性与差异，提供标准化对接方案，帮助开发者构建稳定高效的跨国采购数据系统。
一、业务场景与核心挑战
二、通用对接框架（跨平台共性流程）
三、主流平台接口差异对比
四、代码实现：多平台统一对接框架
import requests
import time
import json
from abc import ABC, abstractmethod
import base64
import hashlib
from googletrans import Translator # 需安装：pip install googletrans==4.0.0-rc1
# 汇率转换工具（示例：实时汇率需对接API）
EXCHANGE_RATES = {
"USD": 7.2, # 美元→人民币
"EUR": 8.0, # 欧元→人民币
"SGD": 5.3, # 新加坡元→人民币
"MYR": 1.7 # 马来西亚令吉→人民币
}
class BaseItemApi(ABC):
"""抽象基类：定义item_get接口的统一规范"""
def __init__(self, platform, credentials, region):
self.platform = platform # 平台名称（amazon/ebay/lazada/aliexpress）
self.credentials = credentials # 认证凭证（字典）
self.region = region # 区域（如us/gb/sg）
self.translator = Translator() # 多语言翻译器
self.token = None
self.token_expire = 0
@abstractmethod
def get_token(self):
"""获取访问令牌（各平台实现不同）"""
pass
@abstractmethod
def fetch_raw_item(self, item_id):
"""调用平台API，获取原始商品数据"""
pass
def standardize_data(self, raw_data):
"""标准化商品数据（统一字段格式）"""
# 1. 基础信息
standard = {
"platform": self.platform,
"item_id": self._extract_field(raw_data, "item_id"),
"title": self._translate(self._extract_field(raw_data, "title")),
"title_original": self._extract_field(raw_data, "title"),
"main_image": self._extract_field(raw_data, "main_image"),
"url": self._extract_field(raw_data, "url")
}
# 2. 价格信息（统一转换为人民币）
price = self._extract_field(raw_data, "price")
currency = self._extract_field(raw_data, "currency")
standard["price"] = {
"original": price,
"currency": currency,
"cn_yuan": round(float(price) * EXCHANGE_RATES.get(currency, 1), 2)
}
# 3. 库存与销量
standard["stock"] = self._extract_field(raw_data, "stock")
standard["sales"] = self._extract_field(raw_data, "sales")
# 4. 卖家信息
standard["seller"] = {
"id": self._extract_field(raw_data, "seller_id"),
"name": self._extract_field(raw_data, "seller_name"),
"rating": self._extract_field(raw_data, "seller_rating")
}
return standard
def _extract_field(self, raw_data, field):
"""提取字段（各平台需重写）"""
raise NotImplementedError
def _translate(self, text, dest="zh-cn"):
"""翻译文本（保留英文/数字）"""
if not text:
return ""
try:
return self.translator.translate(text, dest=dest).text
except:
return text # 翻译失败返回原文
def get_item(self, item_id):
"""对外统一接口：获取标准化商品详情"""
try:
# 1. 获取令牌
self.get_token()
# 2. 调用平台API
raw_data = self.fetch_raw_item(item_id)
if not raw_data:
return {"success": False, "msg": f"{self.platform}商品不存在"}
# 3. 标准化数据
standard_data = self.standardize_data(raw_data)
return {"success": True, "data": standard_data}
except Exception as e:
return {"success": False, "msg": f"{self.platform}接口错误: {str(e)}"}
class AmazonItemApi(BaseItemApi):
"""Amazon平台实现"""
def __init__(self, credentials, region="us"):
super().__init__("amazon", credentials, region)
self.host = f"webservices.amazon.{self._get_region_domain(region)}"
self.endpoint = f"https://{self.host}/onca/xml"
def _get_region_domain(self, region):
"""转换区域为域名后缀"""
domains = {"us": "com", "uk": "co.uk", "de": "de", "jp": "co.jp"}
return domains.get(region, "com")
def get_token(self):
"""Amazon无需Token，依赖签名"""
return True
def fetch_raw_item(self, item_id):
"""调用Amazon ItemLookup接口"""
params = {
"Service": "AWSECommerceService",
"AWSAccessKeyId": self.credentials["access_key"],
"AssociateTag": self.credentials["associate_tag"],
"Operation": "ItemLookup",
"ItemId": item_id,
"ResponseGroup": "ItemAttributes,Offers,Images,SalesRank",
"Timestamp": time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
}
# 生成签名（简化版，实际需用SigV4）
params["Signature"] = self._generate_signature(params)
response = requests.get(self.endpoint, params=params, timeout=15)
# 解析XML（此处简化，实际需用xmltodict）
return {"raw_xml": response.text} # 实际项目需解析为字典
def _generate_signature(self, params):
"""生成AWS签名（简化实现）"""
sorted_params = sorted(params.items(), key=lambda x: x[0])
sign_str = "&".join([f"{k}={v}" for k, v in sorted_params])
return hashlib.sha256((sign_str + self.credentials["secret_key"]).encode()).hexdigest()
def _extract_field(self, raw_data, field):
"""提取Amazon字段（基于解析后的XML）"""
# 实际需根据XML结构提取，此处为示例
mappings = {
"item_id": "B07VGRJDFY", # 示例ASIN
"title": "Wireless Headphones with Noise Cancellation",
"main_image": "https://example.com/image.jpg",
"url": "https://www.amazon.com/dp/B07VGRJDFY",
"price": "99.99",
"currency": "USD",
"stock": "50",
"sales": "1200",
"seller_id": "A123456",
"seller_name": "Amazon Official",
"seller_rating": "4.8"
}
return mappings.get(field, "")
class EbayItemApi(BaseItemApi):
"""eBay平台实现"""
def __init__(self, credentials, region="us"):
super().__init__("ebay", credentials, region)
self.marketplace_id = self._get_marketplace_id(region)
self.endpoint = "https://api.ebay.com/buy/browse/v1/item"
def _get_marketplace_id(self, region):
"""转换区域为eBay站点ID"""
return {"us": "EBAY-US", "uk": "EBAY-GB", "de": "EBAY-DE"}.get(region, "EBAY-US")
def get_token(self):
"""获取eBay OAuth Token"""
if self.token and time.time() < self.token_expire - 60:
return self.token
# 调用eBay Token接口
auth_str = f"{self.credentials['client_id']}:{self.credentials['client_secret']}"
auth_base64 = base64.b64encode(auth_str.encode()).decode()
response = requests.post(
"https://api.ebay.com/identity/v1/oauth2/token",
headers={"Authorization": f"Basic {auth_base64}"},
data={"grant_type": "client_credentials", "scope": "https://api.ebay.com/oauth/api_scope/buy.browse"},
timeout=10
)
result = response.json()
self.token = result["access_token"]
self.token_expire = time.time() + result["expires_in"]
return self.token
def fetch_raw_item(self, item_id):
"""调用eBay item接口"""
headers = {
"Authorization": f"Bearer {self.token}",
"X-EBAY-C-MARKETPLACE-ID": self.marketplace_id
}
response = requests.get(f"{self.endpoint}/{item_id}", headers=headers, timeout=15)
return response.json()
def _extract_field(self, raw_data, field):
"""提取eBay字段"""
mappings = {
"item_id": raw_data.get("itemId"),
"title": raw_data.get("title"),
"main_image": raw_data.get("image", {}).get("imageUrl"),
"url": raw_data.get("itemWebUrl"),
"price": raw_data.get("price", {}).get("value"),
"currency": raw_data.get("price", {}).get("currency"),
"stock": raw_data.get("inventory", {}).get("availableQuantity"),
"sales": raw_data.get("soldQuantity"),
"seller_id": raw_data.get("seller", {}).get("sellerId"),
"seller_name": raw_data.get("seller", {}).get("username"),
"seller_rating": raw_data.get("seller", {}).get("feedbackPercentage")
}
return mappings.get(field, "")
# 其他平台（Lazada/速卖通）实现类似，此处省略
# 使用示例
if __name__ == "__main__":
# 1. 配置各平台凭证
amazon_creds = {
"access_key": "your_amazon_access_key",
"secret_key": "your_amazon_secret_key",
"associate_tag": "your_associate_tag"
}
ebay_creds = {
"client_id": "your_ebay_client_id",
"client_secret": "your_ebay_client_secret"
}
# 2. 初始化API客户端
amazon_api = AmazonItemApi(amazon_creds, region="us")
ebay_api = EbayItemApi(ebay_creds, region="us")
# 3. 获取商品详情（统一接口调用）
amazon_item = amazon_api.get_item("B07VGRJDFY") # Amazon ASIN
ebay_item = ebay_api.get_item("123456789012") # eBay Item ID
# 4. 输出标准化结果
print("Amazon商品详情：")
print(json.dumps(amazon_item, indent=2, ensure_ascii=False))
print("\nEbay商品详情：")
print(json.dumps(ebay_item, indent=2, ensure_ascii=False))