京東商品評論數(shù)據(jù)包含豐富的用戶反饋信息，對市場分析、產(chǎn)品改進和用戶需求挖掘具有重要價值。本文將系統(tǒng)講解京東商品評論接口的技術(shù)實現(xiàn)，重點解決接口參數(shù)構(gòu)造、反爬機制應對、數(shù)據(jù)解析與分析等核心問題，提供一套套合規(guī)高效的技術(shù)方案，同時嚴格嚴格平臺規(guī)則與數(shù)據(jù)采集規(guī)范。

一、京東評論接口評論接口原理與合規(guī)要點

京東商品評論評論數(shù)據(jù)通過 API 接口動態(tài)加載，采用 JSON 格式返回，包含評論內(nèi)容、評分、用戶信息等關(guān)鍵數(shù)據(jù)。實現(xiàn)現(xiàn)接口需遵循以下合規(guī)要點：

數(shù)據(jù)用途限制：僅用于個人學習研究、市場調(diào)研，不得得用于商業(yè)競爭或惡意分析
請求頻率控制：單 IP 單小時請求不超過 60 次，單商品評論采集間隔不低于 15 秒
用戶協(xié)議尊重：不繞過京東正常訪問限制，不使用破解手段術(shù)技術(shù)獲取數(shù)據(jù)
隱私保護：自動過濾評論中包含的手機號、地址等個人隱私信息

京東評論接口的核心技術(shù)流程如下：

plaintext

商品ID解析 → 評論參數(shù)生成 → 評論請求發(fā)送 → 數(shù)據(jù)解析與清洗 → 結(jié)構(gòu)化存儲

二、核心技術(shù)實現(xiàn)：從接口分析到數(shù)據(jù)提取

1. 京東商品 ID 解析工具

京東商品 ID（skuId）是獲取評論的基礎，可從商品 URL 或頁面元數(shù)據(jù)中提?。?/p>

python

運行

import re
import requests
from lxml import etree

class JdSkuIdParser:
"""京東商品ID解析器，提取skuId"""

def __init__(self):
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Referer": "https://www.jd.com/"
}

def parse_from_url(self, product_url):
"""從商品URL中提取skuId"""
patterns = [
r"item.jd.com/(d+).html", # 標準商品頁URL
r"sku=(d+)", # 包含sku參數(shù)的URL
r"product.jd.com/(d+).html" # 產(chǎn)品頁URL
]

for pattern in patterns:
match = re.search(pattern, product_url)
if match:
return match.group(1)
return None

def parse_from_page(self, product_url):
"""從商品頁面中提取skuId"""
try:
response = requests.get(
product_url,
headers=self.headers,
timeout=10,
allow_redirects=True
)
response.encoding = "utf-8"

# 嘗試從meta標簽提取
tree = etree.HTML(response.text)
meta_tag = tree.xpath('//meta[@name="skuId"]@conten/')
if meta_tag:
return meta_tag[0]

# 嘗試從腳本標簽提取
script_tags = tree.xpath('//script/text()')
for script in script_tags:
match = re.search(r'skuIds*=s*"(d+)"', script)
if match:
return match.group(1)
match = re.search(r'skuIds*:s*(d+)', script)
if match:
return match.group(1)

return None
except Exception as e:
print(f"頁面提取skuId失敗: {str(e)}")
return None

def get_sku_id(self, product_url):
"""獲取商品skuId，先從URL提取，失敗則從頁面提取"""
sku_id = self.parse_from_url(product_url)
if sku_id:
return sku_id
return self.parse_from_page(product_url)

2. 評論接口參數(shù)生成器

京東評論接口需要特定參數(shù)組合，包括商品 ID、頁碼、評分篩選等，部分參數(shù)需要動態(tài)生成：

python

運行

import time
import random
import hashlib

class JdCommentParamsGenerator:
"""京東評論接口參數(shù)生成器"""

def __init__(self):
# 評論類型映射
self.comment_types = {
"all": 0, # 全部評論
"good": 1, # 好評
"medium": 2, # 中評
"poor": 3, # 差評
"image": 5 # 有圖評論
}

# 排序方式映射
self.sort_types = {
"default": 5, # 默認排序
"latest": 6 # 最新排序
}

def generate_params(self, sku_id, page=1, comment_type="all", sort="default", page_size=10):
"""
生成評論請求參數(shù)

:param sku_id: 商品skuId
:param page: 頁碼
:param comment_type: 評論類型
:param sort: 排序方式
:param page_size: 每頁評論數(shù)
:return: 評論請求參數(shù)字典
"""
# 基礎參數(shù)
params = {
"productId": sku_id,
"score": self.comment_types.get(comment_type, 0),
"sortType": self.sort_types.get(sort, 5),
"page": page,
"pageSize": page_size,
"isShadowSku": 0,
"fold": 1,
"busiType": "pms",
"isProto": 0
}

