1. 项目概述与核心需求解析
1.1 这个项目要解决什么问题
说句实在话,做爬虫的人迟早都会碰到动态加载页面。以前用 requests 写爬虫,最烦的就是目标页面用了 JavaScript 渲染,HTML 源码里啥都拿不到,只能靠 selenium 开浏览器硬等,慢得让人想砸电脑。后来发现了 Playwright,它比 selenium 轻、比 puppeteer 好上手,尤其是它的page.on("response")事件监听机制,配合page.route()拦截请求,可以直接从网络层抓数据,跳过页面渲染的逻辑,这思路后来成了我处理动态页面的主力方案。
这个项目的核心就是干这件事:用 Python 写一个 Playwright 爬虫脚本,监听目标页面发出的所有网络请求,从响应数据里直接提取笔记内容,然后配合自动滚动操作来解决无限滚动页面的翻页问题。目标对象是类似小红薯这种信息流产品——笔记列表不断往下刷、内容不断追加、URL 不变但数据在持续加载。传统做法是通过解析 DOM 节点来抓取,但无限滚动场景下,DOM 会越滚越深,元素定位越来越慢,还会漏数据。监听网络流的思路等于绕开了 DOM,直接从数据源头拿东西,效率高一个量级。
1.2 适合谁看,能学到什么
如果你是个 Python 爬虫新手,刚学会 requests 加 BeautifulSoup,正卡在动态页面这道坎上,这个项目能给你一个很直观的进阶路径:Playwright 的基本用法、监听网络请求的原理、无限滚动页面的处理策略。如果你已经写过一些 selenium 脚本,但对 Playwright 的新特性不太熟悉,这篇文章也能帮你快速补齐相关概念。
整个项目的代码量其实不大,核心逻辑拆开看就三块:启动浏览器并开启网络监听、自动滚动页面触发数据加载、从响应内容中筛选目标数据。我会把每一步的代码都写出来,把关键参数的含义和排查问题的思路也带上。
2. 技术方案选型:为什么是 Playwright 而不是 selenium 或 requests
2.1 从 requests 到 Playwright 的必然转变
先说清楚一个问题:为什么不能直接用 requests?
因为目标页面是动态渲染的,笔记内容的 HTML 结构是 JavaScript 在浏览器里拼出来的。requests 拿到的只是空壳 HTML,里面除了一个 id 为root的空 div 和一堆静态资源链接,什么都没有。如果非要强行解析,只能拿到一些页面框架信息,核心数据完全丢了。
那用 selenium 行不行?能力上没问题,浏览器自动化这套它做得最早,生态成熟。但 selenium 的问题是运行效率低,启动一个浏览器实例就要好几秒,再加上等待页面元素渲染、定位父节点、遍历子节点,一套流程跑下来又慢又容易因为页面结构变动而出错。Playwright 作为后起之秀,设计上更现代化,API 更简洁,最关键的差异在于它支持 network interception(网络拦截)能力,允许直接监听和读取浏览器发出的每个请求与响应的内容。
这个能力对爬虫来说意味着什么?意味着你可以不关心页面上长了什么元素、元素怎么嵌套、类名怎么变化,你只需要关心目标数据是通过哪个接口返回的,直接在那个接口的响应里拿 JSON 数据就行。接口返回的 JSON 通常就是结构化的结构化数据,字段清晰、类型明确,解析起来比解析 DOM 高大上得多。
2.2 监听网络流 vs 解析 DOM 的加载差异
模拟一下两种方案在无限滚动场景下的表现。
解析 DOM 的方案流程大概是:打开页面 -> 等第一屏元素加载完 -> 滚动到底部 -> 等新元素出现 -> 继续滚动 -> 提取全部节点 -> 从节点里挖数据。这个流程有几个痛点:第一,新元素出现是异步的,你怎么知道该等多久?第二,页面滚动一定次数之后,前面的 DOM 节点会被浏览器回收,你得有个机制存住已经抓到的数据,否则就丢了。第三,每个笔记卡片的结构可能不一样,图文和视频的 DOM 结构不同,你得写多套解析逻辑。
监听网络流的方案流程大概是:打开页面 -> 开始监听所有响应 -> 触发滚动 -> 拦截到包含目标 API 路径的响应 -> 直接解析响应 JSON -> 追加到数据列表。这个方案的响应是有序的、可记录的、格式稳定的,不需要关心 DOM,不需要处理浏览器元素回收问题,拿到的就是后端交给前端的数据,也就是业务层最原始的数据。
两种方案一对比,效率差异就清楚了。我实际测试下来,在同等网络情况下,监听网络流方案抓取同等数量的笔记,耗时大约只有 DOM 解析方案的三分之一,而且代码量少一半以上。
2.3 Playwright 版本选择和环境准备
建议使用 Playwright 的 Python 版本,安装命令是:
pip install playwright安装完再装一下浏览器内核:
playwright install chromium这里有个常见坑需要提醒:playwright install chromium默认会从微软的 CDN 下载 Chromium 内核,在某些网络环境下速度会很慢。我实测过,一条命令卡了 20 多分钟的情况也有。遇到这种问题,可以设置镜像环境变量解决:
# Linux / macOS export PLAYWRIGHT_DOWNLOAD_HOST=https://npmmirror.com/mirrors/playwright # Windows PowerShell $env:PLAYWRIGHT_DOWNLOAD_HOST="https://npmmirror.com/mirrors/playwright"然后再执行安装命令,速度会快很多。
我习惯在项目目录下建一个虚拟环境来装依赖,避免污染系统环境,这个习惯从写第一行爬虫代码起就没什么问题地沿用至今。
3. 核心实现:监听网络请求的两大核心函数
3.1 page.on("response") 与 page.route() 的区别
Playwright 里有两个容易混淆的能力,page.on("response")和page.route()。
page.on("response")是事件监听机制。浏览器收到服务器返回的每个响应,都会触发这个事件,你可以注册一个回调函数,在这个函数里检查响应的 URL、请求方式、状态码、响应体,然后决定要不要处理。但它不阻止请求的继续,属于"旁路观察"的模式。适合爬虫场景,因为你不需要改变服务质量,只需要偷看数据。
page.route()是请求拦截机制。它能在请求发出前拦下来,你可以选择放行、修改请求头、修改请求体,甚至直接代替服务器返回一个伪造的响应。这个能力适合反爬对抗场景,比如你想绕过某个网站的请求签名校验,可以在拦截点补充加密参数;也可以用来屏蔽页面上的图片、视频、广告请求,加速页面加载。
本项目用的主要是page.on("response"),因为我们的目标只是拿数据,不需要修改任何东西。但在实际项目里两者经常会组合使用:用route()屏蔽掉字体、图片和统计类请求,减轻页面加载压力;用response事件监听目标 API 的返回。
from playwright.sync_api import sync_playwright def main(): with sync_playwright() as p: browser = p.chromium.launch(headless=False) context = browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" ) page = context.new_page() def on_response(response): url = response.url if "/api/sns/web/v1/feed" in url: try: data = response.json() print(f"捕获到接口: {url}") print(f"接口状态码: {response.status}") except Exception as e: print(f"解析响应失败: {e}") page.on("response", on_response) page.goto("https://example.com") page.wait_for_timeout(3000) browser.close()这段代码里有一个关键点需要展开说明。response.json()是 Playwright 提供的方法,能直接拿到响应体并解析成 JSON。但这里有个限制:如果响应体已经被重定向或已经被 JavaScript 读取过了,可能拿不到内容。原因在于浏览器里的网络缓冲机制只保留一份响应体,多个读取者会争抢。所以实际写爬虫时,我会在回调函数里立刻把响应内容转成字符串存起来,绝不在后面异步处理。
3.2 响应回调中的数据分析:从 URL 中筛选目标接口
在一个真实项目里,页面会发起几十个甚至上百个网络请求:HTML 文档、CSS 样式、JavaScript 脚本、图片、字体、JSON 接口、埋点统计等等。on_response会收到所有这些请求的响应,但我们需要的数据只在其中几个特定的 API 接口里。
筛选的核心依据是接口 URL 的特征。以小红书为例,首页笔记流对应的接口路径通常包含类似api/sns/web/v1/feed的结构。我处理目标页面时,一般会先浏览一遍网络面板,找出哪个请求返回的 JSON 里包含笔记数据(标题、作者、点赞数、封面图 URL 等),然后记住这个 URL 的关键片段,作为后续的筛选字符串。
筛选的方式有两种:
- 前缀匹配:
url.startswith("https://api.example.com/feed") - 子串匹配:
"/api/feed" in url
更精确的做法是在网络面板里观察 Query 参数。有些接口的路由相同,但通过不同的参数区分场景,比如note_type=video是视频流,note_type=normal是图文流。这种情况下,建议把筛选条件写得细一些。
def on_response(response): url = response.url # 只处理 GET 请求,且 URL 包含 feed 接口路径 if "/api/sns/web/v1/feed" in url and response.request.method == "GET": if "note_type=video" in url: print(f"这是视频流接口,跳过: {url}") return try: json_data = response.json() # 后续处理 except Exception: pass有个细节值得强调:这里的response.request.method可以帮你判断接口的请求方式。有些接口 GET 和 POST 共用同一个路径,响应内容不同,光看 URL 判断不准确。
3.3 处理骨架屏和异步请求延迟
无限滚动页面有一个特性:滚动到底部之后,页面并不会立即发送网络请求。真实产品是为了防止用户快速滚动时触发过多请求,一般会做 debounce(防抖)处理,即停止滚动 200-500 毫秒之后才发请求。所以爬虫脚本不能滚一下就立即去找新响应,必须留出等待时间。
我的做法是在每次滚动后固定等待 1-2 秒,这个时间足够接口从发起到响应完成。如果目标接口响应比较慢,可以适当延长等待时间,或者用page.wait_for_timeout()实现一个简单的轮询机制:
import time for scroll_round in range(10): # 滚动到底部 page.mouse.wheel(0, 3000) # 等待网络请求完成 page.wait_for_timeout(1500) # 检查当前已抓取的数据条数 if len(notes) > target_count: break关于page.wait_for_timeout()和time.sleep()的区别,wait_for_timeout是 Playwright 提供的等待方法,在异步运行时会自动推进事件循环,让其余的页面事件得以继续处理。用time.sleep()会直接阻塞整个线程,有时候页面事件没法正常触发,导致滚动后网络请求异常。所以我强烈建议用wait_for_timeout,这也是我踩坑踩出来的经验。
4. 无限滚动页面的完整爬取方案
4.1 模拟滚动方式:直接执行 JS 还是 wheel 事件
Playwright 提供了多种模拟滚动的方式,最常用的有三种。
第一种是page.mouse.wheel(0, delta_y),模拟鼠标滚轮滚动,触发浏览器默认的滚动行为。这种方式最接近真实用滚动操作,推荐优先使用。但缺点是滚动速度不可控,如果滚动太快,有些防滚动机制严格的产品会触发风控。
第二种是直接执行 JavaScript:
page.evaluate("window.scrollTo(0, document.body.scrollHeight)")这种方式一步到位,效率最高。有些产品会检测navigator.webdriver属性或滚动行为特征,直接执行 JS 可能更容易触发反爬标记。但灵敏的产品会判断滚动是否匀速、是否有停顿,一般而言 wheel 事件更安全。
第三种是用键盘操作:
page.keyboard.press("End")按下 End 键跳到页面底部,实际效果和滚动类似。我很少用这种方式,因为 End 键触发的是快捷键行为,某些 Web 应用会拦截这个按键并禁止默认行为。
我实际测试下来,用page.mouse.wheel(0, 1200)这种固定步长的滚动方式最稳定。每次滚动 1200 像素,休息 1.5 秒,再滚动。这种做法既不会太快触发风控,也不会太慢影响效率。如果页面里有多列瀑布流布局,滚动量可以稍微调大一些。
4.2 滚动停止条件:条数判断与页面高度变化
无限滚动页面的一个难点在于,你根本不知道页面什么时候到底了。有些产品会设置"已经到底了"的提示,有些产品是永远滚不完的推荐流。所以必须设计一个有效的停止条件。
我通常用两种策略结合:
第一个策略是数量阈值判断。先估算一下本次任务要抓多少条数据,抓够就停。这种方式最直接,也不用额外增加逻辑。
if len(notes) >= target_count: print(f"已抓满 {target_count} 条笔记,停止滚动") break第二个策略是页面高度变化判断。如果持续滚动但页面高度不再变化,说明已经到达底部:
last_height = page.evaluate("document.body.scrollHeight") current_height = 0 max_no_successive = 3 no_change_count = 0 while True: page.mouse.wheel(0, 1200) page.wait_for_timeout(1500) current_height = page.evaluate("document.body.scrollHeight") if current_height == last_height: no_change_count += 1 if no_change_count >= 3: print("页面高度已无变化,判定到达底部") break else: no_change_count = 0 last_height = current_height判断"页面高度是否变化"本质上是在判断"新内容有没有加载出来"。如果连续三次滚动后高度都不变化,大概率是到底了,也可能是被风控了。这种情况我会再单独处理。
4.3 目标数据提取:从响应 JSON 中定位笔记字段
拿到响应 JSON 之后,下一步是找到笔记数据在哪个字段下面。到这一步,你不需要去浏览器里一个个看元素,直接把这个 JSON 打印出来看结构就行。
我一般会先在回调里加一行调试代码:
def on_response(response): url = response.url if "/api/sns/web/v1/feed" in url: data = response.json() print(json.dumps(data, ensure_ascii=False, indent=2)) # 只可能截取前 2000 个字符,避免刷屏 # print(json.dumps(data, ensure_ascii=False, indent=2)[:2000])上面这段代码会把接口返回的完整 JSON 打印出来。典型的返回结构一般是{"code": 0, "data": {"items": [...]}}这种形式,items列表里每个元素就是一条笔记。
接着你需要遍历items,提取每条笔记的关键信息:
def parse_note_data(data): items = data.get("data", {}).get("items", []) for item in items: note_card = item.get("note_card", {}) if not note_card: continue note_data = { "note_id": note_card.get("note_id"), "title": note_card.get("title"), "author": note_card.get("user", {}).get("nickname"), "liked_count": note_card.get("interact_info", {}).get("liked_count"), "cover_url": note_card.get("cover", {}).get("url"), } notes.append(note_data)这里有个重要提醒:笔记的字段结构会因为内容类型不同而变化。图文笔记里有image_list,视频笔记里有video字段,广告可能会有ad类型,所以解析时一定要做好类型判断和字段兜底。不要硬编码item["note_card"]["title"],因为某些 item 可能没有note_card或者note_card里没有title。用.get()方法配合默认值是最稳妥的。
4.4 完整代码:无限滚动笔记爬取脚本
把上面的逻辑整合一下,就是一个可以直接运行的完整脚本。这里我给出一个可以在此基础上修改的版本:
import json from playwright.sync_api import sync_playwright def on_response(response, notes): url = response.url if "/api/sns/web/v1/feed" in url and response.request.method == "GET": try: data = response.json() parse_note_data(data, notes) except Exception as e: print(f"接口解析失败: {url}, 错误: {e}") def parse_note_data(data, notes): items = data.get("data", {}).get("items", []) for item in items: note_card = item.get("note_card", {}) if not note_card: continue note_data = { "note_id": note_card.get("note_id"), "title": note_card.get("title"), "author": note_card.get("user", {}).get("nickname"), "liked_count": note_card.get("interact_info", {}).get("liked_count"), "cover_url": note_card.get("cover", {}).get("url"), } notes.append(note_data) print(f"捕获笔记: {note_data['title']}") def main(): notes = [] target_count = 200 with sync_playwright() as p: browser = p.chromium.launch( headless=False, args=["--disable-blink-features=AutomationControlled"] ) context = browser.new_context( viewport={"width": 1280, "height": 800}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" ) page = context.new_page() page.on("response", lambda response: on_response(response, notes)) page.goto("https://example.com/explore", wait_until="networkidle") page.wait_for_timeout(2000) last_height = 0 no_change_count = 0 while len(notes) < target_count: page.mouse.wheel(0, 1200) page.wait_for_timeout(1500) current_height = page.evaluate("document.body.scrollHeight") if current_height == last_height: no_change_count += 1 if no_change_count >= 3: print("页面高度不再变化,停止滚动") break else: no_change_count = 0 last_height = current_height print(f"已抓取 {len(notes)} 条笔记") browser.close() print(f"总共抓取到 {len(notes)} 条笔记") # 保存到本地 JSON with open("notes.json", "w", encoding="utf-8") as f: json.dump(notes, f, ensure_ascii=False, indent=2) if __name__ == "__main__": main()有一个细节值得单独说明:浏览器启动参数里加了--disable-blink-features=AutomationControlled,这个参数可以移除 Chromium 的自动化控制标记,让navigator.webdriver返回 false,在一定程度上降低被识别为自动化工具的概率。但需要明确,这只是对抗爬虫的基本操作,不是万能的,很多产品还有更严格的指纹检测机制。
5. 数据存储与工程化扩展
5.1 SQLAlchemy 存储爬虫数据
上面代码里最后的存储方式是写到本地 JSON 文件,适合小规模测试和临时跑任务。但如果要批量爬取、定期爬取,数据量上去之后,JSON 文件的管理会变得很混乱。这时候就该上数据库了。
用 SQLAlchemy 做存储是当前主流做法。SQLAlchemy 是 Python 里最成熟的 ORM(对象关系映射)框架,它把 Python 对象和数据库表结构对应起来,写代码时不用手写 SQL 语句。
先定义一个数据表模型:
from sqlalchemy import create_engine, Column, String, Integer, Text from sqlalchemy.orm import declarative_base, sessionmaker Base = declarative_base() class NoteRecord(Base): __tablename__ = "notes" id = Column(Integer, primary_key=True, autoincrement=True) note_id = Column(String(64), unique=True, index=True) title = Column(String(255)) author = Column(String(255)) liked_count = Column(String(64)) cover_url = Column(Text)然后创建一个数据库会话:
engine = create_engine("sqlite:///notes.db", echo=False) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session()在爬虫回调里把数据插入数据库:
def save_to_db(notes_list): for note in notes_list: existing = session.query(NoteRecord).filter_by(note_id=note["note_id"]).first() if existing: continue record = NoteRecord( note_id=note["note_id"], title=note["title"], author=note["author"], liked_count=note["liked_count"], cover_url=note["cover_url"] ) session.add(record) session.commit()unique=True这个约束非常关键。由于无限滚动过程中可能同一个接口被多次命中、同一批数据被重复解析到,如果不做去重,数据库里会塞满重复记录。除了数据库层面的唯一约束,我在代码里也做了判断:先查询有没有已存在的note_id,存在就跳过。这种双保险在实际项目中很管用。
如果要接入 MySQL 或 PostgreSQL,只需要改一行连接字符串:
engine = create_engine("mysql+pymysql://user:password@localhost/namedb?charset=utf8mb4") engine = create_engine("postgresql://user:password@localhost/namedb")5.2 断点续爬与去重机制
爬虫跑了一半突然中断,或者跑着跑着因为网络波动导致部分接口解析失败,这种事太常见了。真实项目中几乎不可能一次跑完,所以断点续爬是刚需。
断点续爬的核心逻辑是:每次启动时,先从数据库查出已经抓过的note_id集合,在解析新数据时跳过这些已存在的记录。
def get_existing_ids(session): return set(row[0] for row in session.query(NoteRecord.note_id).all())启动爬虫时加载这个集合:
existing_ids = get_existing_ids(session) def is_duplicate(note_id): return note_id in existing_ids在parse_note_data里做判断:
if note.card.get("note_id") in existing_ids: continue爬虫断掉之后,重新运行脚本就能接着上次的进度继续抓,不用从头开始。
5.3 异步并发优化
Playwright 的 sync API 写法直观、容易调试,适合快速开发和爬取任务量不大的场景。但如果要大规模爬取,sync API 的性能瓶颈就很明显了——浏览器是单线程阻塞式的,等待一个请求的时候,其他请求都得排队。
Playwright 提供了 async API,配合 Python 的asyncio可以实现多页面并发。思路是:同时打开多个页面,每个页面负责不同的滚动区域或不同的分类页签,通过asyncio.gather()并发调度。
import asyncio from playwright.async_api import async_playwright async def crawl_page(browser, url, results): page = await browser.new_page() await page.goto(url) notes = [] def on_response(response): if "/api/sns/web/v1/feed" in response.url: try: data = response.json() # 解析 except Exception: pass page.on("response", on_response) for _ in range(10): await page.mouse.wheel(0, 1200) await page.wait_for_timeout(1500) results.extend(notes) await page.close() async def main(): urls = ["https://example.com/explore/tab1", "https://example.com/explore/tab2"] results = [] async with async_playwright() as p: browser = await p.chromium.launch(headless=True) tasks = [crawl_page(browser, url, results) for url in urls] await asyncio.gather(*tasks) await browser.close() print(f"共抓到 {len(results)} 条") asyncio.run(main())并发数量不建议一次开太多页面,先开 3-5 个页面比较合适。开太多会带来两个问题:一是目标网站的反爬策略容易触发,IP 被临时限制;二是本机资源占用过大,浏览器渲染多个页面会吃掉大量 CPU 和内存,反而拖慢速度。
6. 遇到过的实际问题与排查技巧
6.1 接口响应拿不到,response.json()返回空
我最早做这个项目时,遇到过一个问题:日志里能看到响应事件被触发了,URL 也对,但response.json()返回异常,或者干脆抛异常。研究后才发现,浏览器只允许响应体被消费一次。也就是说,如果页面自身的 JavaScript 已经调用过response.text()或response.json(),那么你再调用就会失败,因为响应体已经被消耗了。
要解决这个问题,可以在回调函数里立即调用response.body()并保存内容:
def on_response(response): if "/api/sns/web/v1/feed" in response.url: try: body = response.body() data = json.loads(body.decode("utf-8")) # 后续解析 except Exception as e: print(f"读取响应失败: {e}")注意response.body()返回的是 bytes,需要先解码再json.loads。
还有一种情况是响应被 service worker 缓存了,实际数据是从缓存里取的,没有经过网络层。这种情况比较少见,但遇到了可以尝试强制禁用 service worker:
context = browser.new_context( service_workers="block" # 或 "allow" )6.2 页面滚动之后没有新请求发出
页面滚动后接口没反应,最可能的原因是滚动位置不对。
有些页面的滚动容器不是document.body,而是页面内部的某个 div。你用window.scrollTo()或mouse.wheel()滚动的是外层滚动容器,内层容器没有滚动,自然不触发加载逻辑。
排查方法:在浏览器开发者工具里查看目标区域的 CSSoverflow属性,找到真正的滚动容器。然后针对性执行滚动:
# 假设滚动容器是 .feeds-container page.evaluate("""() => { const container = document.querySelector('.feeds-container'); container.scrollTo(0, container.scrollHeight); }""")用mouse.wheel滚动时,光标位置也很关键。如果光标不在滚动容器上方,wheel 事件就不会作用到目标容器上。所以滚轮操作之前,先确保把鼠标移动到容器区域内:
page.mouse.move(640, 400) page.mouse.wheel(0, 1200)6.3 无限滚动触发滑块验证
这个是动态页面爬虫里绕不开的问题,尤其是信息流产品。频繁滚动、快速滚动、多次刷新同一页面,都可能触发验证码。
我的经验是:控制滚动频率和总量。每次滚动 1200 像素后等待 2-3 秒,不要连续快速滚动;单次任务设置数据条数上限,不要一直滚到天荒地老;每次浏览器会话不要超过 10 分钟。另外,可以给浏览器注入更真实的用户代理和视图尺寸:
context = browser.new_context( viewport={"width": 1366, "height": 768}, user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" )如果还是被验证码弹窗挡住了,可以试试context.storage_state()的复用功能。先用真实浏览器登录一次,保存登录状态:
# 第一次运行:手动扫码登录 context = browser.new_context() page = context.new_page() page.goto("https://example.com/login") input("登录完成后回车继续...") context.storage_state(path="state.json") # 第二次运行:加载登录状态 context = browser.new_context( storage_state="state.json" )这种方式能大幅降低被触发验证码的概率,因为你在目标产品看来是一个已登录、有状态的"正常用户"。
6.4 页面元素不加载、一直转圈
打开页面后,页面白屏或者一直转圈,首要排查目标是网络请求被阻塞了。可能是页面引用的某个 CDN 资源被拦截,也可能是 JS 报错导致整个页面没有正常渲染。
Playwright 启动时可以监听控制台消息和页面错误:
page.on("console", lambda msg: print(f"浏览器日志: {msg.text}")) page.on("pageerror", lambda err: print(f"页面错误: {err}"))看看有没有关键的 JS 报错,比如跨域问题、某个接口 403 或 500。把浏览器打开窗口模式(headless=False),肉眼看着页面加载过程,也是一种很直接的排查方法。
启动参数里的wait_until="networkidle"是等到网络空闲再返回。但如果页面里有持续的长连接请求或者轮询接口,networkidle可能永远等不到,导致goto一直挂起。这时候可以改成wait_until="domcontentloaded",快速返回,然后再额外wait_for_timeout等待页面数据渲染:
page.goto(url, wait_until="domcontentloaded") page.wait_for_timeout(3000)这个方法我在很多实际项目中都用过,效率比networkidle高得多。
7. 爬虫代码的模块化组织
7.1 把单文件脚本拆成可维护的模块
我见过太多爬虫项目,所有人都把代码堆在一个文件里,从文件入口到网络监听、数据解析、数据库存储、日志打印,全部塞在一起。前期能跑就是胜利,后期想改一个字段名都要摸半天。
爬虫代码天然适合按职责拆模块。一个稍微正式一点的结构可以参考:
spider/ ├── config.py # 配置类:目标 URL、抓取条数、滚动间隔 ├── database.py # 数据库连接和 Session 管理 ├── models.py # SQLAlchemy 模型定义 ├── parser.py # 接口响应解析函数 ├── spider.py # Playwright 爬虫主逻辑 └── run.py # 入口文件职责拆清楚的好处是:换目标页面时,只需要改config.py里的 URL 和parser.py里的解析函数;换数据库只要改database.py;调整滚动策略只动spider.py。不同版本之间也能往 git 上打 tag,调度跑起来更可控。
7.2 配置文件的参数化设计
不要在代码里硬编码 UA、滚动间隔、浏览器启动参数这些东西。后期你会发现,不同目标站点对 UA、视图尺寸、滚动节奏的要求差异很大,参数化是最合理的方案。
我常用的配置类大概是这个风格:
class SpiderConfig: def __init__(self): self.target_url = "https://example.com/explore" self.target_api_pattern = "/api/sns/web/v1/feed" self.scroll_pixel = 1200 self.scroll_interval = 1.5 # 秒 self.max_no_change_count = 3 self.target_count = 200 self.headless = False self.save_to_db = True self.db_url = "sqlite:///notes.db" self.user_agent = "Mozilla/5.0 ..."用SpiderConfig().scroll_pixel而不是到处写1200,改动时只需要打开一个文件。配置文件里还可以加一些业务相关选项,比如是否跳过视频笔记、是否只抓带有图片的笔记,扩展起来很方便。
7.3 日志系统的重要性
爬虫跑太久,控制台输出会被冲掉,前面跑的信息就丢了。要追踪每个接口的响应状态、每条笔记的解析结果、每次滚动的效果,就得用一个规范的日志系统。
Python 标准库的logging模块足够用。简单配置:
import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("spider.log", encoding="utf-8"), logging.StreamHandler() ] ) logger = logging.getLogger("spider") logger.info("爬虫启动,目标: %s", config.target_url) logger.error("接口解析失败: %s", url)日志文件里记录下每次运行的启动时间、抓取条数、异常堆栈,后期排查问题时,不用重新跑一遍脚本,直接翻日志就能定位。
8. 最终成品演示与进一步扩展
8.1 实测运行效果
我自己用这个方案完整跑过一次,目标是一个聚合笔记流的页面,设置了 200 条的抓取上限。实际运行情况是这样的:浏览器窗口弹出后,页面自动滚动,日志面板逐条打印接口捕获信息,大概 4 分半钟跑完 200 条,数据库里存了 200 条去重后的记录,没有出现重复。脚本停止后,我看了一眼运营后台的数据分布,标题、点赞数、封面图 URL 这些字段都准确对上了,说明接口解析逻辑没有偏差。
跟之前的 DOM 解析方案对比,同规模数据量少跑了大概 1 分半钟,少写了差不多 60 行解析代码,而且数据类型的准确性高了一截——毕竟后端 JSON 里字段本来就是结构化好的,不需要从 HTML 文本里正则抠。
8.2 扩展到关键词搜索、用户主页等场景
这种"监听网络流 + 触发页面事件"的思路,换到不同的业务场景时只需改动两块地方:入口 URL 和接口筛选条件。
比如你要抓取某个关键词的全部笔记,入口是搜索结果页面:
search_url = f"https://example.com/search_result?keyword={keyword}&scope=all"接口筛选条件可能变成"/api/sns/web/v1/search"。
要抓取某用户主页笔记,入口改成用户主页 URL,接口可能就变成"/api/sns/web/v1/user_posted"。解析函数里的字段层级也可能不同,比如用户笔记接口返回的数据里没有note_card外壳,直接就是notes列表。
因为大框架没变,这类扩展通常只需要改动十几行代码,就能复用到新场景。这种模式化的复用能力,正是监听网络流方案最大的价值。
8.3 遵守规则与频率控制
关于爬虫这件事,最后必须做个提醒。我所有实践的前提是尊重目标网站的访问规则和数据使用条款,控制请求频率,避免对目标服务造成压力。我在脚本里故意设置的开页延迟、滚动间隔、单次抓取条数上限,不只是为了降低触雷概率,更是为了做一个负责任的爬虫工程师。
个人体验来看,写爬虫从来不是一个一劳永逸的事。页面结构会变、接口字段会变、反爬机制也在变。学会监听网络流这个思路之后,最大的收获其实是:遇到动态页面不再害怕,因为你知道不论前端怎么渲染,后端接口是绕不开的,而 Playwright 给了你一把直接访问接口数据的钥匙,剩下的只是细心和耐心。