京东商品评论接口开发全指南：从数据获取到分析应用实战

商品评论作为用户购物决策的重要参考，蕴含着丰富的产品反馈和市场需求信息。京东商品评论接口为开发者提供了获取这些宝贵数据的正规渠道，在电商数据分析、竞品研究、用户体验优化等场景中发挥着关键作用。本文将全面剖析京东商品评论接口的技术细节，从接口架构、参数设计到高可用调用实现，结合完整代码示例，帮助开发者快速掌握接口开发要点，构建稳定高效的评论数据获取与分析系统。

一、接口技术架构与核心数据模型

京东商品评论接口基于开放平台服务架构，采用 RPC 风格设计，通过标准化的参数传递和签名验证机制，确保数据交互的安全性和可靠性。其核心功能是根据商品 SKU ID 和筛选条件，分页返回对应的用户评论数据，支持按时间、评分等维度进行排序筛选。

核心参数体系

京东商品评论接口的参数分为基础认证参数、查询筛选参数和扩展控制参数三大类，具体说明如下：

• 认证参数：app_key（应用标识）、timestamp（时间戳）、sign（签名），是接口调用的必要身份凭证

• 查询参数：skuId（商品 ID，必填）、page（页码，默认 1）、pageSize（每页条数，最大 50）、score（评分筛选，0 - 全部，1 - 差评，2 - 中评，3 - 好评）

• 控制参数：sortType（排序方式，0 - 默认，1 - 时间倒序，2 - 有用度倒序）、isReply（是否有商家回复，0 - 全部，1 - 有回复）

响应数据结构

接口返回的评论数据采用多层 JSON 结构封装，包含评论列表、分页信息和统计数据三大模块，典型结构如下：

{

"code": 200,

"message": "success",

"result": {

"commentSummary": { // 评论统计信息

"totalCount": 1256, // 总评论数

"goodCount": 1120, // 好评数

"generalCount": 98, // 中评数

"poorCount": 38 // 差评数

},

"comments": [ // 评论列表

{

"id": "1234567890123", // 评论ID

"userNick": "京东用户***", // 用户名（脱敏）

"score": 5, // 评分

"content": "商品质量很好，物流也很快，非常满意", // 评论内容

"creationTime": "2024-05-15 14:30:25", // 评论时间

"usefulVoteCount": 28, // 有用数

"images": [ // 评论图片

"https://img10.360buyimg.com/comment/jfs/t1/12345/6/7890/123456/abcd1234ef567890/xxx.jpg"

],

"productSize": "16GB+512GB", // 购买规格

"reply": "感谢您的好评，祝您使用愉快！" // 商家回复

}

// 更多评论...

],

"pagination": { // 分页信息

"page": 1,

"pageSize": 20,

"totalPage": 63

}

}

}

点击获取key和secret

二、开发环境搭建与前置准备

环境配置要求

• 开发语言：Python 3.8 及以上版本（推荐 3.10，支持类型注解等新特性）

• 依赖库：requests（HTTP 请求）、pandas（数据处理）、tenacity（重试机制）、python-dateutil（时间处理）

• 运行环境：支持 Windows、macOS、Linux 等主流操作系统，需联网访问京东开放平台

依赖安装命令

通过 pip 工具快速安装所需依赖库：

pip install requests pandas tenacity python-dateutil

开放平台配置步骤

1. 登录京东开放平台完成开发者账号注册与实名认证

1. 创建应用并提交审核，获取app_key和app_secret（应用密钥）

1. 在应用管理中申请 "商品评论查询" 接口权限（接口名称：jingdong.comment.product.get）

1. 查阅接口文档确认调用限制：单应用默认调用频率为 100 次 / 分钟，单次最多返回 50 条评论

三、接口开发实战实现

步骤 1：签名生成工具实现

京东接口采用 HMAC-SHA256 算法生成签名，确保请求参数未被篡改，实现代码如下：

import time

import hashlib

import hmac

import urllib.parse

def generate_jd_sign(params, app_secret):

"""

生成京东接口请求签名

:param params: 请求参数字典

:param app_secret: 应用密钥

:return: 签名字符串

"""

# 1. 按参数名ASCII码升序排序

sorted_params = sorted(params.items(), key=lambda x: x[0])

# 2. 拼接为key=value&key=value格式

query_string = urllib.parse.urlencode(sorted_params)

