?

商品評論接口是電商數(shù)據(jù)分析的重要入口，通過評論數(shù)據(jù)可以挖掘用戶需求、分析產(chǎn)品優(yōu)缺點、監(jiān)控輿情走向。本文將詳細講解淘寶商品評論接口的技術(shù)實現(xiàn)，重點解決評論分頁機制、反爬策略應對、數(shù)據(jù)解析與情感分析等核心問題，提供一套合規(guī)、高效的技術(shù)方案，同時嚴格遵守平臺規(guī)則與數(shù)據(jù)采集規(guī)范。

一、評論接口基礎(chǔ)原理與合規(guī)要點

淘寶商品評論數(shù)據(jù)存儲在商品詳情頁的評論模塊，通過動態(tài)加載方式呈現(xiàn)。實現(xiàn)評論接口需理解其基本原理并遵守以下合規(guī)要點：

數(shù)據(jù)范圍：僅采集商品公開評論（不包含追評、問答等非評論內(nèi)容）
請求頻率：單商品評論請求間隔不低于 10 秒，單 IP 日請求不超過 1000 次
使用規(guī)范：數(shù)據(jù)僅用于個人學習、市場調(diào)研，不得用于商業(yè)競爭或惡意分析
反爬尊重：不使用破解、繞過等方式獲取數(shù)據(jù)，模擬正常用戶瀏覽行為

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

plaintext

商品ID解析 → 評論參數(shù)構(gòu)造 → 分頁請求發(fā)送 → 評論數(shù)據(jù)提取 → 數(shù)據(jù)清洗與分析

點擊獲取key和secret

二、核心技術(shù)實現(xiàn)：從評論獲取到數(shù)據(jù)解析

1. 商品 ID 解析工具

獲取商品評論的前提是解析出正確的商品 ID（item_id），可從商品詳情頁 URL 或頁面內(nèi)容中提取：

python

運行

import re
import requests
from lxml import etree

class ProductIdParser:
"""商品ID解析器，從URL或頁面中提取item_id"""

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

def parse_from_url(self, product_url):
"""從商品URL中提取item_id"""
# 匹配常見的商品URL格式
patterns = [
r"item.taobao.com/item.htm?.*id=(d+)",
r"detail.tmall.com/item.htm?.*id=(d+)",
r"id=(d+)" # 通用匹配
]

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):
"""從商品頁面內(nèi)容中提取item_id（URL解析失敗時使用）"""
try:
response = requests.get(
product_url,
headers=self.headers,
timeout=10,
allow_redirects=True
)
response.encoding = "utf-8"

# 從頁面HTML中提取item_id
tree = etree.HTML(response.text)
# 嘗試從meta標簽提取
meta_content = tree.xpath('//meta[@name="mobile-agent"]/@content')
if meta_content:
match = re.search(r"item_id=(d+)", meta_content[0])
if match:
return match.group(1)

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

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

def get_product_id(self, product_url):
"""獲取商品ID，先嘗試從URL提取，失敗則從頁面提取"""
product_id = self.parse_from_url(product_url)
if product_id:
return product_id
return self.parse_from_page(product_url)

2. 評論參數(shù)構(gòu)造器

淘寶評論請求需要特定的參數(shù)組合，包括商品 ID、頁碼、排序方式等，其中部分參數(shù)需要動態(tài)生成：

python

運行

import time
import hashlib
import random

class CommentParamsGenerator:
"""評論請求參數(shù)生成器"""

def __init__(self):
self.app_key = "12574478" # 模擬應用標識
self.sort_types = {
"default": 0, # 默認排序
"latest": 1, # 最新評論
"good": 2, # 好評
"poor": 3 # 差評
}

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

