https://www.xiaohongshu.com/api/sns/note/v1/comment_listpython# 典型请求参数示例params = {"note_id": "64a1b2c3d4e5f6g7h8i9j0k1", # 笔记ID"sort": "newest", # 排序方式"page": 1, # 页码"page_size": 20 # 每页数量}pythonimport requestsimport jsonimport timefrom urllib.parse import urlencodeclass 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:breakcomments.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)}")breakreturn commentsif __name__ == "__main__":scraper = XiaohongshuCommentScraper()note_id = "64a1b2c3d4e5f6g7h8i9j0k1" # 替换为实际笔记IDcomments = scraper.get_comments(note_id, max_pages=3)print(f"成功抓取 {len(comments)} 条评论")# 保存为CSV文件with open("comments.csv", "w", encoding="utf-8") as f:f.write("ID,用户,内容,点赞数,创建时间\n")for c in comments:f.write(f"{c['id']},{c['user']},{c['content']},{c['like_count']},{c['create_time']}\n")
分布式爬取架构
pythonfrom concurrent.futures import ThreadPoolExecutordef 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
反爬对抗策略
pythonclass AntiCrawler: @staticmethoddef get_proxy():# 返回可用代理IPreturn {"http": "123.45.67.89:8080"} @staticmethoddef random_delay():time.sleep(random.uniform(1, 3))pythonimport sqlite3class 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()
该方案通过模拟移动端请求实现评论抓取,采用分页机制和动态延迟策略规避反爬限制。实际部署时建议配合代理IP池和异常重试机制,确保数据采集的稳定性和合规性。
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70046834/viewspace-3090423/ ,如需转载,请注明出处,否则将追究法律责任。