阿里巴巴开放平台的 item_review 接口用于获取商品的用户评论数据，是分析商品口碑、用户反馈和市场需求的重要工具。通过该接口可以获取评论内容、评分、买家信息、追加评论等关键数据，为产品优化、市场策略制定提供数据支持。
一、item_review 接口核心特性分析
- 接口功能与定位
item_review 接口专注于返回特定商品的用户评论数据，其核心价值在于：
提供多维度的评论筛选（按评分、时间、是否有图等）
支持分页获取大量评论数据
包含详细的评论内容和用户反馈信息
提供评论相关的统计数据（如各星级占比） - 接口权限与调用限制
使用 item_review 接口需遵守阿里巴巴开放平台的使用规范：
限制类型 具体规则
权限要求 需要申请 "商品评论" 相关权限
调用频率 通常限制为 60 次 / 分钟
数据返回 单次最多返回 40 条评论，最多支持 100 页
字段权限 部分敏感字段（如买家详细信息）需特殊申请 - 核心参数解析
必选参数
参数名 类型 说明
app_key String 应用 ID
method String 接口名称，固定为 "alibaba.item.review.get"
timestamp String 时间戳，格式为 yyyy-MM-dd HH:mm:ss
format String 返回格式，通常为 "json"
v String 版本号，固定为 "2.0"
sign String 签名
offerId String 商品 ID
page Integer 页码，从 1 开始
pageSize Integer 每页记录数，1-40
可选参数
参数名 类型 说明
grade Integer 评分筛选，1-5 分，0 表示全部
hasImage Boolean 是否只看有图评论，true/false
startDate String 评论开始时间，格式 yyyy-MM-dd
endDate String 评论结束时间，格式 yyyy-MM-dd
sort String 排序方式，"newest" 最新，"helpful" 最有帮助
二、返回数据结构分析
item_review 接口返回的数据结构包含以下核心部分：
基础信息：状态码、请求 ID 等
统计信息：总评论数、各星级评论数量等
评论列表：每条评论的详细信息
评论列表中的关键字段：
reviewId：评论 ID
buyerId：买家 ID（加密处理）
buyerNick：买家昵称
grade：评分（1-5 星）
content：评论内容
gmtCreate：评论时间
reply：商家回复
images：评论图片 URL 列表
attribute：购买属性（如颜色、尺寸）
appendContent：追加评论内容
useful：有用数
三、Python 实现方案
以下是 item_review 接口的完整 Python 实现，包含基础调用、数据获取、情感分析和可视化功能：
import requests
import time
import hashlib
import json
import logging
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import re
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Dict, List, Optional, Tuple
import jieba
from jieba.analyse import extract_tags
from snownlp import SnowNLP # 用于情感分析
配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
配置中文显示
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
plt.rcParams["axes.unicode_minus"] = False
class AlibabaItemReview:
"""阿里巴巴item_review接口封装类，用于获取和分析商品评论"""
def __init__(self, app_key: str, app_secret: str, access_token: str = ""):
"""
初始化客户端
:param app_key: 应用的app_key
:param app_secret: 应用的app_secret
:param access_token: 访问令牌
"""
self.app_key = app_key
self.app_secret = app_secret
self.access_token = access_token
self.gateway_url = "https://gw.open.1688.com/openapi/param2/2.0/"
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
"User-Agent": "AlibabaItemReview/1.0.0 (Python)"
})
# 频率控制
self.api_calls = 0
self.rate_limit = 60 # 每分钟最多调用次数
self.last_reset_time = time.time()
# 加载停用词
self.stopwords = self._load_stopwords()
def _load_stopwords(self) -> set:
"""加载中文停用词"""
try:
# 简单的停用词列表，实际应用中可替换为更完整的停用词表
stopwords = {
'的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一',
'一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有',
'看', '好', '自己', '这', '个', '也', '还', '可以', '买', '用', '后', '觉得'
}
return stopwords
except Exception as e:
logging.warning(f"加载停用词失败: {e}，使用默认空集合")
return set()
def _generate_sign(self, params: Dict) -> str:
"""生成签名"""
# 排序参数
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 拼接签名字符串
sign_str = self.app_secret
for k, v in sorted_params:
sign_str += f"{k}{v}"
sign_str += self.app_secret
# 计算MD5
return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
def _check_rate_limit(self) -> bool:
"""检查API调用频率限制"""
current_time = time.time()
# 每分钟重置计数
if current_time - self.last_reset_time > 60:
self.api_calls = 0
self.last_reset_time = current_time
# 检查是否超过限制
if self.api_calls >= self.rate_limit:
sleep_time = 60 - (current_time - self.last_reset_time)
logging.warning(f"API调用频率超限，等待 {sleep_time:.1f} 秒")
time.sleep(sleep_time + 1)
self.api_calls = 0
self.last_reset_time = time.time()
self.api_calls += 1
return True
def call_review_api(self, offer_id: str, page: int = 1, page_size: int = 40,
filters: Optional[Dict] = None) -> Optional[Dict]:
"""
调用item_review接口获取商品评论
:param offer_id: 商品ID
:param page: 页码
:param page_size: 每页数量
:param filters: 筛选条件
:return: 评论数据
"""
if not self.access_token:
logging.error("请先设置access_token")
return None
# 检查频率限制
if not self._check_rate_limit():
return None
# 构建基础参数
base_params = {
"app_key": self.app_key,
"access_token": self.access_token,
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"format": "json",
"v": "2.0",
"sign_method": "md5",
"offerId": offer_id,
"page": page,
"pageSize": page_size
}
# 合并筛选参数
if filters:
base_params.update(filters)
# 生成签名
sign = self._generate_sign(base_params)
base_params["sign"] = sign
try:
# 构建完整URL
url = f"{self.gateway_url}alibaba.item.review.get"
response = self.session.post(url, data=base_params, timeout=15)
response.raise_for_status()
# 解析响应
result = response.json()
# 处理错误
if "error_response" in result:
error = result["error_response"]
logging.error(f"API调用错误: {error.get('msg')} (错误码: {error.get('code')})")
return None
# 返回结果
return result.get("alibaba_item_review_get_response", {})
except requests.exceptions.RequestException as e:
logging.error(f"API请求异常: {str(e)}")
return None
except json.JSONDecodeError:
logging.error(f"API响应解析失败: {response.text[:200]}...")
return None
def get_all_reviews(self, offer_id: str, max_pages: int = 10, page_size: int = 40,
filters: Optional[Dict] = None) -> Tuple[List[Dict], Dict]:
"""
获取多页评论数据
:param offer_id: 商品ID
:param max_pages: 最大页数
:param page_size: 每页数量
:param filters: 筛选条件
:return: 评论列表和统计信息
"""
all_reviews = []
stats = {}
page = 1
while page <= max_pages:
logging.info(f"获取第 {page} 页评论")
result = self.call_review_api(offer_id, page, page_size, filters)
if not result or "result" not in result:
break
review_result = result["result"]
# 保存统计信息（第一页获取一次即可）
if page == 1:
stats = {
"total_count": review_result.get("totalCount", 0),
"total_page": review_result.get("totalPage", 0),
"grade_distribution": review_result.get("gradeDistribution", {})
}
# 获取当前页评论
reviews = review_result.get("reviews", {}).get("review", [])
if not reviews:
break
# 格式化评论数据并添加到列表
formatted_reviews = [self._format_review(review) for review in reviews]
all_reviews.extend(formatted_reviews)
# 检查是否已达最后一页
if page >= review_result.get("totalPage", max_pages):
break
page += 1
time.sleep(1) # 避免请求过于频繁
logging.info(f"共获取 {len(all_reviews)} 条评论")
return all_reviews, stats
def _format_review(self, review: Dict) -> Dict:
"""格式化评论数据"""
return {
"review_id": review.get("reviewId"),
"buyer_id": review.get("buyerId"),
"buyer_nick": review.get("buyerNick"),
"grade": int(review.get("grade", 0)),
"content": review.get("content", ""),
"create_time": review.get("gmtCreate"),
"reply": review.get("replyContent", ""),
"reply_time": review.get("replyGmtCreate"),
"images": review.get("images", {}).get("image", []),
"attribute": review.get("attribute", ""),
"append_content": review.get("appendContent", ""),
"append_time": review.get("appendGmtCreate"),
"useful": int(review.get("useful", 0)),
"is_vip": review.get("isVip", False)
}
def analyze_reviews(self, reviews: List[Dict]) -> Dict:
"""分析评论数据"""
if not reviews:
return {"error": "没有评论数据可分析"}
# 1. 评分分析
grade_counts = defaultdict(int)
for review in reviews:
grade_counts[review["grade"]] += 1
# 2. 时间分布
time_distribution = defaultdict(int)
for review in reviews:
if review["create_time"]:
date_str = review["create_time"].split()[0]
time_distribution[date_str] += 1
# 3. 情感分析
sentiment_results = self._analyze_sentiment(reviews)
# 4. 关键词提取
keywords = self._extract_keywords(reviews)
# 5. 评论长度分析
content_lengths = [len(review["content"]) for review in reviews if review["content"]]
length_stats = {}
if content_lengths:
length_stats = {
"avg": round(sum(content_lengths) / len(content_lengths), 1),
"min": min(content_lengths),
"max": max(content_lengths),
"median": np.median(content_lengths)
}
# 6. 有图评论比例
image_count = sum(1 for review in reviews if review["images"])
image_ratio = image_count / len(reviews) if reviews else 0
# 7. 追加评论比例
append_count = sum(1 for review in reviews if review["append_content"])
append_ratio = append_count / len(reviews) if reviews else 0
return {
"total_reviews": len(reviews),
"grade_analysis": {
"distribution": dict(grade_counts),
"avg_grade": sum(k * v for k, v in grade_counts.items()) / len(reviews) if reviews else 0
},
"time_distribution": dict(time_distribution),
"sentiment_analysis": sentiment_results,
"keywords": keywords,
"content_length": length_stats,
"image_ratio": image_ratio,
"append_ratio": append_ratio
}
def _analyze_sentiment(self, reviews: List[Dict]) -> Dict:
"""对评论进行情感分析"""
sentiment_scores = []
positive_count = 0
negative_count = 0
neutral_count = 0
for review in reviews:
content = review["content"]
if not content:
continue
# 使用SnowNLP进行情感分析，返回值0-1，越接近1越积极
s = SnowNLP(content)
score = s.sentiments
sentiment_scores.append(score)
# 分类情感
if score > 0.6:
positive_count += 1
elif score < 0.4:
negative_count += 1
else:
neutral_count += 1
# 按评分分析情感
grade_sentiment = defaultdict(list)
for review in reviews:
content = review["content"]
if content:
s = SnowNLP(content)
grade_sentiment[review["grade"]].append(s.sentiments)
grade_sentiment_avg = {
grade: sum(scores)/len(scores) if scores else 0
for grade, scores in grade_sentiment.items()
}
return {
"total_analyzed": len(sentiment_scores),
"positive_ratio": positive_count / len(sentiment_scores) if sentiment_scores else 0,
"negative_ratio": negative_count / len(sentiment_scores) if sentiment_scores else 0,
"neutral_ratio": neutral_count / len(sentiment_scores) if sentiment_scores else 0,
"avg_score": sum(sentiment_scores)/len(sentiment_scores) if sentiment_scores else 0,
"grade_sentiment_avg": grade_sentiment_avg
}
def _extract_keywords(self, reviews: List[Dict], top_n: int = 20) -> Dict:
"""提取评论关键词"""
# 合并所有评论内容
all_content = " ".join([review["content"] for review in reviews if review["content"]])
# 清洗文本
all_content = re.sub(r"[^\w\s]", " ", all_content) # 去除特殊字符
all_content = re.sub(r"\s+", " ", all_content).strip() # 合并空格
# 提取关键词
keywords = extract_tags(
all_content,
topK=top_n,
withWeight=True,
allowPOS=('n', 'v', 'a') # 只提取名词、动词、形容词
)
# 按评分提取关键词
grade_keywords = {}
for grade in [1, 2, 3, 4, 5]:
grade_content = " ".join([
review["content"] for review in reviews
if review["content"] and review["grade"] == grade
])
if grade_content:
grade_content = re.sub(r"[^\w\s]", " ", grade_content)
grade_content = re.sub(r"\s+", " ", grade_content).strip()
grade_keywords[grade] = extract_tags(
grade_content,
topK=10,
withWeight=True,
allowPOS=('n', 'v', 'a')
)
return {
"overall": keywords,
"by_grade": grade_keywords
}
def get_negative_reasons(self, reviews: List[Dict], top_n: int = 10) -> List[Dict]:
"""获取负面评论的主要原因"""
# 筛选负面评论（评分<=3星）
negative_reviews = [
review for review in reviews
if review["grade"] <= 3 and review["content"]
]
if not negative_reviews:
return []
# 提取负面评论关键词
negative_content = " ".join([review["content"] for review in negative_reviews])
negative_content = re.sub(r"[^\w\s]", " ", negative_content)
negative_content = re.sub(r"\s+", " ", negative_content).strip()
keywords = extract_tags(
negative_content,
topK=top_n,
withWeight=True,
allowPOS=('n', 'v', 'a')
)
# 为每个关键词找到示例评论
result = []
for keyword, weight in keywords:
# 找到包含该关键词的评论示例
examples = []
for review in negative_reviews[:50]: # 只在前50条中查找
if keyword in review["content"] and len(examples) < 3:
examples.append({
"review_id": review["review_id"],
"content": review["content"],
"grade": review["grade"]
})
result.append({
"keyword": keyword,
"weight": weight,
"count": sum(1 for r in negative_reviews if keyword in r["content"]),
"examples": examples
})
return result
def visualize_analysis(self, analysis: Dict, output_dir: str = ".") -> None:
"""可视化分析结果"""
# 1. 评分分布饼图
if "grade_analysis" in analysis and analysis["grade_analysis"]["distribution"]:
grades = analysis["grade_analysis"]["distribution"]
labels = [f"{k}星" for k in grades.keys()]
sizes = list(grades.values())
plt.figure(figsize=(8, 6))
plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=90)
plt.title(f'评论评分分布 (平均评分: {analysis["grade_analysis"]["avg_grade"]:.1f})')
plt.axis('equal')
plt.tight_layout()
plt.savefig(f"{output_dir}/grade_distribution.png")
plt.close()
logging.info(f"评分分布图表已保存至 {output_dir}/grade_distribution.png")
# 2. 情感分析柱状图
if "sentiment_analysis" in analysis:
sentiment = analysis["sentiment_analysis"]
labels = ['正面', '负面', '中性']
sizes = [
sentiment["positive_ratio"],
sentiment["negative_ratio"],
sentiment["neutral_ratio"]
]
plt.figure(figsize=(8, 6))
plt.bar(labels, sizes, color=['green', 'red', 'gray'])
plt.title(f'评论情感分布 (平均情感值: {sentiment["avg_score"]:.2f})')
plt.ylim(0, 1)
plt.ylabel('比例')
for i, v in enumerate(sizes):
plt.text(i, v + 0.01, f'{v:.1%}', ha='center')
plt.tight_layout()
plt.savefig(f"{output_dir}/sentiment_distribution.png")
plt.close()
logging.info(f"情感分布图表已保存至 {output_dir}/sentiment_distribution.png")
# 3. 关键词云图（需要wordcloud库支持）
if "keywords" in analysis and analysis["keywords"]["overall"]:
try:
from wordcloud import WordCloud
# 准备关键词文本
keyword_text = " ".join([kw for kw, _ in analysis["keywords"]["overall"]])
# 生成词云
wc = WordCloud(
font_path="simhei.ttf", # 替换为你的中文字体路径
background_color="white",
width=800,
height=600
).generate(keyword_text)
plt.figure(figsize=(10, 7))
plt.imshow(wc, interpolation="bilinear")
plt.axis("off")
plt.title("评论关键词云图")
plt.tight_layout(pad=0)
plt.savefig(f"{output_dir}/keyword_cloud.png")
plt.close()
logging.info(f"关键词云图已保存至 {output_dir}/keyword_cloud.png")
except ImportError:
logging.warning("wordcloud库未安装，无法生成关键词云图")
except Exception as e:
logging.warning(f"生成关键词云图失败: {e}")
# 4. 时间趋势图
if "time_distribution" in analysis and analysis["time_distribution"]:
time_data = sorted(analysis["time_distribution"].items(), key=lambda x: x[0])
dates = [item[0] for item in time_data]
counts = [item[1] for item in time_data]
# 只显示最近30天的数据
if len(dates) > 30:
dates = dates[-30:]
counts = counts[-30:]
plt.figure(figsize=(12, 6))
plt.plot(dates, counts, marker='o', linestyle='-')
plt.title('评论数量时间趋势')
plt.xlabel('日期')
plt.ylabel('评论数量')
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig(f"{output_dir}/review_time_trend.png")
plt.close()
logging.info(f"时间趋势图表已保存至 {output_dir}/review_time_trend.png")
def export_to_excel(self, reviews: List[Dict], analysis: Dict, filename: str) -> None:
"""导出评论数据和分析结果到Excel"""
if not reviews and not analysis:
logging.warning("没有数据可导出")
return
try:
with pd.ExcelWriter(filename) as writer:
# 评论数据
if reviews:
df_reviews = pd.DataFrame(reviews)
df_reviews.to_excel(writer, sheet_name='评论数据', index=False)
# 分析结果
if analysis:
# 评分分析
if "grade_analysis" in analysis:
df_grade = pd.DataFrame([
{"评分": k, "数量": v, "比例": v/analysis["total_reviews"]}
for k, v in analysis["grade_analysis"]["distribution"].items()
])
df_grade.to_excel(writer, sheet_name='评分分析', index=False)
# 情感分析
if "sentiment_analysis" in analysis:
df_sentiment = pd.DataFrame([{
"类型": "正面",
"比例": analysis["sentiment_analysis"]["positive_ratio"]
}, {
"类型": "负面",
"比例": analysis["sentiment_analysis"]["negative_ratio"]
}, {
"类型": "中性",
"比例": analysis["sentiment_analysis"]["neutral_ratio"]
}])
df_sentiment.to_excel(writer, sheet_name='情感分析', index=False)
# 关键词
if "keywords" in analysis and analysis["keywords"]["overall"]:
df_keywords = pd.DataFrame([
{"关键词": kw, "权重": weight}
for kw, weight in analysis["keywords"]["overall"]
])
df_keywords.to_excel(writer, sheet_name='关键词分析', index=False)
logging.info(f"数据已导出至 {filename}")
except Exception as e:
logging.error(f"导出Excel失败: {e}")
示例调用
if name == "main":
# 替换为实际的参数（从1688开放平台获取）
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
ACCESS_TOKEN = "your_access_token"
OFFER_ID = "61234567890" # 示例商品ID
# 初始化API客户端
review_analyzer = AlibabaItemReview(APP_KEY, APP_SECRET, ACCESS_TOKEN)
# 1. 获取评论数据
print("=== 获取评论数据 ===")
# 设置筛选条件：获取最近30天的评论
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
filters = {
"startDate": start_date.strftime("%Y-%m-%d"),
"endDate": end_date.strftime("%Y-%m-%d"),
# "grade": 3, # 可以指定只获取特定评分的评论
# "hasImage": True, # 只获取有图评论
"sort": "newest" # 按最新排序
}
# 获取最多5页评论
reviews, stats = review_analyzer.get_all_reviews(
offer_id=OFFER_ID,
max_pages=5,
page_size=40,
filters=filters
)
print(f"商品ID: {OFFER_ID}")
print(f"获取评论总数: {len(reviews)}")
print(f"平台总评论数: {stats.get('total_count', 0)}")
print(f"评分分布: {stats.get('grade_distribution', {})}")
# 2. 分析评论数据
print("\n=== 评论数据分析 ===")
if reviews:
analysis = review_analyzer.analyze_reviews(reviews)
print(f"平均评分: {analysis['grade_analysis']['avg_grade']:.1f}")
print(f"情感分析: 正面{analysis['sentiment_analysis']['positive_ratio']:.1%}, "
f"负面{analysis['sentiment_analysis']['negative_ratio']:.1%}, "
f"中性{analysis['sentiment_analysis']['neutral_ratio']:.1%}")
print(f"有图评论比例: {analysis['image_ratio']:.1%}")
print(f"追加评论比例: {analysis['append_ratio']:.1%}")
print("\n热门关键词:")
for kw, weight in analysis["keywords"]["overall"][:10]:
print(f" {kw}: {weight:.4f}")
# 3. 获取负面评论原因
print("\n=== 负面评论主要原因 ===")
negative_reasons = review_analyzer.get_negative_reasons(reviews)
for i, reason in enumerate(negative_reasons[:5], 1):
print(f"{i}. 关键词: {reason['keyword']} (出现{reason['count']}次)")
if reason["examples"]:
print(f" 示例: {reason['examples'][0]['content'][:50]}...")
# 4. 可视化分析结果
review_analyzer.visualize_analysis(analysis)
# 5. 导出数据到Excel
review_analyzer.export_to_excel(reviews, analysis, "商品评论分析.xlsx")
else:
print("未获取到评论数据，无法进行分析")
四、接口调用注意事项
- 错误处理与调试
常见错误及解决方案：
错误码 说明 解决方案
10001 缺少必要参数 检查必填参数是否完整
10002 参数格式错误 检查参数格式，特别是日期格式
110 access_token 无效 重新获取有效的 access_token
40 权限不足 申请 item_review 接口的访问权限
429 调用频率超限 优化代码，增加调用间隔
20000 商品不存在 检查 offerId 是否正确 - 性能优化建议
批量获取：合理设置 page_size（建议 20-40），减少请求次数
增量更新：通过 startDate 参数只获取新增评论，减少数据传输量
数据缓存：对评论数据进行缓存，避免重复调用
并发控制：多线程调用时控制并发数，建议不超过 5 个线程
按需筛选：使用 filters 参数过滤不需要的数据，减少处理量
五、应用场景与扩展建议
典型应用场景
商品口碑监测：实时监控商品评论，及时发现负面反馈
竞品分析：对比分析同类商品的用户评价，找出优势与不足
产品优化：基于用户评论中的关键词和反馈，改进产品设计
客服辅助：自动识别需要关注的负面评论，优先处理
市场调研：分析特定品类的用户需求和痛点
扩展建议
实现实时监控：通过定时任务定期获取评论，监控评论趋势变化
构建情感词典：针对特定行业构建专业情感词典，提高情感分析准确性
评论图片分析：结合图像识别技术，分析评论中的图片内容
多维度对比：对比不同时间段、不同地区的评论差异
自动摘要生成：基于评论内容自动生成用户反馈摘要报告
通过 item_review 接口获取的评论数据蕴含着丰富的用户反馈信息，合理利用这些数据可以帮助企业更好地了解市场需求，优化产品和服务，提升竞争力。在实际应用中，需注意数据的合规使用和隐私保护，同时不断优化分析算法，提高分析结果的准确性和实用性。