:param item_id: 商品ID
:param page: 頁碼
:param sort: 排序方式
:param page_size: 每頁評論數(shù)
:return: 評論請求參數(shù)字典
"""
# 基礎(chǔ)參數(shù)
params = {
"itemId": item_id,
"pageNum": page,
"pageSize": page_size,
"sortType": self.sort_types.get(sort, 0),
"auctionNumId": item_id,
"userType": 0,
"platform": "h5",
"needFold": 0,
"callback": f"jsonp_{int(time.time() * 1000)}_{random.randint(1000, 9999)}"
}

# 生成時間戳和簽名
t = str(int(time.time() * 1000))
params["t"] = t
params["sign"] = self._generate_sign(params)

return params

def _generate_sign(self, params):
"""生成簽名，模擬平臺參數(shù)驗證機制"""
# 按參數(shù)名排序并拼接
sorted_params = sorted(params.items(), key=lambda x: x[0])
sign_str = "&".join([f"{k}={v}" for k, v in sorted_params if k != "sign"])
# 加入固定密鑰（僅作示例）
sign_str += "&secret=taobao_comment_demo_key"
# 計算MD5簽名
return hashlib.md5(sign_str.encode()).hexdigest().upper()

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

處理評論分頁請求，包含反爬機制應對策略：

python

運行

import time
import random
import requests
from fake_useragent import UserAgent

class CommentRequester:
"""評論請求發(fā)送器，負責發(fā)送請求并處理反爬"""

def __init__(self, proxy_pool=None):
self.comment_api = "https://h5api.m.taobao.com/h5/mtop.taobao.review.list.get/1.0/"
self.proxy_pool = proxy_pool or []
self.ua = UserAgent()
self.session = requests.Session()
self.last_request_time = 0
self.min_interval = 10 # 評論請求最小間隔(秒)

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

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

def _check_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(0, 2)
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_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
)

# 檢查響應狀態(tài)
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ù)解析器

解析評論響應內(nèi)容，提取結(jié)構(gòu)化的評論數(shù)據(jù)：

python

運行

import re
import json
from datetime import datetime

class CommentParser:
"""評論數(shù)據(jù)解析器，提取結(jié)構(gòu)化評論信息"""

def __init__(self):
# 處理JSONP格式的正則
self.jsonp_pattern = re.compile(r'jsonp_d+_d+((.*?))')
# 表情符號清理正則
self.emoji_pattern = re.compile(r'[U00010000-U0010ffff]', flags=re.UNICODE)

def parse_jsonp(self, jsonp_text):
"""解析JSONP格式響應為JSON"""
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 ""
# 去除HTML標簽
text = re.sub(r'<[^>]+>', '', text)
# 去除表情符號
text = self.emoji_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("commentTime", "")
if comment_time:
try:
comment_time = datetime.strptime(comment_time, "%Y-%m-%d %H:%M:%S")
except ValueError:
comment_time = None

# 提取商品屬性（如顏色、尺寸等）
auction_params = comment_item.get("auctionParam", "")
product_attrs = {}
if auction_params:
for param in auction_params.split(";"):
if "：" in param:
key, value = param.split("：", 1)
product_attrs[key.strip()] = value.strip()

return {
"comment_id": comment_item.get("id", ""),
"user_nick": comment_item.get("userNick", ""),
"user_level": comment_item.get("userVipLevel", 0),
"comment_text": self.clean_comment_text(comment_item.get("content", "")),
"comment_time": comment_time,
"star_rating": comment_item.get("star", 0), # 星級評分
"product_attrs": product_attrs, # 購買的商品屬性
"praise_count": comment_item.get("useful", 0), # 有用數(shù)
"reply_count": comment_item.get("replyCount", 0), # 回復數(shù)
"has_image": len(comment_item.get("images", [])) > 0 # 是否有圖
}
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 or json_data.get("ret", [""])[0] != "SUCCESS::調(diào)用成功":
return None

result = {
"total_comments": 0,
"total_pages": 0,
"current_page": 0,
"comments": []
}

data = json_data.get("data", {})
comments = data.get("comments", [])

# 提取分頁信息
result["total_comments"] = data.get("total", 0)
result["current_page"] = data.get("pageNum", 1)
page_size = data.get("pageSize", 20)
result["total_pages"] = (result["total_comments"] + page_size - 1) // page_size

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

return result

5. 評論情感分析工具

對評論內(nèi)容進行情感傾向分析，判斷好評、中評、差評：

python

運行

import jieba
import jieba.analyse
from snownlp import SnowNLP

class CommentSentimentAnalyzer:
"""評論情感分析工具"""

def __init__(self):
# 加載情感分析所需的詞典
jieba.initialize()

def get_sentiment_score(self, comment_text):
"""獲取評論情感得分（0-1，越高越正面）"""
if not comment_text:
return 0.5 # 無內(nèi)容默認中性
try:
return SnowNLP(comment_text).sentiments
except:
return 0.5

def analyze_sentiment(self, comment_text):
"""分析評論情感傾向"""
score = self.get_sentiment_score(comment_text)
if score >= 0.7:
return "positive", score
elif score <= 0.3:
return "negative", score
else:
return "neutral", score

def extract_keywords(self, comment_text, top_k=5):
"""提取評論關(guān)鍵詞"""
if not comment_text:
return []
try:
return jieba.analyse.extract_tags(comment_text, topK=top_k)
except:
return []

def process_comments(self, comments):
"""批量處理評論，添加情感分析結(jié)果"""
processed = []
for comment in comments:
sentiment, score = self.analyze_sentiment(comment["comment_text"])
keywords = self.extract_keywords(comment["comment_text"])

processed_comment = comment.copy()
processed_comment["sentiment"] = sentiment
processed_comment["sentiment_score"] = round(score, 4)
processed_comment["keywords"] = keywords

processed.append(processed_comment)
return processed

三、完整評論采集服務封裝

將上述組件整合為完整的評論采集服務：

python

運行

class TaobaoCommentService:
"""淘寶商品評論采集服務"""

def __init__(self, proxy_pool=None):
self.product_id_parser = ProductIdParser()
self.params_generator = CommentParamsGenerator()
self.requester = CommentRequester(proxy_pool=proxy_pool)
self.parser = CommentParser()
self.sentiment_analyzer = CommentSentimentAnalyzer()

def get_comments(self, product_url, max_pages=5, sort="default", analyze_sentiment=True):
"""
采集商品評論

