📚 playwright实战：动态内容处理 —— 分页、无限滚动与懒加载 – pd的爬虫笔记

📚 动态内容处理 —— 分页、无限滚动与懒加载

🎯 本讲目标

1. ✅ 掌握 无限滚动（Infinite Scroll） 的自动化处理
2. ✅ 实现 分页爬取（Pagination） 的通用封装
3. ✅ 学会用 Locator.count() + page.wait_for_function() 做精准等待
4. ✅ 所有操作都基于已有的 create_stealth_browser()，零重复代码

🔧 场景一：无限滚动（如小红书、微博、抖音列表）–实战爬取小红书首页

这类页面不会跳转 URL，而是滚动到底部后动态加载新内容。

✅ 核心策略

• 滚动前记录当前 item 数量
• 滚动到底部
• 等待 item 数量增加（说明新数据来了）
• 重复直到无新数据 or 达到目标数量

💡 通用函数封装

async def scroll_to_load(
    page,
    item_selector: str,
    max_items: int = 50,
    scroll_pause: float = 1.5,
    timeout_per_scroll: int = 10000,
):
    """
    自动滚动页面直到加载足够多的元素或无法加载更多
    :param page: Playwright Page 对象
    :param item_selector: 列表项的选择器，如 ".note-item" 或 "[data-testid='post']"
    :param max_items: 最大期望加载条数
    :param scroll_pause: 每次滚动后的等待时间（秒）
    :param timeout_per_scroll: 每次等待新内容的超时（毫秒）
    """
    items_locator = page.locator(item_selector)
    
    last_count = 0
    while True:
        current_count = await items_locator.count()
        print(f"当前已加载 {current_count} 条数据...")
        
        if current_count >= max_items:
            print("✅ 已达到目标数量，停止滚动")
            break
            
        if current_count == last_count:
            print("⚠️ 连续两次数量未变，可能已到底部")
            break
            
        # 滚动到底部
        await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
        last_count = current_count
        
        try:
            # 等待新元素出现（最多等 timeout_per_scroll 毫秒）
            await page.wait_for_function(
                f"document.querySelectorAll('{item_selector}').length > {current_count}",
                timeout=timeout_per_scroll
            )
        except Exception as e:
            print(f"⏱️ 等待超时，可能无更多内容: {e}")
            break
            
        await asyncio.sleep(scroll_pause)  # 额外缓冲

实例使用:

async def scrape_xiaohongshu(url):
    playwright, browser, context = await create_stealth_browser(headless=False)
    try:
        page = await context.new_page()
        await page.goto(url, wait_until="domcontentloaded")
        
        # 等待首屏加载
        await page.wait_for_selector(".note-item", timeout=10000)
        # 🔍 智能处理登录弹窗：最多等 5 秒，出现就关掉
        login_popup = page.locator(".login-container")
        try:
            # 等待弹窗出现（最多 5 秒）
            await login_popup.wait_for(state="visible", timeout=5000)
            print("⚠️ 检测到登录弹窗，正在关闭...")
            
            # 尝试多种关闭方式（优先级从高到低）
            close_btn = page.locator(".login-container .close-button")
            
            if await close_btn.is_visible():
                await close_btn.click()
                # 等待弹窗消失
                await login_popup.wait_for(state="hidden", timeout=3000)
                print("✅ 登录弹窗已关闭")
                
        except Exception as e:
            # 弹窗未出现，属于正常情况
            print("ℹ️ 未检测到登录弹窗，继续执行...")
        
        # 开始滚动加载
        await scroll_to_load(
            page,
            item_selector=".note-item",
            max_items=50,
            scroll_pause=3.0
        )
        
        # 提取所有标题
        titles = await page.locator(".note-item .title").all_text_contents()
        for i, title in enumerate(titles[:10]):
            print(f"{i+1}. {title}")
            
    finally:
        await context.close()
        await browser.close()
        await playwright.stop()

if __name__ == "__main__":
    url = "https://www.xiaohongshu.com/explore"
    asyncio.run(scrape_xiaohongshu(url))

问题:只加载到28条数据就退出

而且关闭无头模式时,看到内容确实也在加载,滚动条确实滚动,但是就是加载不到目标值50个

这看似矛盾，实则揭示了小红书前端一个关键机制：