# 3. 使用HMAC-SHA256算法计算签名

signature = hmac.new(

app_secret.encode('utf-8'),

query_string.encode('utf-8'),

hashlib.sha256

).hexdigest().upper()

return signature

步骤 2：评论接口客户端封装

封装完整的接口调用逻辑，包括参数构建、签名生成、请求发送和响应处理：

import requests

import json

from typing import Dict, Optional, List

class JdCommentAPI:

def __init__(self, app_key: str, app_secret: str):

self.app_key = app_key

self.app_secret = app_secret

self.api_url = "https://api.jd.com/routerjson" # 京东开放平台网关

def get_product_comments(self,

sku_id: str,

page: int = 1,

page_size: int = 20,

score: int = 0,

sort_type: int = 1,

is_reply: int = 0) -> Optional[Dict]:

"""

获取商品评论数据

:param sku_id: 商品SKU ID

:param page: 页码

:param page_size: 每页条数

:param score: 评分筛选（0-全部，1-差评，2-中评，3-好评）

:param sort_type: 排序方式（0-默认，1-时间倒序，2-有用度倒序）

:param is_reply: 是否有回复（0-全部，1-有回复）

:return: 评论数据字典或None

"""

# 1. 构建请求参数

params = {

"method": "jingdong.comment.product.get", # 接口方法名

"app_key": self.app_key,

"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), # 京东时间格式

"format": "json",

"v": "2.0",

"skuId": sku_id,

"page": page,

"pageSize": page_size,

"score": score,

"sortType": sort_type,

"isReply": is_reply

}

# 2. 生成签名

params["sign"] = generate_jd_sign(params, self.app_secret)

try:

# 3. 发送POST请求

response = requests.post(

self.api_url,

data=params,

headers={"Content-Type": "application/x-www-form-urlencoded"},

timeout=15

)

# 4. 检查响应状态

response.raise_for_status()

# 5. 解析响应数据

result = json.loads(response.text)

# 6. 处理错误响应

if "error_response" in result:

error = result["error_response"]

print(f"接口调用失败: {error.get('msg')} (错误码: {error.get('code')})")

return None

# 7. 提取有效数据

return result.get("jingdong_comment_product_get_response", {}).get("result", {})

except requests.exceptions.RequestException as e:

print(f"HTTP请求异常: {str(e)}")

return None

except json.JSONDecodeError as e:

print(f"JSON解析错误: {str(e)}")

return None

步骤 3：评论数据解析与结构化

对接口返回的原始数据进行清洗和结构化处理，提取关键信息：

import pandas as pd

from datetime import datetime

def parse_comments(raw_data: Dict) -> Optional[pd.DataFrame]:

"""

解析评论原始数据为DataFrame

:param raw_data: 接口返回的原始数据

:return: 结构化的评论数据DataFrame

"""

if not raw_data or "comments" not in raw_data:

return None

comment_list = []

for comment in raw_data["comments"]:

# 解析评论图片

images = comment.get("images", [])

image_count = len(images)

first_image = images[0] if image_count > 0 else None

# 构建评论信息字典

comment_info = {

"comment_id": comment.get("id"),

"user_nick": comment.get("userNick"),

"score": comment.get("score"),

"content": comment.get("content", ""),

"creation_time": comment.get("creationTime"),

"useful_count": comment.get("usefulVoteCount", 0),

"image_count": image_count,

"first_image": first_image,

"product_size": comment.get("productSize", ""),

"has_reply": 1 if comment.get("reply") else 0,

"reply_content": comment.get("reply", "")

}

comment_list.append(comment_info)

# 转换为DataFrame

df = pd.DataFrame(comment_list)

# 数据类型转换

if not df.empty:

df["creation_time"] = pd.to_datetime(df["creation_time"])

df["score"] = df["score"].astype(int)

df["useful_count"] = df["useful_count"].astype(int)

return df

def parse_comment_summary(raw_data: Dict) -> Optional[Dict]:

"""解析评论统计信息"""

if not raw_data or "commentSummary" not in raw_data:

return None

summary = raw_data["commentSummary"]

total = summary.get("totalCount", 0)

return {

"total_count": total,

"good_count": summary.get("goodCount", 0),

"general_count": summary.get("generalCount", 0),

"poor_count": summary.get("poorCount", 0),

"good_rate": round(summary.get("goodCount", 0) / total * 100, 2) if total > 0 else 0,

"pagination": raw_data.get("pagination", {})

}

