在跨境电商选品、竞品分析和口碑监测中，精准获取亚马逊商品评论数据是核心能力。本文深度解析Amazon Product Advertising API的
ItemReviews接口（即item_review），提供从AWS认证到生产级Python代码的完整实战方案，包含情感分析、关键词提取等高级应用。一、接口概述与核心特性
1.1 接口定位与价值
Amazon
item_review接口用于获取指定ASIN商品的用户评论数据，核心价值在于：- 多维度反馈：包含文本评论、星级评分（1-5星）、评论标题、图片/视频附件
- 智能筛选：支持按时间（最新/最早）、评分（1-5星）、有用性投票数排序
- 评论者信息：匿名化处理后的昵称、认证购买标识（Verified Purchase）、评论者所在地
- 商家互动：包含商家回复内容、回复时间，用于评估售后服务质量
- 分页机制：单次最多返回10条评论，支持深度分页获取全量数据
- 多站点支持：覆盖美国、英国、德国、日本、加拿大等15个区域站点
1.2 权限与调用限制
使用接口需遵守严格的开发者协议：
| 限制类型 | 具体规则 | 实战建议 |
|---|---|---|
| 权限要求 | 注册Amazon开发者账号，申请Product Advertising API权限 | 需企业认证，个人账号权限受限 |
| 调用频率 | 1次/秒，每日配额5000次（基础版） | 实现本地缓存，重复ASIN查询间隔≥24小时 |
| 数据返回 | 单次最多10条评论，支持分页 | 设置max_pages=50避免无限循环 |
| 地区限制 | 需明确指定站点（US/DE/JP等） | 不同站点需分别申请API权限 |
| 商业使用 | 非商业用途免费，商业用途需付费 | 大规模调用建议购买AWS企业套餐 |
1.3 数据模型
评论数据可表示为结构化集合：其中，，
二、准备工作
2.1 注册与认证
- 创建AWS账号：访问AWS控制台，完成企业实名认证
- 生成IAM凭证：进入IAM服务，创建用户并获取
Access Key ID和Secret Access Key - 申请PA-API权限：注册Amazon Associates，申请Product Advertising API访问权限
- 获取Associate Tag：在Associates控制台创建追踪标签（如
my-store-20）
2.2 环境配置
bash
pip install requests>=2.31.0
pip install boto3>=1.28.0 # AWS SDK（推荐）pip install pandas>=2.0.0 # 数据分析pip install matplotlib>=3.7.0 # 可视化pip install textblob>=0.17.1 # 情感分析pip install wordcloud>=1.9.0 # 词云生成pip install nltk>=3.8.0 # 自然语言处理# 下载NLTK数据（首次运行）python -c "import nltk; nltk.download('stopwords'); nltk.download('punkt')"三、核心参数与请求构造
3.1 请求参数详解
| 参数 | 必填 | 说明 | 示例 |
|---|---|---|---|
| Service | 是 | 服务名，固定AWSECommerceService | AWSECommerceService |
| Operation | 是 | 操作类型，ItemReviews | ItemReviews |
| ItemId | 是 | 商品ASIN编码 | B07VGRJDFY |
| ResponseGroup | 是 | 返回字段组，Reviews | Reviews |
| ReviewPage | 否 | 页码，从1开始 | 1 |
| Sort | 否 | 排序：Recent/Helpful | Recent |
| FilterByStar | 否 | 星级筛选：AllStars/five_star/four_star等 | five_star |
| AWSAccessKeyId | 是 | IAM访问密钥 | AKIAIOSFODNN7EXAMPLE |
| AssociateTag | 是 | 推广标签 | my-store-20 |
| Timestamp | 是 | UTC时间戳，ISO 8601格式 | 2025-01-15T12:00:00Z |
3.2 签名机制（AWS Signature V4）
Amazon采用业界最严格的HMAC-SHA256签名算法，强烈推荐使用AWS SDK自动化处理。手动签名步骤如下：
- 构造规范请求（Canonical Request）
- 创建待签字符串（String-to-Sign）
- 计算签名密钥（Signing Key）：
AWS4 + Secret Key + Date + Region + Service - 生成签名（Signature）：HMAC-SHA256 + Base64编码 + URL编码
四、生产级Python实现
4.1 完整客户端封装（支持SDK签名）
Python
import timeimport loggingfrom datetime import datetimefrom typing import Dict, List, Optionalimport requestsfrom botocore.auth import SigV4Authfrom botocore.awsrequest import AWSRequest# 配置日志logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s")logger = logging.getLogger(__name__)class AmazonReviewAPI:
"""Amazon商品评论API客户端（生产级）"""
# 地区端点映射
ENDPOINTS = {
'us': 'webservices.amazon.com',
'ca': 'webservices.amazon.ca',
'uk': 'webservices.amazon.co.uk',
'de': 'webservices.amazon.de',
'fr': 'webservices.amazon.fr',
'jp': 'webservices.amazon.co.jp',
'it': 'webservices.amazon.it',
'es': 'webservices.amazon.es',
'in': 'webservices.amazon.in'
}
def __init__(self, aws_access_key: str, aws_secret_key: str,
associate_tag: str, locale: str = 'us'):
"""
初始化客户端
:param aws_access_key: AWS Access Key ID
:param aws_secret_key: AWS Secret Access Key
:param associate_tag: 推广标签
:param locale: 地区代码（us/de/jp等）
"""
self.aws_access_key = aws_access_key
self.aws_secret_key = aws_secret_key
self.associate_tag = associate_tag
self.locale = locale
self.endpoint = self.ENDPOINTS.get(locale, 'webservices.amazon.com')
self.base_url = f"https://{self.endpoint}/onca/xml"
# 频率控制
self.last_request_time = 0
self.request_interval = 1.1 # 1.1秒间隔，确保不超过1次/秒限制
logger.info(f"✅ API客户端初始化成功，地区: {locale}")
def _check_rate_limit(self):
"""频率控制"""
current_time = time.time()
elapsed = current_time - self.last_request_time
if elapsed < self.request_interval:
sleep_time = self.request_interval - elapsed
logger.warning(f"⏳ 触发频率限制，等待 {sleep_time:.2f} 秒...")
time.sleep(sleep_time)
self.last_request_time = time.time()
def get_reviews(
self,
item_id: str,
page: int = 1,
sort: str = "Recent",
filter_by_star: str = "AllStars",
max_pages: int = 10
) -> tuple[List[Dict], Optional[Dict]]:
"""
获取商品评论（支持分页）
:param item_id: 商品ASIN
:param page: 起始页码
:param sort: 排序方式（Recent/Helpful）
:param filter_by_star: 星级筛选（AllStars/five_star等）
:param max_pages: 最大获取页数
:return: (评论列表, 商品信息)
"""
reviews = []
item_info = None
for current_page in range(page, page + max_pages):
params = {
'Service': 'AWSECommerceService',
'Operation': 'ItemReviews',
'IdType': 'ASIN',
'ItemId': item_id,
'ResponseGroup': 'Reviews',
'AWSAccessKeyId': self.aws_access_key,
'AssociateTag': self.associate_tag,
'Timestamp': datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ'),
'Version': '2013-08-01',
'ReviewPage': current_page,
'Sort': sort,
'FilterByStar': filter_by_star }
# 生成AWS签名
request = AWSRequest(method="GET", url=self.base_url, params=params)
SigV4Auth(
credentials={"access_key": self.aws_access_key, "secret_key": self.aws_secret_key},
service_name="ProductAdvertisingAPI",
region_name=self.locale ).add_auth(request)
# 频率控制
self._check_rate_limit()
try:
logger.info(f"🚀 正在获取第 {current_page} 页评论...")
response = requests.get(self.base_url, params=params, headers=dict(request.headers), timeout=20)
response.raise_for_status()
# 解析XML响应
import xml.etree.ElementTree as ET
root = ET.fromstring(response.content)
# 检查错误
error = root.find(".//Error")
if error is not None:
logger.error(f"❌ API错误: {error.findtext('Code')} - {error.findtext('Message')}")
break
# 提取商品信息（仅在第一页）
if current_page == 1:
item_info = {
"asin": item_id,
"title": root.findtext(".//Item/ItemAttributes/Title", ""),
"total_reviews": int(root.findtext(".//TotalReviews", "0")),
"average_rating": float(root.findtext(".//AverageRating", "0.0")),
"review_url": root.findtext(".//ReviewURL", "")
}
logger.info(f"📦 商品: {item_info['title'][:50]}... | 总评论数: {item_info['total_reviews']}")
# 提取评论列表
page_reviews = []
for review in root.findall(".//Review"):
review_data = {
"review_id": review.findtext("ReviewId", ""),
"rating": int(review.findtext("Rating", "0")),
"title": review.findtext("Summary", ""),
"content": review.findtext("Content", ""),
"reviewer_name": review.findtext("Reviewer/Name", ""),
"review_date": review.findtext("Date", ""),
"helpful_votes": int(review.findtext("HelpfulVotes", "0")),
"total_votes": int(review.findtext("TotalVotes", "0")),
"verified_purchase": review.findtext("VerifiedPurchase", "0") == "1",
"images": [img.text for img in review.findall(".//Image")],
"response": {
"content": review.findtext("Response/Content", ""),
"date": review.findtext("Response/Date", "")
} if review.find("Response") is not None else None
}
page_reviews.append(review_data)
reviews.extend(page_reviews)
logger.info(f"✅ 第 {current_page} 页获取成功，累计 {len(reviews)} 条评论")
# 判断是否还有更多评论
has_more = root.findtext(".//HasMore", "false") == "true"
if not has_more:
logger.info("🎯 已获取全部评论")
break
except Exception as e:
logger.error(f"❌ 请求异常: {e}")
break
return reviews, item_info
def analyze_reviews(self, reviews: List[Dict]) -> Dict:
"""
深度分析评论数据（情感分析、关键词提取、趋势分析）
"""
if not reviews:
return {}
df = pd.DataFrame(reviews)
# 1. 评分分析
rating_counts = df['rating'].value_counts().sort_index()
rating_distribution = {f"{star}星": count for star, count in rating_counts.items()}
avg_rating = df['rating'].mean()
# 2. 情感分析（基于TextBlob）
from textblob import TextBlob
sentiments = []
for content in df['content'].fillna(""):
if content:
blob = TextBlob(content)
sentiments.append(blob.sentiment.polarity) # -1到1
else:
sentiments.append(0)
df['sentiment'] = sentiments
positive_ratio = len(df[df['sentiment'] > 0.1]) / len(df)
negative_ratio = len(df[df['sentiment'] < -0.1]) / len(df)
neutral_ratio = 1 - positive_ratio - negative_ratio
# 3. 关键词提取（TF-IDF）
from sklearn.feature_extraction.text import TfidfVectorizer
# 合并标题和内容
corpus = (df['title'].fillna("") + " " + df['content'].fillna("")).tolist()
vectorizer = TfidfVectorizer(
max_features=50,
stop_words='english',
min_df=1,
ngram_range=(1, 2)
)
tfidf_matrix = vectorizer.fit_transform(corpus)
feature_names = vectorizer.get_feature_names_out()
mean_scores = tfidf_matrix.mean(axis=0).A1
top_keywords = sorted(
[(feature_names[i], mean_scores[i]) for i in range(len(feature_names))],
key=lambda x: x[1],
reverse=True
)[:15]
# 4. 时间趋势分析
df['review_date'] = pd.to_datetime(df['review_date'], errors='coerce')
monthly_trend = df.groupby(df['review_date'].dt.to_period('M')).size().to_dict()
# 5. 商家回复率
response_rate = len(df[df['response'].notna()]) / len(df)
# 6. 评价带图/视频率
media_rate = len(df[df['images'].apply(lambda x: len(x) > 0)]) / len(df)
return {
"rating_analysis": {
"distribution": rating_distribution,
"average": round(avg_rating, 2)
},
"sentiment_analysis": {
"positive_ratio": round(positive_ratio, 3),
"negative_ratio": round(negative_ratio, 3),
"neutral_ratio": round(neutral_ratio, 3)
},
"keywords": {
"overall": top_keywords,
"positive": self._extract_keywords_by_sentiment(df, 1),
"negative": self._extract_keywords_by_sentiment(df, -1)
},
"trend": monthly_trend,
"response_rate": round(response_rate, 3),
"media_rate": round(media_rate, 3)
}
def _extract_keywords_by_sentiment(self, df: pd.DataFrame, sentiment_type: int) -> List[Tuple[str, float]]:
"""按情感提取关键词"""
if sentiment_type == 1:
subset = df[df['sentiment'] > 0.2]
else:
subset = df[df['sentiment'] < -0.2]
if subset.empty:
return []
corpus = (subset['title'].fillna("") + " " + subset['content'].fillna("")).tolist()
from sklearn.feature_extraction.text import TfidfVectorizer
vectorizer = TfidfVectorizer(max_features=20, stop_words='english', min_df=1)
tfidf_matrix = vectorizer.fit_transform(corpus)
feature_names = vectorizer.get_feature_names_out()
mean_scores = tfidf_matrix.mean(axis=0).A1
return sorted(
[(feature_names[i], mean_scores[i]) for i in range(len(feature_names))],
key=lambda x: x[1],
reverse=True
)[:10]
def visualize_analysis(self, analysis: Dict, item_info: Optional[Dict] = None):
"""
可视化分析结果（评分分布、情感分析、关键词词云）
"""
import matplotlib.pyplot as plt from wordcloud import WordCloud
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 1. 评分分布柱状图
if "rating_analysis" in analysis:
ratings = list(analysis["rating_analysis"]["distribution"].keys())
counts = list(analysis["rating_analysis"]["distribution"].values())
axes[0, 0].bar(ratings, counts, color=['#ff6b6b', '#ff9f43', '#feca57', '#48dbfb', '#0abde3'])
axes[0, 0].set_title("Rating Distribution", fontsize=14, fontweight='bold')
axes[0, 0].set_xlabel("Rating")
axes[0, 0].set_ylabel("Count")
# 2. 情感分析饼图
if "sentiment_analysis" in analysis:
sentiment_data = analysis["sentiment_analysis"]
labels = ['Positive', 'Negative', 'Neutral']
sizes = [sentiment_data["positive_ratio"],
sentiment_data["negative_ratio"],
sentiment_data["neutral_ratio"]]
axes[0, 1].pie(sizes, labels=labels, autopct='%1.1f%%',
colors=['#2ecc71', '#e74c3c', '#95a5a6'])
axes[0, 1].set_title("Sentiment Analysis", fontsize=14, fontweight='bold')
# 3. 关键词词云
if "keywords" in analysis:
keywords_dict = {kw: score for kw, score in analysis["keywords"]["overall"][:30]}
wordcloud = WordCloud(width=800, height=400, background_color='white',
colormap='viridis').generate_from_frequencies(keywords_dict)
axes[1, 0].imshow(wordcloud, interpolation='bilinear')
axes[1, 0].axis('off')
axes[1, 0].set_title("Keywords WordCloud", fontsize=14, fontweight='bold')
# 4. 时间趋势折线图
if "trend" in analysis and analysis["trend"]:
dates = list(analysis["trend"].keys())
counts = list(analysis["trend"].values())
axes[1, 1].plot(dates, counts, marker='o', linewidth=2, color='#3498db')
axes[1, 1].set_title("Review Trend Over Time", fontsize=14, fontweight='bold')
axes[1, 1].set_xlabel("Month")
axes[1, 1].set_ylabel("Review Count")
plt.xticks(rotation=45)
plt.suptitle(f"Amazon Product Review Analysis - {item_info['title'][:50] if item_info else 'Unknown'}",
fontsize=16, fontweight='bold')
plt.tight_layout()
plt.show()
def get_negative_feedback(self, reviews: List[Dict], threshold: int = 3) -> List[Dict]:
"""
提取负面反馈（评分≤threshold）并聚类主要问题
"""
negative_reviews = [r for r in reviews if r["rating"] <= threshold]
if not negative_reviews:
return []
# 提取负面关键词
from collections import Counter
all_words = []
for review in negative_reviews:
text = f"{review['title']} {review['content']}".lower()
words = re.findall(r'\b[a-zA-Z]{3,}\b', text)
words = [w for w in words if w not in self.stop_words]
all_words.extend(words)
word_counts = Counter(all_words)
top_issues = word_counts.most_common(10)
# 为每个问题提供示例评论
feedback_list = []
for keyword, count in top_issues:
examples = [r for r in negative_reviews if keyword in f"{r['title']} {r['content']}".lower()][:2]
feedback_list.append({
"keyword": keyword,
"count": count,
"examples": examples })
return feedback_list
def export_to_excel(self, reviews: List[Dict], analysis: Dict,
item_info: Dict, filename: str):
"""
导出数据到Excel（包含评论明细和分析摘要）
"""
import pandas as pd
# 评论明细Sheet
df_reviews = pd.DataFrame(reviews)
if not df_reviews.empty:
df_reviews['review_date'] = pd.to_datetime(df_reviews['review_date'], errors='coerce')
df_reviews['has_response'] = df_reviews['response'].notna()
# 分析摘要Sheet
summary_data = {
"Metric": ["Total Reviews", "Average Rating", "Positive Ratio", "Negative Ratio",
"Response Rate", "Media Rate"],
"Value": [item_info.get("total_reviews", 0), analysis.get("rating_analysis", {}).get("average", 0),
analysis.get("sentiment_analysis", {}).get("positive_ratio", 0),
analysis.get("sentiment_analysis", {}).get("negative_ratio", 0),
analysis.get("response_rate", 0), analysis.get("media_rate", 0)]
}
df_summary = pd.DataFrame(summary_data)
# 关键词Sheet
if "keywords" in analysis:
df_keywords = pd.DataFrame(analysis["keywords"]["overall"],
columns=["Keyword", "Score"]).head(20)
else:
df_keywords = pd.DataFrame()
# 导出到Excel
with pd.ExcelWriter(filename, engine='openpyxl') as writer:
df_reviews.to_excel(writer, sheet_name='Reviews Detail', index=False)
df_summary.to_excel(writer, sheet_name='Analysis Summary', index=False)
df_keywords.to_excel(writer, sheet_name='Top Keywords', index=False)
logger.info(f"💾 数据已导出至: {filename}")# 使用示例if __name__ == "__main__":
# 配置凭证（需替换为真实值）
AWS_ACCESS_KEY = "your_access_key"
AWS_SECRET_KEY = "your_secret_key"
ASSOCIATE_TAG = "your_associate_tag"
LOCALE = "us" # 美国站点
ITEM_ID = "B07VGRJDFY" # 示例ASIN
# 初始化API
amazon = AmazonReviewAPI(AWS_ACCESS_KEY, AWS_SECRET_KEY, ASSOCIATE_TAG, LOCALE)
# 获取评论（最多5页）
reviews, item_info = amazon.get_reviews(
item_id=ITEM_ID,
page=1,
sort="Recent",
max_pages=5
)
if reviews:
# 深度分析
analysis = amazon.analyze_reviews(reviews)
# 可视化
amazon.visualize_analysis(analysis, item_info)
# 导出Excel
amazon.export_to_excel(reviews, analysis, item_info, "amazon_reviews_analysis.xlsx")
# 负面反馈分析
negative_issues = amazon.get_negative_feedback(reviews, threshold=3)
print(f"\n🔍 TOP 5负面问题:")
for issue in negative_issues[:5]:
print(f" - {issue['keyword']} ({issue['count']}次提及)")五、高级应用场景
5.1 竞品评论对比分析
Python
def compare_products(asins: List[str]):
"""多商品评论对比"""
comparison = {}
for asin in asins:
reviews, info = amazon.get_reviews(asin, max_pages=3)
if reviews:
analysis = amazon.analyze_reviews(reviews)
comparison[asin] = {
"title": info['title'],
"avg_rating": analysis["rating_analysis"]["average"],
"positive_ratio": analysis["sentiment_analysis"]["positive_ratio"],
"top_keywords": [kw[0] for kw in analysis["keywords"]["overall"][:5]]
}
# 生成对比报告
df = pd.DataFrame(comparison).T print(df.to_string())5.2 评论趋势预警监控
Python
import scheduledef monitor_reviews(asin: str, interval_hours: int = 24):
"""定时监控评论变化"""
def job():
reviews, info = amazon.get_reviews(asin, page=1, max_pages=1)
if reviews:
new_count = len(reviews)
logger.info(f"📊 {info['title'][:30]}... 新增评论: {new_count}")
# 可接入钉钉/Slack告警
# send_alert(f"商品 {asin} 新增 {new_count} 条评论")
schedule.every(interval_hours).hours.do(job)5.3 Buy with Prime插件集成（无代码方案）
对于自建站卖家，Amazon官方提供Buy with Prime插件，可一键同步评论数据：
- 零代码部署：嵌入JS脚本即可显示评论模块
- 实时同步：评论随Amazon更新自动刷新
- 转化率提升：研究表明平均提升38%
六、关键注意事项
6.1 合规红线（⚠️ 重中之重）
- 禁止数据缓存：Amazon规定评论数据缓存不超过24小时，违者账号暂停
- Associate Tag必须使用：每个请求必须包含有效推广标签
- API用途限制：仅限推广和选品，禁止用于价格监控爬虫
- 用户隐私保护：不得存储或泄露评论者个人信息
- GDPR合规：处理欧盟用户数据需遵循数据最小化原则
6.2 技术优化
- 配额监控：每小时调用量接近4000次时触发告警
- 异步调用：生产环境建议使用
aiohttp提升并发效率 - 智能重试：对
RequestThrottled错误实现指数退避（2s/4s/8s） - 数据去重：同一ASIN的评论ID可能存在重复，需去重处理
6.3 常见错误码
| 错误码 | 含义 | 解决方案 |
|---|---|---|
SignatureDoesNotMatch | 签名错误 | 检查Secret Key、参数排序、时间戳格式 |
TooManyRequests | 频率超限 | 降低QPS，或申请更高配额 |
ItemNotFound | ASIN不存在 | 验证ASIN有效性，或商品已下架 |
InvalidParameterValue | 参数无效 | 检查Sort/FilterByStar参数拼写 |
6.4 第三方替代方案
如果官方API权限申请困难，可使用Pangolin Scrape API：
Python
# 无需AWS签名，但成本较高（约$0.01/次）response = requests.get(
"https://api.pangolinfo.com/v1/amazon/reviews",
headers={"Authorization": f"Bearer {API_KEY}"},
params={"asin": "B07VGRJDFY", "page": 1})七、总结
Amazon
item_review接口是获取全球用户反馈的金钥匙，但严格的AWS认证和配额管理对开发者提出了较高要求。本文提供的SDK封装方案可自动处理签名，大幅降低开发难度。核心要点回顾：
- 企业认证：必须申请企业级PA-API权限，个人开发者受限
- 签名自动化：使用
botocore代替手动HMAC-SHA256，避免错误 - 深度分析：集成TextBlob情感分析和TF-IDF关键词提取，挖掘数据价值
- 合规第一：严格遵守Amazon服务条款，避免账号封禁风险
- 频率控制：1次/秒限制下，建议缓存评论数据24小时
生产建议：部署在AWS EC2上，使用IAM Role管理密钥，避免硬编码。对于大规模数据需求，可考虑SP-API（Selling Partner API），支持更高配额和实时数据。
如遇任何疑问或有进一步的需求，请随时与我私信或者评论联系。
