京东 item_review 接口深度分析及 Python 实现

简介: 京东item_review接口用于获取商品用户评论数据,涵盖评价内容、评分、晒单、追评等详细信息,适用于口碑分析、需求挖掘、服务评估等场景,是电商运营与产品优化的重要数据来源。

京东的 item_review 接口是用于获取商品用户评论数据的专业接口,能够获取京东平台上指定商品的用户评价、评分、晒单、追评等详细信息。这些评论数据对于商品口碑分析、用户需求挖掘、服务质量评估等场景具有重要价值,是电商运营和产品改进的重要数据来源。
一、接口核心特性分析

  1. 接口功能与定位
    核心功能:获取京东商品的用户评论数据,包括评价内容、评分、评价时间、用户信息、晒单图片等
    数据维度:
    基础评价:评价 ID、商品 ID、用户昵称、评价时间、评价内容
    评分数据:商品评分、包装评分、物流评分等多维度评分
    多媒体内容:晒单图片、视频
    追评信息:追加评价内容、追评时间
    互动数据:有用数、回复数、点赞数
    应用场景:
    商品口碑监控与舆情分析
    用户需求与痛点挖掘
    售后服务质量评估
    竞品评价对比分析
    产品改进建议提取
  2. 认证机制
    京东开放平台采用 appkey + access_token 的认证方式:
    开发者在京东开放平台注册应用,获取 appkey 和 appsecret
    通过 appkey 和 appsecret 获取 access_token(通常有效期为 24 小时)
    每次接口调用需在请求参数中携带有效 access_token
    评论接口属于中等敏感数据接口,需要申请相应权限
  3. 核心参数与响应结构
    请求参数
    参数名 类型 是否必填 说明
    sku_id String 是 商品 SKU ID
    access_token String 是 访问令牌
    page Integer 否 页码,默认 1
    page_size Integer 否 每页条数,默认 10,最大 50
    sort_type Integer 否 排序方式:0 - 推荐,1 - 时间,2 - 评分高,3 - 评分低
    score Integer 否 评分筛选:0 - 全部,1 - 好评,2 - 中评,3 - 差评
    has_image Integer 否 是否有图:0 - 全部,1 - 有图
    响应核心字段
    分页信息:总评论数、总页数、当前页码
    评价列表:每条评价包含
    评价基本信息:评价 ID、用户信息、评价时间
    评分信息:各维度评分
    评价内容:文本内容、标签
    多媒体:晒单图片 URL 列表
    追评:内容与时间
    互动数据:有用数、回复数
    二、Python 脚本实现
    以下是调用京东 item_review 接口的完整 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']}%")

三、接口调用注意事项

  1. 调用限制与规范
    QPS 限制:京东开放平台对评论接口的 QPS 限制通常为 5-10 次 / 秒
    数据权限:评论接口需要申请相应权限,部分高级字段可能需要额外审批
    分页限制:最多可获取前 100 页评论数据(约 5000 条)
    调用频率:批量获取时建议设置 2-3 秒间隔,避免触发频率限制
    合规使用:评论数据受用户隐私保护,不得泄露用户个人信息
  2. 常见错误及解决方案
    错误码 说明 解决方案
    401 未授权或 token 无效 重新获取 access_token,检查权限是否正确
    403 权限不足 申请评论接口的访问权限
    404 商品不存在或无评论 确认 sku_id 是否正确,该商品可能没有评论
    429 调用频率超限 降低调用频率,实现请求限流
    500 服务器内部错误 实现重试机制,最多 3 次,间隔指数退避
    10001 参数错误 检查参数格式和取值范围是否正确
  3. 数据解析要点
    文本清洗:评价内容可能包含 HTML 标签、特殊符号等,需要预处理
    评分体系:京东评分通常为 1-5 分,5 分为最高分
    时间格式:评价时间可能为时间戳或字符串,需统一转换为标准格式
    图片 URL:部分图片 URL 需要处理才能直接访问
    多维度评分:区分商品、包装、物流等不同维度的评分
    四、应用场景与扩展建议
    典型应用场景
    商品口碑监控系统:实时监控商品评价变化,及时发现负面评价
    用户需求分析工具:从评论中提取用户对产品功能、特性的需求
    服务质量评估平台:分析包装和物流评分,评估供应链服务质量
    竞品评价对比系统:对比同类商品的评价数据,找出优势与不足
    产品改进建议提取:从评论中挖掘产品改进的具体建议
    扩展建议
    实现评论关键词提取:使用 TF-IDF 或 TextRank 算法提取核心评价点
    构建情感趋势分析:追踪评论情感随时间的变化趋势
    开发负面评价预警:当负面评价比例超过阈值时触发警报
    实现评论分类模型:基于内容自动分类评论(如质量、价格、服务等)
    构建多维度评分看板:直观展示商品、包装、物流等不同维度的评分变化
    开发评论可视化系统:生成情感分布饼图、关键词云图、评分趋势图等
    通过合理使用京东 item_review 接口,开发者可以构建全面的商品评价分析系统,为电商运营决策、产品改进和用户服务优化提供数据支持。使用时需严格遵守京东开放平台的使用规范和数据保护条款,确保合法合规地获取和使用评论数据。
