# 用DrissionPage构建高稳定性的社交媒体数据采集器：以音视频评论为例 最近在帮一个做内容分析的朋友处理数据需求，他需要持续追踪某个热门话题下的用户反馈，但手动收集不仅效率低下，还容易遗漏关键信息。这让我重新审视了Python生态中的网页自动化工具——除了老牌的Selenium和轻量的Requests，还有一个相对低调但实力不俗的选择：DrissionPage。 如果你也在寻找一个既能处理动态加载内容，又不需要复杂浏览器驱动管理的方案，那么今天分享的这套思路或许能给你带来一些启发。我们不会照搬任何现成的教程，而是从实际工程角度出发，探讨如何构建一个**健壮、可维护、适应性强**的数据采集器。目标读者是那些已经熟悉Python基础语法，但在处理复杂网页交互时感到棘手的开发者。 ## 1. 为什么选择DrissionPage？重新定义自动化工具选型 当我们需要从现代社交媒体平台采集数据时，面临的第一个挑战往往是工具链的选择。传统的`Requests`库简单直接，但对于大量依赖JavaScript渲染的页面束手无策；`Selenium`功能强大，但需要匹配特定版本的浏览器驱动，环境配置常常让人头疼。而`Playwright`或`Puppeteer`虽然现代，但学习曲线相对陡峭。 **DrissionPage** 在这中间找到了一个巧妙的平衡点。它底层基于`requests-html`和`undetected-chromedriver`，但通过精心设计的API将这些能力封装得更加友好。我最欣赏它的几个特点： * **无驱动依赖**：你不需要单独下载ChromeDriver或GeckoDriver，DrissionPage会自动处理浏览器启动问题。 * **混合模式**：可以在同一个会话中无缝切换“请求模式”（类似Requests）和“浏览器模式”（类似Selenium），根据页面特性选择最高效的访问方式。 * **智能等待与元素定位**：内置了多种等待策略和灵活的元素选择器，减少了编写冗余等待代码的需要。 * **网络监听能力**：这是采集动态加载数据的利器，可以直接拦截和分析XHR/Fetch请求，获取原始JSON数据。 下面这个简单的对比表，可以帮助你快速理解不同工具在数据采集场景下的定位： | 特性维度 | Requests | Selenium | Playwright | **DrissionPage** | | :--- | :--- | :--- | :--- | :--- | | **JavaScript支持** | 不支持 | 完全支持 | 完全支持 | 完全支持 | | **驱动/环境配置** | 无需 | 复杂，需版本匹配 | 自动安装，较简单 | **无需，自动管理** | | **执行速度** | **极快** | 慢 | 中等 | 中等偏快 | | **API简洁度** | 简单 | 较复杂 | 较复杂 | **简洁直观** | | **网络请求监听** | 不支持 | 需插件或复杂配置 | 支持 | **原生支持，API简单** | | **适用场景** | 静态页面、API调用 | 复杂交互、E2E测试 | 复杂交互、跨浏览器测试 | **动态页面数据采集、轻量自动化** | > 提示：工具选型没有绝对的好坏，只有是否适合当前场景。对于需要高频采集、页面交互逻辑复杂的社交媒体数据，DrissionPage的“开箱即用”和“混合模式”特性往往能显著降低开发和维护成本。 ## 2. 实战核心：逆向分析与数据包监听策略 直接模拟点击和滚动来获取评论，是最直观的方法，但往往也是最脆弱的。页面UI的任何微小改动都可能导致你的定位器失效。更高级、更稳定的做法是**理解数据是如何被加载的**，然后直接从源头获取。 现代Web应用（包括各大社交媒体平台）普遍采用前后端分离架构。你在页面上看到的评论列表，通常不是直接写在HTML里的，而是浏览器执行JavaScript后，通过Ajax（XHR/Fetch）请求从服务器获取JSON数据，再动态渲染到页面上的。我们的目标就是找到并模拟这个请求。 ### 2.1 开启开发者工具，定位关键请求 1. **打开目标页面**：在Chrome或Edge浏览器中，打开一个包含评论的音视频页面。 2. **进入Network面板**：按 `F12` 打开开发者工具，切换到 `Network`（网络）选项卡。 3. **筛选XHR/Fetch请求**：在筛选器中选择 `XHR` 或 `Fetch`。清除当前记录，然后触发评论加载（如点击“展开评论”或向下滚动）。 4. **寻找评论数据请求**：观察新出现的请求，重点关注请求URL中包含 `comment`、`list`、`api` 等关键词的条目。点击该请求，查看其 `Preview`（预览）或 `Response`（响应）标签页，确认里面是否包含结构化的评论数据（通常是JSON格式）。 你会发现，请求的URL可能像这样：`https://www.example.com/api/comment/list/?aweme_id=...&cursor=...`。其中 `aweme_id` 是视频的唯一标识，`cursor` 是分页游标。这就是我们的“数据接口”。 ### 2.2 使用DrissionPage监听并获取数据 DrissionPage的 `listen` 模块让拦截这类请求变得异常简单。我们不需要去手动解析和复制复杂的请求头、Cookie，只需要告诉它监听包含特定关键词的URL即可。 ```python from DrissionPage import ChromiumPage # 创建页面对象 page = ChromiumPage() # 开始监听所有URL中包含 'comment/list' 的请求 page.listen.start('comment/list/') # 访问目标页面 page.get('https://www.douyin.com/video/your_video_id') # 等待并获取第一个匹配的响应 response = page.listen.wait() # 响应体通常已经是解析好的JSON data = response.response.body print(data.keys()) # 查看数据结构 ``` 这段代码的精髓在于 `page.listen.wait()`，它会阻塞程序，直到监听到符合条件的请求并完成响应。获取到的 `data` 就是最原始的API返回数据，比从HTML中解析要干净、可靠得多。 ## 3. 构建健壮的采集器：工程化思维与代码设计 掌握了核心的数据获取方法后，我们需要用工程化的思维来搭建整个采集流程。一个好的采集脚本应该具备错误处理、日志记录、数据持久化和一定的反反爬虫能力。 ### 3.1 项目结构与配置管理 不建议将所有代码都写在一个文件里。一个清晰的结构有助于长期维护： ``` douyin_comment_crawler/ ├── config.py # 配置文件，存放URL模板、关键词、请求头等 ├── crawler.py # 核心爬虫逻辑 ├── storage.py # 数据存储相关（文件、数据库） ├── utils.py # 工具函数（日志、请求重试等） └── main.py # 主程序入口 ``` 在 `config.py` 中，我们可以定义一些常量： ```python # config.py class Config: # 目标视频ID列表 TARGET_AWEME_IDS = [ '7467513490379509043', # ... 其他视频ID ] # 监听URL的关键词 LISTEN_KEYWORD = 'comment/list/' # 请求头（可模拟更真实的浏览器） HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...', } # 数据存储路径 OUTPUT_DIR = './data' ``` ### 3.2 核心采集类实现 在 `crawler.py` 中，我们实现一个 `CommentCrawler` 类，封装所有采集逻辑。 ```python # crawler.py import time import logging from DrissionPage import ChromiumPage, SessionPage from typing import Optional, Dict, List from dataclasses import dataclass from config import Config @dataclass class Comment: """评论数据类""" nickname: str ip_location: str content: str create_time: int aweme_id: str class CommentCrawler: def __init__(self, headless: bool = True): """ 初始化爬虫 :param headless: 是否使用无头模式（不显示浏览器界面） """ self.page = ChromiumPage(headless=headless) self.page.set.user_agent(Config.HEADERS.get('User-Agent')) self.logger = logging.getLogger(__name__) # 启动监听 self.page.listen.start(Config.LISTEN_KEYWORD) def fetch_comments_by_aweme_id(self, aweme_id: str, max_pages: int = 10) -> List[Comment]: """ 根据视频ID获取评论 :param aweme_id: 视频唯一ID :param max_pages: 最大尝试翻页次数 :return: 评论对象列表 """ comments = [] url = f"https://www.douyin.com/video/{aweme_id}" self.logger.info(f"开始采集视频 {aweme_id} 的评论，URL: {url}") try: self.page.get(url) # 等待页面基本加载，可以尝试点击“展开评论”按钮（如果存在） # 这里使用更稳健的定位方式，避免CSS选择器因UI更新失效 comment_btn = self.page.ele('@class=comment-btn or @class=展开评论 or text()=展开评论', timeout=5) if comment_btn: comment_btn.click() time.sleep(1) for page_num in range(1, max_pages + 1): self.logger.debug(f"正在获取第 {page_num} 页评论...") # 等待并获取评论数据包 resp = self.page.listen.wait(timeout=15) if not resp: self.logger.warning(f"第 {page_num} 页未监听到评论数据，可能已加载完毕或超时。") break data = resp.response.body # 解析当前页评论 page_comments = self._parse_comment_data(data, aweme_id) comments.extend(page_comments) self.logger.info(f"第 {page_num} 页获取到 {len(page_comments)} 条评论。") # 判断是否还有更多评论（根据返回数据中的 has_more, cursor 等字段） if not data.get('has_more', False): self.logger.info("所有评论已加载完毕。") break # 触发下一页加载：模拟滚动或点击“加载更多” # 方法1：滚动到页面底部特定元素 # load_more_ele = self.page.ele('text()=加载更多 or @class=load-more', timeout=3) # if load_more_ele: # self.page.scroll.to_see(load_more_ele) # load_more_ele.click() # 方法2：直接执行JS滚动（更通用） self.page.run_js('window.scrollTo(0, document.body.scrollHeight);') time.sleep(2) # 等待新内容加载 except Exception as e: self.logger.error(f"采集视频 {aweme_id} 评论时发生错误: {e}", exc_info=True) finally: return comments def _parse_comment_data(self, raw_data: Dict, aweme_id: str) -> List[Comment]: """解析原始API返回的评论数据""" comment_list = [] try: # 不同平台API结构不同，这里是示例，需要根据实际响应调整 comments = raw_data.get('comments', []) for item in comments: comment = Comment( nickname=item.get('user', {}).get('nickname', ''), ip_location=item.get('ip_label', '未知'), content=item.get('text', ''), create_time=item.get('create_time', 0), aweme_id=aweme_id ) comment_list.append(comment) except KeyError as e: self.logger.error(f"解析评论数据时键错误: {e}，原始数据: {raw_data}") return comment_list def close(self): """关闭浏览器，释放资源""" self.page.quit() ``` 这个类的设计有几个关键点： 1. **使用数据类**：`Comment` 数据类让数据结构更清晰，便于后续处理和存储。 2. **分离解析逻辑**：`_parse_comment_data` 方法独立出来，方便适配不同平台或API变更。 3. **健壮的错误处理**：使用 `try...except` 捕获异常并记录日志，避免因单条数据解析失败导致整个任务崩溃。 4. **灵活的翻页控制**：通过判断API返回的 `has_more` 字段（名称可能不同）来控制循环，比盲目滚动或点击更可靠。 ### 3.3 数据存储与持久化 将数据保存到文件或数据库是必不可少的环节。在 `storage.py` 中，我们可以提供多种存储方式。 ```python # storage.py import csv import json import sqlite3 from datetime import datetime from pathlib import Path from typing import List from crawler import Comment class CommentStorage: def __init__(self, output_dir: str = './data'): self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) def save_to_csv(self, comments: List[Comment], filename: str = None): """保存评论到CSV文件""" if not filename: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'comments_{timestamp}.csv' filepath = self.output_dir / filename fieldnames = ['nickname', 'ip_location', 'content', 'create_time', 'aweme_id'] with open(filepath, 'w', encoding='utf-8-sig', newline='') as f: # utf-8-sig 解决Excel打开乱码 writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for comment in comments: writer.writerow({ 'nickname': comment.nickname, 'ip_location': comment.ip_location, 'content': comment.content, 'create_time': datetime.fromtimestamp(comment.create_time).isoformat() if comment.create_time else '', 'aweme_id': comment.aweme_id }) print(f"数据已保存至: {filepath}") def save_to_json(self, comments: List[Comment], filename: str = None): """保存评论到JSON文件""" if not filename: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'comments_{timestamp}.json' filepath = self.output_dir / filename data = [{ 'nickname': c.nickname, 'ip_location': c.ip_location, 'content': c.content, 'create_time': c.create_time, 'aweme_id': c.aweme_id } for c in comments] with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"数据已保存至: {filepath}") def save_to_sqlite(self, comments: List[Comment], db_name: str = 'comments.db'): """保存评论到SQLite数据库""" db_path = self.output_dir / db_name conn = sqlite3.connect(db_path) cursor = conn.cursor() # 创建表（如果不存在） cursor.execute(''' CREATE TABLE IF NOT EXISTS comments ( id INTEGER PRIMARY KEY AUTOINCREMENT, nickname TEXT, ip_location TEXT, content TEXT, create_time INTEGER, aweme_id TEXT, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 插入数据 for comment in comments: cursor.execute(''' INSERT INTO comments (nickname, ip_location, content, create_time, aweme_id) VALUES (?, ?, ?, ?, ?) ''', (comment.nickname, comment.ip_location, comment.content, comment.create_time, comment.aweme_id)) conn.commit() conn.close() print(f"数据已保存至数据库: {db_path}") ``` ## 4. 应对反爬策略与提升采集效率 没有任何一个公开平台会欢迎无节制的爬虫。我们的代码需要保持礼貌，并做好被限制的准备。 ### 4.1 基础反反爬虫措施 * **设置合理的请求间隔**：在翻页或请求间加入随机延时，模拟人类操作。 ```python import random import time def random_delay(min_sec=1, max_sec=3): """随机延时""" time.sleep(random.uniform(min_sec, max_sec)) ``` * **轮换User-Agent**：准备一个User-Agent列表，每次请求随机选择。 * **使用会话（Session）**：DrissionPage的 `SessionPage` 模式可以维持会话状态，在某些场景下比频繁创建新浏览器实例更高效、更不易被识别。 * **处理验证码**：虽然完全自动化解决验证码很困难，但可以设置超时和重试，并在出现验证码时通过日志报警，转为人工处理。 ### 4.2 异步与并发采集 如果需要采集大量视频的评论，串行操作会非常慢。我们可以结合 `asyncio` 和 `aiohttp`（用于API直连）或使用多线程/进程来管理多个 `ChromiumPage` 实例。 > 注意：并发控制需要格外小心。过高的并发请求会迅速触发平台的风控机制，导致IP被封。建议先从较低的并发数（如2-3个任务）开始测试，并确保每个任务都有独立的、隔离的浏览器环境或会话。 一个简单的多线程示例框架： ```python # multi_thread_crawler.py import threading from queue import Queue from crawler import CommentCrawler from config import Config def worker(task_queue: Queue, result_queue: Queue): """工作线程函数""" crawler = CommentCrawler(headless=True) while not task_queue.empty(): try: aweme_id = task_queue.get_nowait() except: break try: comments = crawler.fetch_comments_by_aweme_id(aweme_id) result_queue.put((aweme_id, comments)) finally: task_queue.task_done() crawler.close() def main(): aweme_ids = Config.TARGET_AWEME_IDS task_queue = Queue() for aid in aweme_ids: task_queue.put(aid) result_queue = Queue() threads = [] # 创建3个工作线程 for _ in range(3): t = threading.Thread(target=worker, args=(task_queue, result_queue)) t.start() threads.append(t) # 等待所有任务完成 task_queue.join() # 收集结果 all_results = [] while not result_queue.empty(): all_results.append(result_queue.get()) # ... 处理所有结果 ``` ### 4.3 监控、日志与错误恢复 一个成熟的采集系统需要有“眼睛”和“记忆”。使用Python标准库的 `logging` 模块记录运行状态、错误信息。 ```python # utils.py import logging def setup_logging(log_file='crawler.log'): logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, encoding='utf-8'), logging.StreamHandler() # 同时输出到控制台 ] ) ``` 在 `main.py` 中整合所有模块，并加入简单的错误恢复机制，比如记录成功采集的视频ID，下次运行时跳过。 ```python # main.py import logging from config import Config from crawler import CommentCrawler from storage import CommentStorage from utils import setup_logging def main(): setup_logging() logger = logging.getLogger(__name__) storage = CommentStorage(Config.OUTPUT_DIR) all_comments = [] crawler = CommentCrawler(headless=True) # 生产环境建议使用无头模式 try: for aweme_id in Config.TARGET_AWEME_IDS: logger.info(f"=== 开始处理视频: {aweme_id} ===") comments = crawler.fetch_comments_by_aweme_id(aweme_id, max_pages=20) all_comments.extend(comments) logger.info(f"视频 {aweme_id} 处理完成，共获取 {len(comments)} 条评论。") # 每处理完一个视频，可以即时保存一次，防止数据丢失 # storage.save_to_json(comments, f'comments_{aweme_id}.json') except KeyboardInterrupt: logger.info("用户中断采集。") except Exception as e: logger.critical(f"采集过程发生严重错误: {e}", exc_info=True) finally: crawler.close() # 最终保存所有数据 if all_comments: storage.save_to_csv(all_comments) storage.save_to_sqlite(all_comments) logger.info(f"所有数据保存完毕，总计 {len(all_comments)} 条评论。") else: logger.warning("未采集到任何评论数据。") if __name__ == '__main__': main() ``` 最后，记得在实际部署时，将配置中的目标ID、请求头等参数替换成你自己的。运行环境最好使用固定的Python版本（如3.8+），并通过 `requirements.txt` 管理依赖：`DrissionPage>=3.0.0`。这套方案的核心优势在于其**可维护性**和**适应性**。当目标网站的API或页面结构发生变化时，你通常只需要调整 `_parse_comment_data` 方法中的解析逻辑，或者更新 `LISTEN_KEYWORD`，而无需重写整个交互流程。