:param product_url: 商品詳情頁URL
:param max_pages: 最大采集頁數(shù)
:param sort: 評論排序方式
:param analyze_sentiment: 是否進行情感分析
:return: 包含評論數(shù)據(jù)的字典
"""
# 1. 獲取商品ID
print("正在解析商品ID...")
item_id = self.product_id_parser.get_product_id(product_url)
if not item_id:
print("無法獲取商品ID，采集失敗")
return None
print(f"獲取商品ID成功: {item_id}")

all_comments = []
current_page = 1
total_pages = 1

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

# 生成請求參數(shù)
params = self.params_generator.generate_params(
item_id=item_id,
page=current_page,
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

# 更新總頁數(shù)
total_pages = result["total_pages"]
# 添加到結(jié)果列表
all_comments.extend(result["comments"])

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

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

current_page += 1

# 3. 情感分析
if analyze_sentiment and all_comments:
print("正在進行評論情感分析...")
all_comments = self.sentiment_analyzer.process_comments(all_comments)

# 4. 返回結(jié)果
return {
"item_id": item_id,
"product_url": product_url,
"total_collected": len(all_comments),
"total_available": result["total_comments"] if result else 0,
"pages_collected": current_page - 1,
"comments": all_comments
}

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

python

運行

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

# 初始化評論采集服務
comment_service = TaobaoCommentService(proxy_pool=proxy_pool)

# 商品詳情頁URL
product_url = "https://item.taobao.com/item.htm?id=1234567890" # 替換為實際商品URL

# 采集評論（最多3頁，按最新排序，進行情感分析）
result = comment_service.get_comments(
product_url=product_url,
max_pages=3,
sort="latest",
analyze_sentiment=True
)

# 處理采集結(jié)果
if result:
print(f"n采集完成！共獲取 {result['total_collected']} 條評論")

# 打印部分結(jié)果
if result["comments"]:
print("n前3條評論摘要：")
for i, comment in enumerate(result["comments"][:3], 1):
print(f"{i}. {comment['comment_text'][:50]}...")
print(f" 情感傾向：{comment['sentiment']}（得分：{comment['sentiment_score']}）")
print(f" 關(guān)鍵詞：{','.join(comment['keywords'])}")
print(f" 發(fā)布時間：{comment['comment_time']}n")
else:
print("評論采集失敗")

if __name__ == "__main__":
main()

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

將采集的評論數(shù)據(jù)存儲為 JSON 和 CSV 格式：

python

運行

import json
import csv
from pathlib import Path
from datetime import datetime

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

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

def save_to_json(self, comment_data):
"""保存評論數(shù)據(jù)到JSON文件"""
item_id = comment_data["item_id"]
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"comments_{item_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"評論數(shù)據(jù)已保存至JSON文件：{file_path}")
return file_path

def save_to_csv(self, comment_data):
"""保存評論數(shù)據(jù)到CSV文件"""
if not comment_data["comments"]:
print("沒有評論數(shù)據(jù)可保存到CSV")
return None

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

# 評論字段
fields = [
"comment_id", "user_nick", "user_level", "comment_text",
"comment_time", "star_rating", "praise_count", "reply_count",
"has_image", "sentiment", "sentiment_score", "keywords"
]

with open(file_path, "w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fields)
writer.writeheader()

for comment in comment_data["comments"]:
# 處理嵌套字段
row = {k: comment.get(k, "") for k in fields}
# 關(guān)鍵詞列表轉(zhuǎn)為字符串
if "keywords" in row and isinstance(row["keywords"], list):
row["keywords"] = ",".join(row["keywords"])
# 時間轉(zhuǎn)為字符串
if isinstance(row["comment_time"], datetime):
row["comment_time"] = row["comment_time"].strftime("%Y-%m-%d %H:%M:%S")

writer.writerow(row)

print(f"評論數(shù)據(jù)已保存至CSV文件：{file_path}")
return file_path

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

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

評論緩存機制：對已采集的商品評論進行緩存，避免重復請求

python

運行

def get_cached_comments(self, item_id, max_age=3600):
"""從緩存獲取評論（實際實現(xiàn)需結(jié)合Redis或文件緩存）"""
# 緩存邏輯實現(xiàn)...
return None

分布式采集：大規(guī)模采集時采用分布式架構(gòu)，分散請求壓力

動態(tài)調(diào)整策略：根據(jù)反爬強度動態(tài)調(diào)整請求間隔和代理使用頻率

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

商業(yè)應用前必須獲得平臺授權(quán)，遵守《電子商務法》相關(guān)規(guī)定
評論數(shù)據(jù)不得用于生成與原平臺競爭的產(chǎn)品或服務
避免采集包含用戶隱私的評論內(nèi)容（如手機號、地址等）
當平臺明確限制評論采集時，應立即停止相關(guān)操作
采集行為不得對平臺正常運營造成影響，尊重 robots 協(xié)議限制

通過本文提供的技術(shù)方案，可構(gòu)建一套功能完善的淘寶商品評論接口系統(tǒng)。該方案注重合規(guī)性和可擴展性，能夠有效應對電商平臺的反爬機制，為商品分析、用戶反饋挖掘等場景提供數(shù)據(jù)支持。在實際應用中，需根據(jù)平臺規(guī)則動態(tài)調(diào)整策略，確保系統(tǒng)的穩(wěn)定性和合法性。

?審核編輯 黃宇