# 生成動態(tài)參數(shù)
params["t"] = str(int(time.time() * 1000))
params["_"] = str(int(time.time() * 1000) + random.randint(100, 999))
params["callback"] = f"fetchJSON_comment98{random.randint(100000, 999999)}"

return params

3. 評論請求發(fā)送器

處理評論請求發(fā)送與反爬機制應對，確保請求穩(wěn)定性：

python

運行

import time
import random
import requests
from fake_useragent import UserAgent

class JdCommentRequester:
"""京東評論請求發(fā)送器"""

def __init__(self, proxy_pool=None):
self.comment_api = "https://club.jd.com/comment/productPageComments.action"
self.proxy_pool = proxy_pool or []
self.ua = UserAgent()
self.session = requests.Session()
self.last_request_time = 0
self.min_interval = 15 # 評論請求最小間隔(秒)

def _get_headers(self):
"""生成請求頭"""
return {
"User-Agent": self.ua.random,
"Accept": "*/*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Referer": "https://item.jd.com/",
"X-Requested-With": "XMLHttpRequest",
"Connection": "keep-alive",
"Host": "club.jd.com"
}

def _get_proxy(self):
"""獲取隨機代理"""
if not self.proxy_pool:
return None
return random.choice(self.proxy_pool)

def _check_request_interval(self):
"""控制請求間隔，避免觸發(fā)反爬"""
current_time = time.time()
elapsed = current_time - self.last_request_time
if elapsed < self.min_interval:
sleep_time = self.min_interval - elapsed + random.uniform(1, 3)
print(f"請求間隔不足，休眠 {sleep_time:.1f} 秒")
time.sleep(sleep_time)
self.last_request_time = time.time()

def fetch_comments(self, params):
"""
發(fā)送評論請求

:param params: 評論請求參數(shù)
:return: 響應內(nèi)容或None
"""
self._check_request_interval()

headers = self._get_headers()
proxy = self._get_proxy()
proxies = {"http": proxy, "https": proxy} if proxy else None

try:
response = self.session.get(
self.comment_api,
params=params,
headers=headers,
proxies=proxies,
timeout=15
)

if response.status_code != 200:
print(f"評論請求失敗，狀態(tài)碼: {response.status_code}")
return None

# 檢查是否被反爬攔截
if self._is_blocked(response.text):
print("評論請求被攔截，可能需要驗證")
if proxy and proxy in self.proxy_pool:
self.proxy_pool.remove(proxy)
return None

return response.text

except Exception as e:
print(f"評論請求異常: {str(e)}")
return None

def _is_blocked(self, response_text):
"""判斷是否被反爬攔截"""
block_keywords = [
"驗證碼",
"訪問過于頻繁",
"請稍后再試",
"系統(tǒng)繁忙"
]
for keyword in block_keywords:
if keyword in response_text:
return True
return False

4. 評論數(shù)據(jù)解析器

解析京東評論接口返回的 JSONP 數(shù)據(jù)，提取結(jié)構(gòu)化評論信息：

python

運行

import re
import json
from datetime import datetime

class JdCommentParser:
"""京東評論數(shù)據(jù)解析器"""

def __init__(self):
# JSONP格式解析正則
self.jsonp_pattern = re.compile(r'fetchJSON_comment98d+((.*?));')
# 隱私信息過濾正則
self.privacy_pattern = re.compile(r'1d{10}|d{6,20}') # 手機號和地址相關(guān)數(shù)字

def parse_jsonp(self, jsonp_text):
"""解析JSONP格式為JSON數(shù)據(jù)"""
match = self.jsonp_pattern.search(jsonp_text)
if not match:
return None
try:
return json.loads(match.group(1))
except json.JSONDecodeError:
print("JSON解析失敗")
return None

def clean_comment_text(self, text):
"""清理評論文本，過濾隱私信息"""
if not text:
return ""
# 過濾手機號和地址相關(guān)數(shù)字
text = self.privacy_pattern.sub('***', text)
# 去除多余空格和換行
text = re.sub(r's+', ' ', text).strip()
return text

def parse_comment_item(self, comment_item):
"""解析單個評論項"""
try:
# 解析評論時間
comment_time = comment_item.get("creationTime", "")
if comment_time:
try:
comment_time = datetime.strptime(comment_time, "%Y-%m-%d %H:%M:%S")
except ValueError:
comment_time = None

