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

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

核心参数体系

响应数据结构

{"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

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

环境配置要求

依赖安装命令

pip install requests pandas tenacity python-dateutil

开放平台配置步骤

三、接口开发实战实现

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

import timeimport hashlibimport hmacimport urllib.parsedef 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 requestsimport jsonfrom typing import Dict, Optional, Listclass JdCommentAPI:def __init__(self, app_key: str, app_secret: str):self.app_key = app_keyself.app_secret = app_secretself.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 Noneexcept json.JSONDecodeError as e:print(f"JSON解析错误: {str(e)}")return None

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

import pandas as pdfrom datetime import datetimedef 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 Nonecomment_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)# 转换为DataFramedf = 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 dfdef parse_comment_summary(raw_data: Dict) -> Optional[Dict]:"""解析评论统计信息"""if not raw_data or "commentSummary" not in raw_data:return Nonesummary = 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"  # 替换为实际商品SKUTOTAL_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)} 条")

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

性能优化策略

def get_incremental_comments(api_client, sku_id, last_time=None):"""增量获取评论数据"""comments = []page = 1while True:raw_data = api_client.get_product_comments(sku_id, page=page)if not raw_data or "comments" not in raw_data:breakcomment_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:breakcomments.append(comment_df)page += 1time.sleep(1)return pd.concat(comments, ignore_index=True) if comments else None

import redisimport picklefrom datetime import timedeltaclass 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] = dataself.redis.setex(cache_key, timedelta(seconds=expire), pickle.dumps(data))return data

高可用保障措施

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_typeclass 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)