相关文章
|
4天前
|
供应链 监控 算法
VVICitem_get - 根据 ID 取商品详情接口深度分析及 Python 实现
VVIC(搜款网)是国内领先的服装批发电商平台,其item_get接口支持通过商品ID获取详尽的商品信息,涵盖价格、规格、库存、图片及店铺数据,助力商家高效开展市场分析、竞品监控与采购决策。
|
4天前
|
缓存 算法 数据安全/隐私保护
VVICitem_search - 根据关键词取关键词取商品列表接口深度分析及 Python 实现
VVIC item_search接口支持关键词搜索服装商品,提供价格、销量、供应商等数据,助力市场调研与采购决策。
|
5天前
|
缓存 自然语言处理 算法
item_search - Lazada 按关键字搜索商品接口深度分析及 Python 实现
Lazada的item_search接口是关键词搜索商品的核心工具,支持多语言、多站点,可获取商品价格、销量、评分等数据,适用于市场调研与竞品分析。
|
测试技术 Python
python接口自动化(五)--接口测试用例和接口测试报告模板(详解)
当今社会在测试领域,接口测试已经越来越多的被提及,被重视,而且现在好多招聘信息要对接口测试提出要求。区别于传统意义上的系统级别测试,很多测试人员在接触到接口测试的时候,也许对测试执行还可以比较顺利的上手,但一 提到相关的文档,比如测试用例和报告,就有些不知所措了。这类问题在我加入的几个测试的群里,经常看到一些人在不断提问。   今天就用这篇文章来说说接口测试用例和报告。
646 2
python接口自动化(五)--接口测试用例和接口测试报告模板(详解)
|
Web App开发 网络协议 jenkins
python接口自动化(四)--接口测试工具介绍(详解)
如果有工具,可以大大提高你的效率,可以达到事半功倍,但是不是所有工具都能够支持你完成这个任务。下面我们就来挑选几个常用和常见的工具,简单介绍一下。如果需要或者有兴趣可以在网上查看各种工具对应的资料进行深入人的学习,这里带领大家了解一下,碰到这些工具会用、了解、知道这些工具就达到目的了,不要到时候说到测试工具,一问三不知,一脸懵逼,那样就尴尬
487 1
python接口自动化(四)--接口测试工具介绍(详解)
|
安全 Java 测试技术
python接口自动化(三)--如何设计接口测试用例(详解)
上篇我们已经介绍了什么是接口测试和接口测试的意义。在开始接口测试之前,我们来想一下,如何进行接口测试的准备工作。或者说,接口测试的流程是什么?有些人就很好奇,接口测试要流程干嘛?不就是拿着接口文档直接利用接口 测试工具测试嘛。其实,如果只是三五个接口,你可以这么做一个临时的接口测试。但是,如果是上百个接口,或者,你们公司的这个项目,第一次做接口测试,那么,我们还是很有必要严格遵守接口测试的流程。
454 0
python接口自动化(三)--如何设计接口测试用例(详解)
|
存储 前端开发 安全
python接口自动化(二)--什么是接口测试、为什么要做接口测试(详解)
上一篇和大家一起科普扫盲接口后,知道什么是接口,接口类型等,对其有了大致了解之后,我们就回到主题-接口测试。
306 0
python接口自动化(二)--什么是接口测试、为什么要做接口测试(详解)
|
Java 测试技术 Python
python+pytest接口自动化(3)-接口测试一般流程及方法
首先我们要明确,通常所接口测试其实就属于功能测试,主要校验接口是否实现预定的功能,虽然有些情况下可能还需要对接口进行性能测试、安全性测试。 在学习接口自动化测试之前,我们先来了解手工接口测试怎样进行。
python+pytest接口自动化(3)-接口测试一般流程及方法
|
设计模式 Dubbo 网络协议
python+pytest接口自动化(1)-接口测试基础
API(Application Programming Interface,应用程序接口)是一些预先定义的接口(如函数、HTTP接口),或指软件系统不同组成部分衔接的约定。用来提供应用程序与开发人员基于某软件或硬件得以访问的一组例程,而又无需访问源码,或理解内部工作机制的细节。
python+pytest接口自动化(1)-接口测试基础

推荐镜像

更多