📘 实战教学：基于 DrissionPage 的小红书自动化采集

核心工具栈与深度逻辑解析

1. 为什么选择 DrissionPage？

传统的爬虫（如 requests）需要耗费大量时间逆向小红书的参数加密。

  • 协议与自动化双重优势：DrissionPage 既能像 Selenium 一样操控浏览器，又能像原生请求一样监听网络数据包。
  • 内置绕过：它默认抹除了 WebDriver 特征，极大地降低了被识别为机器人的概率。

2. 深度逻辑：监听 (Listen) 模式

我们不再通过解析 HTML 源码（DOM）来获取数据。

  • 数据源头截获：当浏览器点击笔记时，服务器会返回一个 JSON 数据包。
  • 逻辑流：启动监听 -> 模拟点击 -> 截获 JSON -> 字典解析 -> 自动去重 -> 批量存储。
  • 优势：JSON 数据比 HTML 包含更丰富的信息（如点赞数、精准评论数、博主 ID 等），且结构稳定，解析速度快。

🛠️ 实战运行步骤

环境搭建

确保 Python 版本 ≥\ge 3.6，并安装核心库：

pip install DrissionPage loguru DataRecorder lxml

一、 核心逻辑与采集流图

在自动化采集过程中，程序并非盲目点击。下表展示了从“初始化”到“落库”的完整逻辑链路：

1. 采集逻辑链路表

阶段动作技术实现 (DrissionPage)目的
1. 侦听启动接口监听page.listen.start('web/v1/feed')在请求发生前布好“捕获网”
2. 检索访问搜索 URLpage.get(search_url)进入特定关键词的目标区域
3. 识别定位笔记卡片page.eles('xpath://section')确定当前页面可交互的“包裹”
4. 触发穿透点击图片target_img.click(by_js=True)强制触发详情页的 XHR 数据请求
5. 捕获等待数据包回传page.listen.wait(timeout=10)截获包含点赞、详情的 JSON 源码
6. 提取字典路径提取extract_note_data(raw_json)将复杂的 JSON 转化为结构化字典
7. 闭环模拟物理退栈page.actions.type('\ue00c')发送 ESC 键关闭弹窗，重置状态

2. 采集流程流图

二、 核心技术：精准定位与 DOM 穿透

在这一阶段，我们的目标是：在成百上千的 HTML 标签中，精准“钩”出那台笔记，并强制触发它的数据接口。

  • 锚点选择 (Xpath Container)： 我们不直接找标题，而是定位最外层的 section。因为小红书的 DOM 结构中，section 标签通常携带了 data-index。通过这个索引，我们可以实现“增量去重”——即只处理新出现在视野里的笔记，不走回头路。
  • DOM 穿透与 JS 触发： 小红书的封面图上方往往覆盖着透明的点击层（用于统计或装饰）。普通的物理点击（Click）容易被遮挡。 我们使用 by_js=True。这相当于绕过了 UI 层的干扰，直接向浏览器内核发送“该元素被激活”的信号。这是保证 listen 能稳定抓到包的关键。

三、 反爬防线：行为克隆与指纹防护

小红书的风控系统主要识别的是“机械化的确定性”。

  • 物理键位模拟 (ESC 避险)： 不要频繁去寻找并点击页面上的 X 关闭按钮。由于小红书详情页是弹窗模式，按下键盘的 ESC 键（编码为 \ue00c）是最通用的退出方式。这种非坐标类指令能极大地降低被判定为“自动化脚本”的风险。
  • 滚动策略 (Adaptive Scrolling)： 程序采用 page.scroll.down(1200)。在实战中，建议配合随机等待。 为什么要等？ 小红书是异步加载（Lazy Loading）。如果滚动过快，DOM 还没渲染出新的 section，XPath 就会抓空。给页面 1-2 秒的“喘息”时间，数据流才会源源不断。

🚀 完整实战代码

import os
import time
from loguru import logger
from DrissionPage import ChromiumPage
from DataRecorder import Recorder