✅ 小红书 Explore 页面使用了「固定容器 + 动态替换」的渲染策略，而不是「无限追加」。

🔍 根本原因：DOM 节点被「替换」而非「追加」
虽然新数据加载了，新的 .note-item 确实被插入 DOM，但与此同时：

⚠️ 旧的 .note-item 被移除了！

• 小红书为了性能和内存控制，只保留最近 N 条（取决于浏览器窗口大小）在 DOM 中；
• 当你向下滚动：
  • 新笔记通过 API 加载；
  • 前端将顶部不可见的旧 .note-item 删除；
  • 在底部插入新的 .note-item；
• 结果：.note-item 总数几乎不变（≈28），但内容已更新。

📌 这不是「虚拟滚动」（Virtual Scrolling）的典型实现（如只渲染可视区），而是一种「滑动窗口式 DOM 更新」——总节点数恒定，内容流动。

🔬 验证方法

// 在 Console 中运行
const observer = new MutationObserver(() => {
  console.log("当前 .note-item 数量:", document.querySelectorAll('.note-item').length);
});
observer.observe(document.body, { childList: true, subtree: true });

滚动时你会看到数量在 27 ↔ 28 ↔ 29 之间微小波动，但不会持续增长。

基于唯一 ID 去重累计

✅ 解决方案：边滚动、边采集、用唯一 ID 去重
既然 DOM 节点是“流动”的，我们就不能等到最后才提取数据——必须在每次滚动后立即抓取当前可见的所有笔记信息，并通过唯一标识（如笔记链接）进行去重。
我们通过以下策略成功突破 28 条限制：

1. 以 <a href="/explore/xxx"> 作为唯一标识
   • 小红书每个笔记的 URL 包含一个 24 位十六进制 ID（如 697eb8e10000000022031f47），全局唯一，适合作为去重 key。
2. 滚动一次 → 立即采集当前所有 .note-item 的标题、作者、点赞数
   • 在元素被前端销毁前完成数据提取，避免“滚完就丢”。
3. 引入“连续无新内容”容忍机制
   • 防止因网络延迟或加载抖动误判为“到底”，提升鲁棒性。
4. 使用 window.scrollBy 模拟更自然的滚动行为
   • 相比直接跳到底部，分段滚动更接近真实用户操作，降低被反爬的概率。

import asyncio
from stealth_config import create_stealth_browser


async def scroll_to_load(
    page,
    item_selector: str,
    id_selector: str,
    max_items: int = 50,
    scroll_pause: float = 1.5,
    max_no_new_attempts: int = 3,  # 允许连续几次无新数据才停止
):
    """
    自动滚动页面直到加载足够多的元素或无法加载更多
    :param page: Playwright Page 对象
    :param item_selector: 列表项的选择器，如 ".note-item"
    :param id_selector: 子容器的唯一标识选择器, 如 "a[href^='/explore/']"
    :param max_items: 最大期望加载条数
    :param scroll_pause: 每次滚动后的等待时间（秒）
    :param max_no_new_attempts: 允许连续几次无新数据才停止
    """
    collected_notes = {}  # { href: {title, author, likes, ...} }
    no_new_count = 0  # 连续无新数据的次数
    
    while len(collected_notes) < max_items  and no_new_count < max_no_new_attempts:
        # === 1. 提取当前所有可见的 .note-item 数据 ===
        note_elements = await page.locator(item_selector).all()
        current_batch = {}

        for el in note_elements:
            try:
                # 提取笔记 ID（从 href）
                link = el.locator(id_selector).first
                href = await link.get_attribute("href") if await link.count() > 0 else None
                if not href:
                    continue

                # 避免重复处理同一 ID（本轮内）
                if href in current_batch:
                    continue

                # 提取标题
                title_el = el.locator(".title")
                title = (await title_el.text_content()).strip() if await title_el.count() > 0 else ""

                # 提取作者（可选）
                author_el = el.locator(".name")
                author = (await author_el.text_content()).strip() if await author_el.count() > 0 else ""

                # 提取点赞数（可选）
                like_el = el.locator(".like-wrapper .count")
                likes = (await like_el.text_content()).strip() if await like_el.count() > 0 else "0"

                current_batch[href] = {
                    "href": href,
                    "title": title,
                    "author": author,
                    "likes": likes,
                }

            except Exception as e:
                # 单个元素解析失败不影响整体
                print(e)
                continue
        
        # === 2. 合并到全局集合（自动去重）===
        new_notes = {k: v for k, v in current_batch.items() if k not in collected_notes}
        collected_notes.update(current_batch)

        print(f"✅ 累计采集 {len(collected_notes)} 条 | 本次新增 {len(new_notes)} 条")

        # === 3. 判断是否停止 ===
        if not new_notes:
            no_new_count += 1
            print(f"⚠️ 第 {no_new_count} 次无新内容")
            if no_new_count >= max_no_new_attempts:
                print("🛑 停止滚动：连续多次无新笔记")
                break
        else:
            no_new_count = 0

        # === 4. 滚动 ===
        await page.evaluate("window.scrollBy(0, window.innerHeight * 0.8)")
        await asyncio.sleep(scroll_pause)

    # 返回按采集顺序排列的列表（可选：按时间倒序需额外处理）
    return list(collected_notes.values())