步骤 4：完整调用与数据保存示例

if __name__ == "__main__":

# 配置应用密钥（替换为实际值）

APP_KEY = "your_app_key_here"

APP_SECRET = "your_app_secret_here"

# 初始化API客户端

comment_api = JdCommentAPI(APP_KEY, APP_SECRET)

# 目标商品SKU ID和获取参数

TARGET_SKU = "100012345678" # 替换为实际商品SKU

TOTAL_PAGES = 3 # 要获取的总页数

# 存储所有评论数据

all_comments = []

summary = None

# 批量获取评论

for page in range(1, TOTAL_PAGES + 1):

print(f"获取第 {page} 页评论数据...")

# 调用接口获取好评数据（score=3）

raw_comments = comment_api.get_product_comments(

sku_id=TARGET_SKU,

page=page,

page_size=20,

score=3, # 只获取好评

sort_type=1, # 按时间倒序

is_reply=0

)

if not raw_comments:

continue

# 解析评论统计信息（只在第一页获取）

if page == 1:

summary = parse_comment_summary(raw_comments)

if summary:

print(f"评论统计: 总{summary['total_count']}条，好评率{summary['good_rate']}%")

# 解析评论数据

comment_df = parse_comments(raw_comments)

if comment_df is not None and not comment_df.empty:

all_comments.append(comment_df)

print(f"第 {page} 页解析完成，共 {len(comment_df)} 条评论")

# 控制请求频率，避免超限

time.sleep(2)

# 合并并保存数据

if all_comments:

final_df = pd.concat(all_comments, ignore_index=True)

output_file = f"jd_comments_{TARGET_SKU}.csv"

final_df.to_csv(output_file, index=False, encoding="utf-8-sig")

print(f"所有评论数据已保存至 {output_file}，共 {len(final_df)} 条")

四、接口优化与高可用设计

性能优化策略

1. 增量获取机制：记录上次获取的最新评论时间，下次只获取新增评论

def get_incremental_comments(api_client, sku_id, last_time=None):

"""增量获取评论数据"""

comments = []

page = 1

while True:

raw_data = api_client.get_product_comments(sku_id, page=page)

if not raw_data or "comments" not in raw_data:

break

comment_df = parse_comments(raw_data)

if comment_df.empty:

break

# 过滤出上次获取之后的评论

if last_time:

comment_df = comment_df[comment_df["creation_time"] > last_time]

if comment_df.empty:

break

comments.append(comment_df)

page += 1

time.sleep(1)

return pd.concat(comments, ignore_index=True) if comments else None

1. 多级缓存架构：实现内存缓存 + Redis 缓存的多级缓存策略

import redis

import pickle

from datetime import timedelta

class CachedCommentAPI(JdCommentAPI):

def __init__(self, app_key, app_secret, redis_host="localhost", redis_port=6379):

super().__init__(app_key, app_secret)

self.redis = redis.Redis(host=redis_host, port=redis_port, db=0)

self.memory_cache = {} # 内存缓存

def get_cached_comments(self, sku_id, page=1, expire=1800, **kwargs):

"""带缓存的评论获取"""

cache_key = f"jd_comment_{sku_id}_page{page}"

# 1. 检查内存缓存

if cache_key in self.memory_cache:

return self.memory_cache[cache_key]

# 2. 检查Redis缓存

cached_data = self.redis.get(cache_key)

if cached_data:

return pickle.loads(cached_data)

# 3. 缓存未命中，调用接口

data = self.get_product_comments(sku_id, page=page,** kwargs)

# 4. 更新缓存

if data:

self.memory_cache[cache_key] = data

self.redis.setex(cache_key, timedelta(seconds=expire), pickle.dumps(data))

return data

高可用保障措施

1. 智能重试机制：基于 tenacity 实现异常自动重试

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class ReliableCommentAPI(JdCommentAPI):

@retry(

stop=stop_after_attempt(3), # 最多重试3次

wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避

retry=retry_if_exception_type((requests.exceptions.RequestException, json.JSONDecodeError))

)

def get_reliable_comments(self, *args, **kwargs):

"""带重试机制的评论获取"""

return super().get_product_comments(*args, **kwargs)

1. 流量控制实现：严格控制调用频率