# 数据提取函数
def extract_note_data(data):
    # 小红书详情包解析
    note = data.get('data', {}).get('items', [{}])[0].get('note_card', {}) if isinstance(data, dict) else {}
    print(note)
    if not note:
        note = data.get('note', {}) if isinstance(data, dict) else {}

    return {
        '博主昵称': note.get('user', {}).get('nickname', '未知'),
        '标题': note.get('title', ''),
        '详情': note.get('desc', ''),
        '评论数': note.get('interact_info', {}).get('comment_count', 0),
        '点赞数': note.get('interact_info', {}).get('liked_count', 0),
        '收藏数': note.get('interact_info', {}).get('collected_count', 0),
        'note': note,
    }


def create_xlsx(keyword):
    filename = f'{keyword.strip()}.xlsx'
    if os.path.exists(filename):
        os.remove(filename)
        logger.info(f'🗑️ 已清除旧文件: {filename}')
    recorder = Recorder(filename)
    recorder.set.show_msg(False)
    return recorder


def handler(page, keyword):
    recorder = create_xlsx(keyword)
    # 1. 启动监听
    page.listen.start('web/v1/feed')

    search_url = f'https://www.xiaohongshu.com/search_result?keyword={keyword}&source=web_profile_page'
    page.get(search_url)
    page.wait.load_start()

    s = set()  # 用于 data-index 去重
    data_count = 0
    max_count = 20
    scroll_interval = 5

    logger.info(f'🚀 开始爬取关键词: 【{keyword}】')

    while data_count < max_count:
        # 2. 【核心修复】直接使用你指定的精准 XPath 获取当前页面的所有卡片
        cards = page.eles('xpath://*[@id="global"]/div[2]/div[2]/div/div/div[3]/div[1]/section')

        # 如果当前页面没加载出卡片，尝试滚动
        if not cards:
            page.scroll.down(1000)
            page.wait(1)
            continue

        for card in cards:
            if data_count >= max_count:
                break

            # 3. 唯一性去重 (使用你原来的 data-index 方案)
            index = card.attr('data-index')
            if not index or index in s:
                continue
            s.add(index)

            try:
                logger.info(f'正在爬取第 {data_count + 1} 条数据...')

                # 4. 【核心修复】执行点击逻辑 (使用你要求的精准路径)
                target_img = card.ele('xpath:./div/a[2]/img')
                if target_img:
                    target_img.click(by_js=True)
                else:
                    card.click(by_js=True)

                # 5. 等待数据包
                res = page.listen.wait(timeout=4)

                if res:
                    raw_data = res.response.body
                    info = extract_note_data(raw_data)
                    recorder.add_data(info)
                    recorder.record()  # 逐条保存或按需保存
                    data_count += 1
                else:
                    logger.warning(f"第 {data_count + 1} 条抓包超时")

                # 6. 关闭详情页 (ESC键)
                page.actions.type('\ue00c')
                page.wait(0.5, 1)  # 给页面反应时间，防止连续点击触发风控

            except Exception as e:
                logger.error(f'⚠️ 单条处理异常: {e}')
                page.actions.type('\ue00c')
                continue

        # 7. 滚动加载更多内容
        logger.info('📜 页面进行滚动以加载更多内容...')
        page.scroll.down(1200)
        page.wait(1, 2)

    logger.success(f'✅ {keyword} 采集完成，共计 {data_count} 条')


def main():
    keywords = ['苹果', '香蕉']
    page = ChromiumPage()

    page.get('https://www.xiaohongshu.com')
    input('🔐 请扫码登录，完成后在此按回车继续...')

    for keyword in keywords:
        handler(page, keyword)

    logger.info("🏁 所有任务已完成")
    page.quit()


if __name__ == '__main__':
    main()

📝 教学笔记与避坑指南

  • 监听卡死：如果发现 listen.wait 一直超时，请检查小红书是否弹出了验证码。
  • XPath 失效：若页面结构大幅变动，建议使用更为通用的 .note-item 类选择器作为保底。
  • 性能优化：在处理大规模数据（1000条以上）时，建议将 recorder.record() 移出内部循环，改在关键词采集结束后统一调用。