async def scrape_xiaohongshu(url):
    playwright, browser, context = await create_stealth_browser(headless=False)
    try:
        page = await context.new_page()
        await page.goto(url, wait_until="domcontentloaded")
        
        # 等待首屏加载
        await page.wait_for_selector(".note-item", timeout=10000)
        # 🔍 智能处理登录弹窗：最多等 5 秒，出现就关掉
        login_popup = page.locator(".login-container")
        try:
            # 等待弹窗出现（最多 5 秒）
            await login_popup.wait_for(state="visible", timeout=5000)
            print("⚠️ 检测到登录弹窗，正在关闭...")
            
            # 尝试多种关闭方式（优先级从高到低）
            close_btn = page.locator(".login-container .close-button")
            
            if await close_btn.is_visible():
                await close_btn.click()
                # 等待弹窗消失
                await login_popup.wait_for(state="hidden", timeout=3000)
                print("✅ 登录弹窗已关闭")
                
        except Exception as e:
            # 弹窗未出现，属于正常情况
            print("ℹ️ 未检测到登录弹窗，继续执行...")
        
        # 🔥 核心：边滚动边采集
        notes = await scroll_to_load(
            page,
            item_selector=".note-item",
            id_selector="a[href^='/explore/']",
            max_items=50,
            scroll_pause=2.0
        )
        
        # 打印结果
        print(f"\n🎯 共采集到 {len(notes)} 条唯一笔记\n")
        for i, note in enumerate(notes[:10]):
            print(f"{i+1}. [{note['likes']}] {note['title']} —— @{note['author']}")

    finally:
        await context.close()
        await browser.close()
        await playwright.stop()

if __name__ == "__main__":
    url = "https://www.xiaohongshu.com/explore"
    asyncio.run(scrape_xiaohongshu(url))

🧪 效果验证
运行修改后的脚本，输出如下：

⚠️ 检测到登录弹窗，正在关闭...
✅ 登录弹窗已关闭
✅ 累计采集 28 条 | 本次新增 28 条
✅ 累计采集 33 条 | 本次新增 5 条
✅ 累计采集 43 条 | 本次新增 10 条
✅ 累计采集 53 条 | 本次新增 10 条

🎯 共采集到 53 条唯一笔记

1. [2476] 谢娜学古筝到底付出了多少 —— @扬州木心古筝工作室
2. [5307] 520官宣💍我們要結婚了～💖👰🏻‍♀️🤵🏻‍♂️ —— @Realjoann
3. [2915] 头一次看到字幕是用的楷体！ —— @糯米碎碎念
4. [2.1万] 过度打扮吸引同性🥹 —— @鲜鲜超鲜
5. [1.6万] 怎么拍出“生命力”📸胶片感我悟了！！ —— @地铁二号线-
6. [6813] 大数据看看我！能劝一个是一个 —— @薯条不沾酱🐣
7. [2.7万] 24岁休学在南非当汉语老师有多爽 —— @拔冰鱼（21:00直播）
8. [1751] 手绘头像-NO.79 —— @毛毛頭MMT
9. [2171] 电影版花千骨‼️陈都灵版小骨你期待吗❓ —— @小爱甜酱
10. [8443] 解谜游戏！是时候决定谁当爸爸了！ —— @玩游戏的阿舔