# 提取商品屬性
product_attr = comment_item.get("productColor", "")
if comment_item.get("productSize", ""):
product_attr += f" {comment_item.get('productSize')}"

# 解析圖片信息
images = comment_item.get("images", [])
image_urls = [img.get("imgUrl") for img in images if img.get("imgUrl")]

return {
"comment_id": comment_item.get("id", ""),
"user_nick": comment_item.get("nickname", ""),
"user_level": comment_item.get("userLevelName", ""),
"comment_text": self.clean_comment_text(comment_item.get("content", "")),
"comment_time": comment_time,
"score": comment_item.get("score", 0), # 評分(1-5)
"product_attr": product_attr.strip(), # 商品屬性
"useful_vote": comment_item.get("usefulVoteCount", 0), # 有用數(shù)
"image_count": len(images), # 圖片數(shù)量
"image_urls": image_urls, # 圖片URL列表
"is_vip": comment_item.get("isVip", False) # 是否VIP用戶
}
except Exception as e:
print(f"解析評論失敗: {str(e)}")
return None

def parse_comments(self, jsonp_text):
"""
解析評論列表

:param jsonp_text: JSONP格式的評論響應
:return: 包含評論和分頁信息的字典
"""
json_data = self.parse_jsonp(jsonp_text)
if not json_data:
return None

result = {
"total_comments": json_data.get("productCommentSummary", {}).get("commentCount", 0),
"good_rate": json_data.get("productCommentSummary", {}).get("goodRate", 0), # 好評率
"current_page": json_data.get("page", 1),
"page_size": json_data.get("pageSize", 10),
"comments": []
}

# 計算總頁數(shù)
result["total_pages"] = (result["total_comments"] + result["page_size"] - 1) // result["page_size"]

# 解析評論列表
comment_items = json_data.get("comments", [])
for item in comment_items:
comment = self.parse_comment_item(item)
if comment:
result["comments"].append(comment)

return result

5. 評論數(shù)據(jù)分析工具

對采集的評論數(shù)據(jù)進行多維度分析，提取有價值信息：

python

運行

import jieba
import jieba.analyse
import pandas as pd
import matplotlib.pyplot as plt
from collections import Counter

# 設置中文顯示
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]

class JdCommentAnalyzer:
"""京東評論數(shù)據(jù)分析工具"""

def __init__(self):
jieba.initialize()

def get_score_distribution(self, comments):
"""獲取評分分布"""
if not comments:
return {}
scores = [comment["score"] for comment in comments]
return dict(Counter(scores))

def extract_keywords(self, comments, top_k=20):
"""提取評論關(guān)鍵詞"""
if not comments:
return []
# 合并所有評論文本
text = " ".join([comment["comment_text"] for comment in comments if comment["comment_text"]])
# 提取關(guān)鍵詞
return jieba.analyse.extract_tags(text, topK=top_k, withWeight=True)

def get_product_attr_analysis(self, comments):
"""分析不同商品屬性的評價情況"""
if not comments:
return {}
attr_scores = {}
for comment in comments:
attr = comment["product_attr"]
if not attr:
continue
if attr not in attr_scores:
attr_scores[attr] = {"count": 0, "total_score": 0}
attr_scores[attr]["count"] += 1
attr_scores[attr]["total_score"] += comment["score"]

# 計算平均分
for attr in attr_scores:
attr_scores[attr]["avg_score"] = round(
attr_scores[attr]["total_score"] / attr_scores[attr]["count"], 1
)

# 按評價數(shù)量排序
return dict(sorted(attr_scores.items(), key=lambda x: x[1]["count"], reverse=True))

def generate_analysis_report(self, comments, output_file=None):
"""生成分析報告"""
if not comments:
return None

report = {
"total_comments": len(comments),
"score_distribution": self.get_score_distribution(comments),
"top_keywords": self.extract_keywords(comments),
"product_attr_analysis": self.get_product_attr_analysis(comments)
}

# 可視化評分分布
self._plot_score_distribution(report["score_distribution"])

# 可視化關(guān)鍵詞
self._plot_keywords(report["top_keywords"])

