京东的 item_review 接口是用于获取商品用户评论数据的专业接口，能够获取京东平台上指定商品的用户评价、评分、晒单、追评等详细信息。这些评论数据对于商品口碑分析、用户需求挖掘、服务质量评估等场景具有重要价值，是电商运营和产品改进的重要数据来源。
一、接口核心特性分析
1. 接口功能与定位
2. 认证机制
3. 核心参数与响应结构
请求参数
响应核心字段
二、Python 脚本实现
import requests
import time
import json
import logging
import re
from typing import Dict, Optional, List
from requests.exceptions import RequestException
from snownlp import SnowNLP # 用于情感分析，需安装：pip install snownlp
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class JDItemReviewAPI:
def __init__(self, appkey: str, appsecret: str):
"""
初始化京东评论API客户端
:param appkey: 京东开放平台appkey
:param appsecret: 京东开放平台appsecret
"""
self.appkey = appkey
self.appsecret = appsecret
self.base_url = "https://api.jd.com"
self.access_token = None
self.token_expires_at = 0 # token过期时间戳
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36"
})
def _get_access_token(self) -> Optional[str]:
"""获取访问令牌"""
# 检查token是否有效
if self.access_token and self.token_expires_at > time.time() + 60:
return self.access_token
logging.info("获取新的access_token")
url = f"{self.base_url}/oauth2/token"
params = {
"grant_type": "client_credentials",
"appkey": self.appkey,
"appsecret": self.appsecret
}
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
result = response.json()
if "access_token" in result:
self.access_token = result["access_token"]
self.token_expires_at = time.time() + result.get("expires_in", 86400) # 默认为24小时
return self.access_token
else:
logging.error(f"获取access_token失败: {result.get('error_description', '未知错误')}")
return None
except RequestException as e:
logging.error(f"获取access_token请求异常: {str(e)}")
return None
def get_item_reviews(self,
sku_id: str,
page: int = 1,
page_size: int = 10,
sort_type: int = 0,
score: int = 0,
has_image: int = 0) -> Optional[Dict]:
"""
获取商品评论
:param sku_id: 商品SKU ID
:param page: 页码
:param page_size: 每页条数
:param sort_type: 排序方式：0-推荐，1-时间，2-评分高，3-评分低
:param score: 评分筛选：0-全部，1-好评，2-中评，3-差评
:param has_image: 是否有图：0-全部，1-有图
:return: 评论数据
"""
# 验证参数
valid_sort_types = [0, 1, 2, 3]
if sort_type not in valid_sort_types:
logging.error(f"无效的排序方式: {sort_type}，支持: {valid_sort_types}")
return None
valid_scores = [0, 1, 2, 3]
if score not in valid_scores:
logging.error(f"无效的评分筛选: {score}，支持: {valid_scores}")
return None
if page_size < 1 or page_size > 50:
logging.error(f"每页条数必须在1-50之间，当前为: {page_size}")
return None
# 获取有效的access_token
if not self._get_access_token():
return None
url = f"{self.base_url}/item/review"
# 构建请求参数
params = {
"sku_id": sku_id,
"access_token": self.access_token,
"page": page,
"page_size": page_size,
"sort_type": sort_type,
"score": score,
"has_image": has_image,
"timestamp": int(time.time() * 1000)
}
try:
response = self.session.get(url, params=params, timeout=15)
response.raise_for_status()
result = response.json()
# 检查响应状态
if result.get("code") == 200:
# 格式化评论数据
return self._format_review_data(result.get("data", {}))
else:
logging.error(f"获取商品评论失败: {result.get('message', '未知错误')} (错误码: {result.get('code')})")
return None
except RequestException as e:
logging.error(f"获取商品评论请求异常: {str(e)}")
return None
except json.JSONDecodeError:
logging.error(f"评论响应解析失败: {response.text[:200]}...")
return None
def _format_review_data(self, review_data: Dict) -> Dict:
"""格式化评论数据"""
# 分页信息
pagination = {
"total_reviews": int(review_data.get("totalCount", 0)),
"total_pages": (int(review_data.get("totalCount", 0)) + int(review_data.get("pageSize", 10)) - 1) // int(review_data.get("pageSize", 10)),
"current_page": int(review_data.get("page", 1)),
"page_size": int(review_data.get("pageSize", 10))
}
# 格式化评论列表
reviews = []
for review in review_data.get("comments", []):
# 处理评价内容（去除HTML标签）
content = self._clean_text(review.get("content", ""))
# 情感分析（0-1之间，越接近1越积极）
sentiment_score = self._analyze_sentiment(content)
sentiment = "positive" if sentiment_score > 0.6 else "negative" if sentiment_score < 0.4 else "neutral"
# 处理评价图片
images = []
if review.get("images"):
images = [img.get("url") for img in review.get("images") if img.get("url")]
# 处理追评
append_comment = None
if review.get("afterUserComment") and review["afterUserComment"].get("hAfterUserComment"):
append_content = self._clean_text(review["afterUserComment"]["hAfterUserComment"].get("content", ""))
if append_content:
append_comment = {
"content": append_content,
"created": review["afterUserComment"]["hAfterUserComment"].get("creationTime")
}
# 处理标签
tags = []
if review.get("tagList"):
tags = [tag.get("name") for tag in review.get("tagList") if tag.get("name")]
reviews.append({
"review_id": review.get("id"),
"user": {
"nickname": review.get("nickname"),
"level_name": review.get("userLevelName")
},
"rating": {
"score": int(review.get("score", 0)), # 总评分
"product_score": int(review.get("productScore", 0)), # 商品评分
"packing_score": int(review.get("packingScore", 0)), # 包装评分
"logistics_score": int(review.get("logisticsScore", 0)) # 物流评分
},
"content": content,
"created_time": review.get("creationTime"),
"images": images,
"append_comment": append_comment,
"useful_vote_count": int(review.get("usefulVoteCount", 0)), # 有用数
"reply_count": int(review.get("replyCount", 0)), # 回复数
"tags": tags,
"sentiment": {
"score": round(sentiment_score, 4),
"label": sentiment
},
"is_vip": review.get("isVip", False), # 是否VIP用户
"order_info": {
"product_color": review.get("productColor"),
"product_size": review.get("productSize"),
"buy_time": review.get("buyTime")
}
})
return {
"pagination": pagination,
"reviews": reviews,
"raw_data": review_data # 保留原始数据
}
def _clean_text(self, text: str) -> str:
"""清理文本，去除HTML标签和特殊字符"""
if not text:
return ""
# 去除HTML标签
clean = re.sub(r'<.*?>', '', text)
# 去除多余空格和换行
clean = re.sub(r'\s+', ' ', clean).strip()
# 去除特殊字符
clean = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9,.?!，。？！]', ' ', clean)
return clean
def _analyze_sentiment(self, text: str) -> float:
"""使用SnowNLP进行情感分析"""
if not text:
return 0.5 # 中性
try:
return SnowNLP(text).sentiments
except:
return 0.5 # 分析失败时返回中性
def get_all_reviews(self, sku_id: str, max_pages: int = 10, score: int = 0, has_image: int = 0) -> List[Dict]:
"""
获取多页评论数据
:param sku_id: 商品SKU ID
:param max_pages: 最大页数限制
:param score: 评分筛选
:param has_image: 是否有图筛选
:return: 所有评论列表
"""
all_reviews = []
page = 1
while page <= max_pages:
logging.info(f"获取第 {page} 页评论")
result = self.get_item_reviews(
sku_id=sku_id,
page=page,
page_size=50, # 使用最大页大小减少请求次数
sort_type=1, # 按时间排序
score=score,
has_image=has_image
)
if not result or not result["reviews"]:
break
all_reviews.extend(result["reviews"])
# 检查是否已到最后一页
if page >= result["pagination"]["total_pages"]:
break
page += 1
# 控制请求频率，遵守京东API的QPS限制
time.sleep(2)
return all_reviews
def analyze_reviews(self, reviews: List[Dict]) -> Dict:
"""分析评论数据，生成统计报告"""
if not reviews:
return {}
total = len(reviews)
sentiment_counts = {"positive": 0, "neutral": 0, "negative": 0}
rating_stats = {
"score": [],
"product_score": [],
"packing_score": [],
"logistics_score": []
}
tag_counts = {}
has_image_count = 0
vip_count = 0
append_comment_count = 0
useful_vote_total = 0
# 提取评论中的关键词（简单实现）
keywords = {}
positive_keywords = ["好", "不错", "满意", "推荐", "快", "值", "优秀", "棒"]
negative_keywords = ["差", "慢", "不好", "失望", "问题", "破损", "糟糕", "后悔"]
# 统计基础数据
for review in reviews:
# 情感统计
sentiment = review["sentiment"]["label"]
sentiment_counts[sentiment] += 1
# 评分统计
for key in rating_stats:
if key in review["rating"]:
rating_stats[key].append(review["rating"][key])
# 标签统计
for tag in review["tags"]:
if tag:
tag_counts[tag] = tag_counts.get(tag, 0) + 1
# 有图评价统计
if review["images"]:
has_image_count += 1
# VIP用户统计
if review["is_vip"]:
vip_count += 1
# 追评统计
if review["append_comment"]:
append_comment_count += 1
# 有用数统计
useful_vote_total += review["useful_vote_count"]
# 关键词统计
content = review["content"].lower()
for kw in positive_keywords:
if kw in content:
keywords[kw] = keywords.get(kw, 0) + 1
for kw in negative_keywords:
if kw in content:
keywords[kw] = keywords.get(kw, 0) + 1
# 计算平均评分
avg_ratings = {}
for key, values in rating_stats.items():
if values:
avg_ratings[key] = round(sum(values) / len(values), 1)
else:
avg_ratings[key] = 0
# 获取热门标签（前10）
top_tags = sorted(tag_counts.items(), key=lambda x: x[1], reverse=True)[:10]
# 获取热门关键词（前10）
top_keywords = sorted(keywords.items(), key=lambda x: x[1], reverse=True)[:10]
return {
"total_reviews": total,
"sentiment_distribution": {
"count": sentiment_counts,
"percentage": {
k: round(v / total * 100, 1) for k, v in sentiment_counts.items()
}
},
"average_rating": avg_ratings,
"image_review_ratio": round(has_image_count / total * 100, 1) if total > 0 else 0,
"vip_review_ratio": round(vip_count / total * 100, 1) if total > 0 else 0,
"append_comment_ratio": round(append_comment_count / total * 100, 1) if total > 0 else 0,
"avg_useful_votes": round(useful_vote_total / total, 1) if total > 0 else 0,
"top_tags": top_tags,
"top_keywords": top_keywords
}
# 示例调用
if __name__ == "__main__":
# 替换为实际的appkey和appsecret（从京东开放平台获取）
APPKEY = "your_appkey"
APPSECRET = "your_appsecret"
# 替换为目标商品SKU ID
SKU_ID = "100012345678"
# 初始化API客户端
api = JDItemReviewAPI(APPKEY, APPSECRET)
# 方式1：获取单页评论
# review_result = api.get_item_reviews(
# sku_id=SKU_ID,
# page=1,
# page_size=10,
# sort_type=1, # 按时间排序
# score=0, # 全部评分
# has_image=0 # 全部评论
# )
# 方式2：获取多页评论
review_result = api.get_all_reviews(
sku_id=SKU_ID,
max_pages=3,
score=0, # 全部评分
has_image=0 # 全部评论
)
if isinstance(review_result, dict) and "reviews" in review_result:
print(f"共获取到 {review_result['pagination']['total_reviews']} 条评论")
print(f"当前第 {review_result['pagination']['current_page']}/{review_result['pagination']['total_pages']} 页\n")
# 打印前3条评论
for i, review in enumerate(review_result["reviews"][:3], 1):
print(f"{i}. 用户: {review['user']['nickname']} ({'VIP' if review['is_vip'] else '普通用户'})")
print(f" 评分: {review['rating']['score']}分 (商品: {review['rating']['product_score']}, 包装: {review['rating']['packing_score']}, 物流: {review['rating']['logistics_score']})")
print(f" 时间: {review['created_time']}")
print(f" 内容: {review['content'][:100]}{'...' if len(review['content'])>100 else ''}")
print(f" 情感: {review['sentiment']['label']} (得分: {review['sentiment']['score']})")
print(f" 有用数: {review['useful_vote_count']}")
if review['images']:
print(f" 图片数: {len(review['images'])}")
if review['tags']:
print(f" 标签: {', '.join(review['tags'])}")
if review['append_comment']:
print(f" 追评: {review['append_comment']['content'][:50]}{'...' if len(review['append_comment']['content'])>50 else ''}")
print("-" * 100)
# 分析评论
analysis = api.analyze_reviews(review_result["reviews"])
print("\n=== 评论分析报告 ===")
print(f"总评论数: {analysis['total_reviews']}")
print(f"情感分布: 正面 {analysis['sentiment_distribution']['percentage']['positive']}%, 中性 {analysis['sentiment_distribution']['percentage']['neutral']}%, 负面 {analysis['sentiment_distribution']['percentage']['negative']}%")
print(f"平均评分: 总评分 {analysis['average_rating']['score']}, 商品 {analysis['average_rating']['product_score']}, 包装 {analysis['average_rating']['packing_score']}, 物流 {analysis['average_rating']['logistics_score']}")
print(f"有图评价占比: {analysis['image_review_ratio']}%")
print(f"VIP评价占比: {analysis['vip_review_ratio']}%")
print("热门标签:")
for tag, count in analysis['top_tags']:
print(f" {tag}: {count}次")
print("热门关键词:")
for kw, count in analysis['top_keywords']:
print(f" {kw}: {count}次")
elif isinstance(review_result, list):
# 处理多页评论结果
print(f"共获取到 {len(review_result)} 条评论")
# 分析评论
analysis = api.analyze_reviews(review_result)
print("\n=== 评论分析报告 ===")
print(f"总评论数: {analysis['total_reviews']}")
print(f"情感分布: 正面 {analysis['sentiment_distribution']['percentage']['positive']}%, 中性 {analysis['sentiment_distribution']['percentage']['neutral']}%, 负面 {analysis['sentiment_distribution']['percentage']['negative']}%")