💡 总结

误区	正确认知
“滚动就能加载更多 DOM”	❌ 小红书只维护固定数量的 DOM 节点
“用 .count() 判断是否到底”	❌ 数量恒定 ≠ 无新数据
“最后统一提取数据”	❌ 早期 DOM 已被销毁

🔧 场景二：传统分页（如电商、新闻站）–实战爬取B站视频

爬取 B站 搜索 “大模型开发” 的前 N 页视频，提取：标题、UP 主、播放量、链接。

关键：

1. 视频元素是在 最后一个div=“video-list row” 下的 bili-video-card元素

page.locator("div.video-list.row").last.locator("div.bili-video-card").all()

2. 一页默认是加载42条视频的 但是能否获取到好像取决于浏览器窗口的大小 如果缩放到80%就能完整显示42条 但是正常打开就会只有36条显示 但locator能获取到42个元素 其中有6个是没有数据的

🔍 问题本质：B站使用了“可视区域渲染”

• B站搜索页虽然 DOM 中预置了 42 个 .bili-video-card 容器，
• 但只有进入视口（viewport）的卡片才会真正加载数据（标题、UP 主等）；
• 其余卡片是空壳或骨架屏（skeleton），innerText 为空或只有占位符；
• 当你缩放浏览器到 80%，视口变大 → 更多卡片进入可视区 → 更多被真实渲染；
• 而默认窗口下，只有前 ~36 条被渲染，后 6 条仍是空的。

✅ 解决方案：

1. 只提取“已渲染”的视频卡片
2. 扩大视口，使所有卡片都进入可视区（设置create_stealth_browser(width=2300)）

完整代码如下：

import asyncio
from stealth_config import create_stealth_browser


async def scrape_bilibili_search(
    keyword: str,
    max_pages: int = 3,
    headless: bool = True
):
    """
    爬取 B站搜索结果的前 max_pages 页
    """
    url = f"https://search.bilibili.com/all?keyword={keyword}"
    
    playwright, browser, context = await create_stealth_browser(headless=headless,width=2300)
    try:
        page = await context.new_page()
        print(f"🔍 正在搜索: {keyword}")
        await page.goto(url, wait_until="domcontentloaded", timeout=30000)

        # === 处理可能的弹窗 ===
        await handle_bilibili_popups(page)

        # === 开始分页爬取 ===
        all_videos = []
        
        for current_page in range(1, max_pages + 1):
            print(f"\n📄 正在处理第 {current_page} 页...")
            
            # 等待视频列表加载
            # video_list = page.locator("div.video-list")
            try:
                await page.locator("div.bili-video-card").first.wait_for(state="visible", timeout=10000)
            except:
                print("⚠️ 视频列表未加载，可能被拦截")
                break

            # 提取当前页所有视频信息
            videos = await extract_bilibili_videos(page)
            all_videos.extend(videos)
            print(f"✅ 第 {current_page} 页提取到 {len(videos)} 个视频")

            # 如果是最后一页，不再点击“下一页”
            if current_page >= max_pages:
                break

            # 查找“下一页”按钮
            next_btn = page.get_by_text("下一页", exact=True)
            
            # 检查按钮是否可用
            if not await next_btn.is_visible() or await next_btn.is_disabled():
                print("🔚 无更多页面，提前结束")
                break
                
            print("➡️ 点击下一页...")
            await next_btn.click()
            
            # 等待新页面加载（关键！）
            await page.wait_for_function(
                f"new URLSearchParams(window.location.search).get('page') == '{current_page + 1}'",
                timeout=10000
            )
            await asyncio.sleep(1.5)  # 缓冲

        return all_videos

    finally:
        await context.close()
        await browser.close()
        await playwright.stop()


async def handle_bilibili_popups(page):
    """处理 B站常见的弹窗"""
    # 1. 青少年模式弹窗
    teen_modal = page.locator(".teen-modal-close")
    if await teen_modal.is_visible(timeout=3000):
        print("🛡️ 检测到青少年模式弹窗，正在关闭...")
        await teen_modal.click()
    
    # 2. 登录弹窗（如果有）
    login_close = page.locator(".login-tip-close")
    if await login_close.is_visible(timeout=3000):
        print("🔑 检测到登录提示，正在关闭...")
        await login_close.click()