# 保存報告
if output_file:
import json
with open(output_file, "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2, default=str)
print(f"分析報告已保存至: {output_file}")

return report

def _plot_score_distribution(self, score_dist):
"""繪制評分分布圖表"""
if not score_dist:
return
plt.figure(figsize=(10, 6))
scores = sorted(score_dist.keys())
counts = [score_dist[s] for s in scores]
plt.bar(scores, counts, color='skyblue')
plt.title('評論評分分布')
plt.xlabel('評分')
plt.ylabel('評論數(shù)量')
plt.xticks(scores)
plt.tight_layout()
plt.show()

def _plot_keywords(self, keywords):
"""繪制關(guān)鍵詞圖表"""
if not keywords:
return
plt.figure(figsize=(12, 8))
words = [kw[0] for kw in keywords[:15]]
weights = [kw[1] for kw in keywords[:15]]
plt.barh(words, weights, color='lightgreen')
plt.title('評論關(guān)鍵詞權(quán)重')
plt.xlabel('權(quán)重')
plt.tight_layout()
plt.show()

三、完整評論采集與分析服務

整合上述組件，實現(xiàn)完整的評論采集與分析流程：

python

運行

class JdCommentService:
"""京東商品評論采集與分析服務"""

def __init__(self, proxy_pool=None):
self.sku_parser = JdSkuIdParser()
self.params_generator = JdCommentParamsGenerator()
self.requester = JdCommentRequester(proxy_pool=proxy_pool)
self.parser = JdCommentParser()
self.analyzer = JdCommentAnalyzer()

def collect_comments(self, product_url, max_pages=5, comment_type="all", sort="default"):
"""
采集商品評論

:param product_url: 商品詳情頁URL
:param max_pages: 最大采集頁數(shù)
:param comment_type: 評論類型
:param sort: 排序方式
:return: 包含評論數(shù)據(jù)的字典
"""
# 1. 獲取商品skuId
print("解析商品ID...")
sku_id = self.sku_parser.get_sku_id(product_url)
if not sku_id:
print("無法獲取商品ID，采集失敗")
return None
print(f"商品skuId: {sku_id}")

all_comments = []
current_page = 1
total_pages = 1
good_rate = 0

# 2. 分頁采集評論
while current_page <= max_pages and current_page <= total_pages:
print(f"采集第 {current_page}/{max_pages} 頁評論...")

# 生成請求參數(shù)
params = self.params_generator.generate_params(
sku_id=sku_id,
page=current_page,
comment_type=comment_type,
sort=sort
)

# 發(fā)送請求
response_text = self.requester.fetch_comments(params)
if not response_text:
print(f"第 {current_page} 頁評論獲取失敗，跳過")
current_page += 1
continue

# 解析評論
result = self.parser.parse_comments(response_text)
if not result:
print(f"第 {current_page} 頁評論解析失敗，跳過")
current_page += 1
continue

# 更新分頁信息
total_pages = min(result["total_pages"], max_pages)
good_rate = result["good_rate"]
# 添加評論
all_comments.extend(result["comments"])

print(f"第 {current_page} 頁解析完成，獲取 {len(result['comments'])} 條評論")

# 檢查是否已采集所有評論
if len(all_comments) >= result["total_comments"]:
print("已獲取全部評論，停止采集")
break

current_page += 1

# 返回結(jié)果
return {
"sku_id": sku_id,
"product_url": product_url,
"total_collected": len(all_comments),
"good_rate": good_rate,
"pages_collected": current_page - 1,
"comments": all_comments
}

def collect_and_analyze(self, product_url, max_pages=5, comment_type="all", sort="default"):
"""采集并分析評論"""
# 采集評論
comment_data = self.collect_comments(
product_url=product_url,
max_pages=max_pages,
comment_type=comment_type,
sort=sort
)

if not comment_data or not comment_data["comments"]:
print("沒有評論數(shù)據(jù)可分析")
return comment_data

# 分析評論
print("開始分析評論數(shù)據(jù)...")
analysis_report = self.analyzer.generate_analysis_report(
comment_data["comments"],
output_file=f"jd_comment_analysis_{comment_data['sku_id']}.json"
)

# 合并結(jié)果
comment_data["analysis_report"] = analysis_report
return comment_data

四、使用示例與數(shù)據(jù)存儲

1. 基本使用示例

python

運行

def main():
# 代理池（實際使用時替換為有效代理）
proxy_pool = [
# "http://123.123.123.123:8080",
# "http://111.111.111.111:8888"
]

# 初始化評論服務
comment_service = JdCommentService(proxy_pool=proxy_pool)

# 京東商品URL
product_url = "https://item.jd.com/100012345678.html" # 替換為實際商品URL

# 采集并分析評論（最多3頁，全部評論，最新排序）
result = comment_service.collect_and_analyze(
product_url=product_url,
max_pages=3,
comment_type="all",
sort="latest"
)

# 處理結(jié)果
if result:
print(f"n采集完成！共獲取 {result['total_collected']} 條評論，好評率: {result['good_rate']*100:.1f}%")

# 打印部分評論
if result["comments"]:
print("n前3條評論：")
for i, comment in enumerate(result["comments"][:3], 1):
print(f"{i}. {comment['comment_text'][:100]}...")
print(f" 評分：{comment['score']}星 | 時間：{comment['comment_time']}")
print(f" 商品屬性：{comment['product_attr']}n")
else:
print("評論采集失敗")

if __name__ == "__main__":
main()

2. 評論數(shù)據(jù)存儲工具

將評論數(shù)據(jù)存儲為多種格式，方便后續(xù)分析：

python

運行

import json
import csv
import pandas as pd
from pathlib import Path
from datetime import datetime

class JdCommentStorage:
"""京東評論數(shù)據(jù)存儲工具"""

def __init__(self, storage_dir="./jd_comments"):
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(exist_ok=True, parents=True)

def save_to_json(self, comment_data):
"""保存為JSON格式"""
sku_id = comment_data["sku_id"]
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"jd_comments_{sku_id}_{timestamp}.json"
file_path = self.storage_dir / filename

with open(file_path, "w", encoding="utf-8") as f:
json.dump(comment_data, f, ensure_ascii=False, indent=2, default=str)

print(f"JSON文件已保存：{file_path}")
return file_path

def save_to_csv(self, comment_data):
"""保存為CSV格式"""
if not comment_data["comments"]:
print("無評論數(shù)據(jù)可保存為CSV")
return None

sku_id = comment_data["sku_id"]
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"jd_comments_{sku_id}_{timestamp}.csv"
file_path = self.storage_dir / filename

# 轉(zhuǎn)換為DataFrame
df = pd.DataFrame(comment_data["comments"])
# 處理時間格式
if "comment_time" in df.columns:
df["comment_time"] = df["comment_time"].apply(
lambda x: x.strftime("%Y-%m-%d %H:%M:%S") if x else ""
)
# 處理列表類型
if "image_urls" in df.columns:
df["image_urls"] = df["image_urls"].apply(lambda x: ",".join(x) if x else "")

df.to_csv(file_path, index=False, encoding="utf-8-sig")
print(f"CSV文件已保存：{file_path}")
return file_path

def save_to_excel(self, comment_data):
"""保存為Excel格式"""
if not comment_data["comments"]:
print("無評論數(shù)據(jù)可保存為Excel")
return None

sku_id = comment_data["sku_id"]
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"jd_comments_{sku_id}_{timestamp}.xlsx"
file_path = self.storage_dir / filename

# 轉(zhuǎn)換為DataFrame
df = pd.DataFrame(comment_data["comments"])
# 處理時間格式
if "comment_time" in df.columns:
df["comment_time"] = df["comment_time"].apply(
lambda x: x.strftime("%Y-%m-%d %H:%M:%S") if x else ""
)

df.to_excel(file_path, index=False)
print(f"Excel文件已保存：{file_path}")
return file_path

五、合規(guī)優(yōu)化與風險提示

1. 系統(tǒng)優(yōu)化策略

智能緩存機制：對已采集的商品評論建立緩存，設置合理過期時間

python

運行

def get_cached_comments(self, sku_id, max_age=86400):
"""從緩存獲取評論數(shù)據(jù)（實際實現(xiàn)需結(jié)合緩存系統(tǒng)）"""
# 緩存邏輯實現(xiàn)...
return None

動態(tài)請求調(diào)整：根據(jù)響應狀態(tài)動態(tài)調(diào)整請求間隔和代理使用策略

分布式任務調(diào)度：大規(guī)模采集時采用任務分片，分散請求壓力

2. 合規(guī)與風險提示

商業(yè)用途必須獲得京東平臺書面授權(quán)，遵守《電子商務法》相關(guān)規(guī)定
不得將采集的評論數(shù)據(jù)用于生成與京東競爭的產(chǎn)品或服務
嚴格控制請求頻率，避免對平臺服務器造成負擔
自動過濾評論中的用戶隱私信息，保護用戶數(shù)據(jù)安全
當檢測到平臺反爬機制加強時，應立即暫停采集并評估風險

通過本文提供的技術(shù)方案，可構(gòu)建一套功能完善的京東商品評論接口系統(tǒng)。該方案遵循合規(guī)原則，實現(xiàn)了從評論采集、解析到分析的全流程處理，為商品研究、用戶需求分析等場景提供數(shù)據(jù)支持。在實際應用中，需根據(jù)平臺規(guī)則動態(tài)調(diào)整策略，確保系統(tǒng)的穩(wěn)定性和合法性。

審核編輯 黃宇