一、技術(shù)可行性分析
- API存在性驗(yàn)證 小紅書官方未公開評(píng)論API，需通過逆向工程獲取接口 典型評(píng)論接口特征：參考小紅書開放平臺(tái)API接口
- 請(qǐng)求參數(shù)解析 csharp 體驗(yàn)AI代碼助手 代碼解讀復(fù)制代碼python # 典型請(qǐng)求參數(shù)示例 params = { "note_id": "64a1b2c3d4e5f6g7h8i9j0k1", # 筆記ID "sort": "newest", # 排序方式 "page": 1, # 頁(yè)碼 "page_size": 20 # 每頁(yè)數(shù)量 }
二、完整技術(shù)實(shí)現(xiàn)方案
python
import requests
import json
import time
from urllib.parse import urlencode
class XiaohongshuCommentScraper:
def __init__(self):
self.session = requests.Session()
self.headers = {
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148",
"X-Request-ID": "api.comment_list",
"Content-Type": "application/json"
}
self.base_url = "https://www.xiaohongshu.com/api/sns/note/v1/comment_list"
def get_comments(self, note_id, max_pages=5):
comments = []
for page in range(1, max_pages+1):
params = {
"note_id": note_id,
"sort": "newest",
"page": page,
"page_size": 20
}
try:
response = self.session.get(
self.base_url,
headers=self.headers,
params=urlencode(params)
)
data = response.json()
if data.get('errcode') != 0:
raise Exception(f"API Error: {data.get('errmsg')}")
page_comments = data.get('data', {}).get('comments', [])
if not page_comments:
break
comments.extend([{
"id": c["id"],
"user": c["user"]["nickname"],
"content": c["content"],
"like_count": c["like_count"],
"create_time": c["create_time"]
} for c in page_comments])
time.sleep(1.5) # 反爬延遲
except Exception as e:
print(f"Error on page {page}: {str(e)}")
break
return comments
if __name__ == "__main__":
scraper = XiaohongshuCommentScraper()
note_id = "64a1b2c3d4e5f6g7h8i9j0k1" # 替換為實(shí)際筆記ID
comments = scraper.get_comments(note_id, max_pages=3)
print(f"成功抓取 {len(comments)} 條評(píng)論")
# 保存為CSV文件
with open("comments.csv", "w", encoding="utf-8") as f:
f.write("ID,用戶,內(nèi)容,點(diǎn)贊數(shù),創(chuàng)建時(shí)間\n")
for c in comments:
f.write(f"{c['id']},{c['user']},{c['content']},{c['like_count']},{c['create_time']}\n")
三、進(jìn)階優(yōu)化方案
- 分布式爬取架構(gòu) python 體驗(yàn)AI代碼助手 代碼解讀復(fù)制代碼python from concurrent.futures import ThreadPoolExecutor def distributed_scraping(note_ids): with ThreadPoolExecutor(max_workers=5) as executor: futures = [executor.submit(scraper.get_comments, nid, 2) for nid in note_ids] results = [f.result() for f in futures] return results
- 反爬對(duì)抗策略 ruby 體驗(yàn)AI代碼助手 代碼解讀復(fù)制代碼python class AntiCrawler: @staticmethod def get_proxy(): # 返回可用代理IP return {"http": "123.45.67.89:8080"} @staticmethod def random_delay(): time.sleep(random.uniform(1, 3))
四、數(shù)據(jù)存儲(chǔ)擴(kuò)展
python
import sqlite3
class DataStorage:
def __init__(self):
self.conn = sqlite3.connect("comments.db")
self._create_table()
def _create_table(self):
cursor = self.conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS comments (
id TEXT PRIMARY KEY,
note_id TEXT,
user TEXT,
content TEXT,
like_count INTEGER,
create_time INTEGER
)
''')
self.conn.commit()
def save_comments(self, comments, note_id):
cursor = self.conn.cursor()
for c in comments:
cursor.execute('''
INSERT OR REPLACE INTO comments
VALUES (?, ?, ?, ?, ?, ?)
''', (c["id"], note_id, c["user"], c["content"], c["like_count"], c["create_time"]))
self.conn.commit()
五、法律與道德規(guī)范
- 遵守《網(wǎng)絡(luò)安全法》和《數(shù)據(jù)安全法》
- 禁止抓取用戶隱私信息（手機(jī)號(hào)、地址等）
- 控制請(qǐng)求頻率（建議QPS≤2）
- 明確標(biāo)注數(shù)據(jù)來源，禁止商業(yè)用途
該方案通過模擬移動(dòng)端請(qǐng)求實(shí)現(xiàn)評(píng)論抓取，采用分頁(yè)機(jī)制和動(dòng)態(tài)延遲策略規(guī)避反爬限制。實(shí)際部署時(shí)建議配合代理IP池和異常重試機(jī)制，確保數(shù)據(jù)采集的穩(wěn)定性和合規(guī)性。