async def extract_bilibili_videos(page):
    """从当前页面提取 B站搜索结果视频信息（适配新版结构）"""
    videos = []
    
    # 定位所有视频卡片容器
    video_cards = await page.locator("div.video-list.row").last.locator("div.bili-video-card").all()
    for card in video_cards:
        try:
            # === 标题 + 链接 ===
            title_el = card.locator("h3.bili-video-card__info--tit").first
            title = await title_el.text_content() if await title_el.is_visible() else "无标题"
            
            link_el = card.locator("a[href^='//www.bilibili.com/video/']").first
            href = await link_el.get_attribute("href") if await link_el.is_visible() else ""
            url = f"https:{href}" if href.startswith("//") else href

            # === UP 主名称（关键：用 span 而不是 a）===
            author_span = card.locator("span.bili-video-card__info--author").first
            up_name = await author_span.text_content() if await author_span.is_visible() else "未知UP主"

            # === 发布时间 ===
            date_span = card.locator("span.bili-video-card__info--date").first
            publish_date = (await date_span.text_content()).strip(" ·") if await date_span.is_visible() else ""

            # === 播放次数 + 弹幕数 ===
            stats_item = card.locator("span.bili-video-card__stats--item >> span")
            play_count = await stats_item.nth(0).text_content() if await stats_item.nth(0).is_visible() else "0"
            danmaku_count = await stats_item.nth(1).text_content() if await stats_item.nth(1).is_visible() else "0"

            videos.append({
                "title": title.strip(),
                "up": up_name.strip(),
                "publish_date": publish_date,
                "url": url,
                "stats": f"{play_count} 播放 · {danmaku_count} 弹幕",
            })
        except Exception as e:
            # 跳过广告或异常卡片（如直播、番剧）
            print(f"遇到错误：{e}")
            continue
            
    return videos

if __name__ == "__main__":
    async def main():
        videos = await scrape_bilibili_search(
            keyword="大模型开发",
            max_pages=2,
            headless=False  # 开发时建议 False，方便调试
        )
        
        print(f"\n🎯 共爬取 {len(videos)} 个视频:\n")
        for i, v in enumerate(videos[:50], 1):
            print(f"{i}. [{v['stats']}] {v['title']}")
            print(f"   👤 UP: {v['up']} | 🔗 {v['url']}\n")

    asyncio.run(main())

输出样例：

🔍 正在搜索: 大模型开发

📄 正在处理第 1 页...
✅ 第 1 页提取到 42 个视频
➡️ 点击下一页...

📄 正在处理第 2 页...
✅ 第 2 页提取到 42 个视频

🎯 共爬取 84 个视频:

1. [3885 播放 · 90 弹幕] 【AI教程】目前B站最全最细的AI大模型零基础全套教程，2026最新版，包含所有干货！七天就能从小白到大神！少走99%的弯路！存下吧！很难找全的！！
   👤 UP: 大模型开发 | 🔗 https://www.bilibili.com/video/BV1ZMf2BzETy/

2. [4093 播放 · 85 弹幕] 【全748集】目前B站最全最细的大模型开发零基础全套教程，2026最新版，包含所有干货！七天就能从小白到大神！少走99%的弯路！存下吧！很难找全的！
   👤 UP: 大模型开发 | 🔗 https://www.bilibili.com/video/BV1wHFhzzE5g/

3. [45.3万 播放 · 228 弹幕] 一个视频给讲清楚：AI大模型应用开发学习路线，避坑指南。
   👤 UP: 码农小蟹 | 🔗 https://www.bilibili.com/video/BV1kimbBMEr9/

4. [44.2万 播放 · 1244 弹幕] 黑马程序员大模型RAG与Agent智能体项目实战教程，基于主流的LangChain技术从大模型提示词到实战项目
   👤 UP: 黑马程序员 | 🔗 https://www.bilibili.com/video/BV1yjz5BLEoY/

5. [1.6万 播放 · 137 弹幕] 2026最新AI大模型应用开发全套教程（LLM+应用落地+RAG+Agent+Langchain）从入门到精通，全部都讲明白了！通俗易懂，学完即就业！
   👤 UP: AI研究所- | 🔗 https://www.bilibili.com/video/BV1TCkKBiEPh/

...