做爬虫的人十有八九都经历过这种时刻页面结构一改辛辛苦苦写好的选择器全部失效或者页面滚动半天数据始终缺几条。后来我换了思路不再盯着 DOM 看而是直接监听浏览器网络层返回的 XHR/JSON 响应。这一换稳定性和效率都上了一个台阶。这套方法的核心就是 Python Playwright 的 Response 监听。今天这篇东西我会从环境搭建讲到精准捕获接口数据再到 CSV 导出和 SQLite 持久化存储把整套网络层采集方案掰开揉碎。适合正在用 Selenium 或 Playwright 解析动态页面、但总被 DOM 变化折磨的朋友也适合想系统掌握接口级采集思路的爬虫初学者。1. 为什么优先选网络层采集从 DOM 解析到 Response 监听1.1 传统 DOM 解析的痛点页面一变代码就挂以前写爬虫最常见的路径是打开浏览器开发者工具查看元素定位到要抓的数据节点然后写 selector。这套思路对付静态页面没问题但碰到现代前端应用麻烦事一大堆。最大的问题是异步渲染时序。现在大多数页面都是先加载 HTML 骨架再通过 XHR 请求拿数据最后用 JavaScript 把数据渲染到页面上。这就导致你根本不知道数据什么时候出现在 DOM 里。很多人的第一反应是time.sleep(3)可网络慢的时候 3 秒不够网络快的时候 3 秒又白白浪费时间。你调大了超时时间脚本整体速度就被拖慢你调小了吧又经常在凌晨跑任务时抽风。第二个痛点是 DOM 结构极其脆弱。前端工程师改个类名、换个标签层级甚至只是给某个区块加了一个 wrapper你的选择器就彻底失效。我遇到过最离谱的一次页面只是把div换成了span代码就崩了排查了半天才发现是这种低级变化。虚拟滚动和懒加载也是重灾区。列表页为了性能只渲染可视区域的节点你必须要模拟滚动、等待图片加载、再滚动才能把所有数据逼出来。数据量一大这套滚动触发-等待渲染-抓取节点的循环不仅慢还容易漏。更致命的一个问题是有些数据根本不在 DOM 里。图表类页面用 Canvas 绘制表格组件的数据只存在于前端状态对象中或者接口返回了 100 个字段页面只展示了 5 个。你想抓那 95 个隐藏字段靠 DOM 解析根本拿不到。1.2 Response 监听的原理直接在网络层拦截数据Playwright 的 Response 监听解决的就是上面这几类问题。它的思路非常直接浏览器发出去的每一个网络请求在收到响应时都会触发response事件。你在事件回调里拿到的是一个Response对象里面有 URL、状态码、响应头最重要的是可以通过json()或body()直接拿到接口返回的原始数据。理解这件事的关键是要区分两个层面。DOM 解析看的是渲染之后的结果相当于你站在商店外面根据橱窗陈列去猜仓库里有什么Response 监听看的是网络传输中的数据相当于你直接跑到仓库门口看发货单原件。仓库里的货可能被陈列方式影响但发货单上的条目是稳定的。技术上Playwright 通过浏览器底层的调试协议把每个网络事件实时推送给 Python 回调。你在代码里注册一个监听函数后续任何满足条件的响应都会经过这个函数。这样做有几个天然优势数据结构稳定。接口的字段名一般不会频繁变化比 className 靠谱得多。拿到的是完整原始数据。无论页面展示不展示接口返回了多少字段你都能拿到。不依赖渲染时序。数据在网络层返回的那一刻就能捕获不需要等待 DOM 更新。性能开销小。监听网络事件比反复查询 DOM 节点轻量得多。1.3 我的设计思路回调只收集主流程处理用 Playwright 写采集脚本最重要的一个设计原则是监听回调里永远只做最轻量的事情。原因很简单Python 回调函数运行在 Playwright 的消息循环线程上。如果你在回调里做耗时操作比如写文件、请求数据库、处理大 JSON会阻塞整个浏览器事件循环导致页面响应变慢甚至影响后续请求的触发。我踩过这个坑第一次写的时候直接在回调里调 SQLite 插入结果页面加载慢了两倍还偶发超时。正确做法是让回调只负责把数据丢进一个队列或者列表然后立刻返回。主流程继续驱动页面操作等操作完成后再集中处理队列里的数据。这个思路贯穿本文所有示例代码你在后面的 CSV 和 SQLite 部分会看到具体落地方式。2. 环境准备与最小监听工程2.1 环境要求与依赖安装这套方案需要 Python 3.8 以上我自己习惯用 3.10 或更高版本类型注解和语法支持更舒服。Playwright 的 Python 库安装很简单但要记得它和 Node.js 版不是一个包。pip install playwright装完库还不够必须安装浏览器内核。Playwright 不是调你电脑上现有的 Chrome而是使用自己维护的浏览器副本。playwright install chromium这一步在国内经常因为网络原因失败。如果遇到下载超时可以设置镜像环境变量再重试或者直接下载后指定浏览器路径用executable_path参数。但优先建议把官方安装流程跑通后面升级维护省心。注意playwright install会下载约 150MB 左右的浏览器文件需要一定磁盘空间。如果服务器环境缺少系统依赖库启动时可能会报一堆.so找不到的错误这时候看错误信息逐个用系统包管理器补上即可。2.2 浏览器启动与基础配置我用 Python 的同步 API 演示因为它简单直接适合大多数采集脚本。启动浏览器和创建页面是固定套路from playwright.sync_api import sync_playwright with sync_playwright() as p: browser p.chromium.launch( headlessTrue, args[--disable-blink-featuresAutomationControlled] ) context browser.new_context( viewport{width: 1280, height: 800}, localezh-CN, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36 ) page context.new_page()几个参数值得解释一下。headlessTrue表示无头模式服务器上采集时看不到界面但跑得快。--disable-blink-featuresAutomationControlled可以去掉一部分自动化特征标记这个参数在遇到基础风控的站点时有用但别指望靠它应对所有防护。locale和user_agent是让浏览器环境更贴近真实用户。2.3 最小可运行的监听示例打印所有 JSON 响应先来一个最朴素的版本感受一下事件驱动的流程from playwright.sync_api import sync_playwright def on_response(response): content_type response.headers.get(content-type, ) if json not in content_type: return if response.status 400: return try: data response.json() except Exception: return print(response.url) print(type(data), len(data) if isinstance(data, (list, dict)) else unknown) with sync_playwright() as p: browser p.chromium.launch(headlessTrue) page browser.new_page() page.on(response, on_response) page.goto(https://example.com/list) page.wait_for_timeout(5000) browser.close()这里page.on(response, on_response)的作用是注册监听器。只要页面后续发起的每个请求收到响应on_response就会被调用一次。我们通过content-type头过滤出 JSON 类型再打印 URL 和数据结构。这个示例跑起来后你会发现一个页面可能触发几十个 JSON 响应里面混杂着统计上报、资源配置、推荐接口等各种东西。下一个阶段要做的事情就是精准筛选出真正包含业务数据的接口。3. 核心实现精准捕获 XHR/JSON 并同步等待3.1 请求类型识别与 URL 过滤策略打印所有 JSON 响应只是第一步真实场景里我们需要的是某个或某几个接口的数据。筛选条件一般有两类请求类型和 URL 特征。每个 Response 对象都可以通过response.request.resource_type查看请求类型常见取值有resource_type含义是否通常需要采集document页面本身的 HTML 文档否xhr传统的 XMLHttpRequest 异步请求是fetchFetch API 发起的请求是scriptJavaScript 文件否stylesheetCSS 样式文件否image图片资源否很多现代前端会用 Fetch API所以判断条件一般是这两种def is_target_response(response): if response.status ! 200: return False content_type response.headers.get(content-type, ) if json not in content_type: return False if response.request.resource_type not in (xhr, fetch): return False if /api/ not in response.url: return False return TrueURL 过滤是更灵活的方式。不同站点的 API 路径风格不同有的带/api/有的带/v1/有的是/gateway/。把这些特征配成一个列表再写一个匹配函数API_KEYWORDS [/api/, /v1/, /gateway/, /list] def is_target_url(url): return any(keyword in url for keyword in API_KEYWORDS)建议先跑一遍不过滤的监听把页面所有 XHR/Fetch 请求的 URL 打出来肉眼确认目标接口的特征再收紧过滤条件。这是排查漏数据最有效的方法。3.2 响应体解析与数据清洗拿到 Response 对象后读取数据有两种方式。response.json()直接返回解析后的 Python 对象方便但要求响应头里的 content-type 包含 json且响应体必须是合法 JSON。有时候接口的 content-type 是text/plain但返回内容实际是 JSON这时response.json()会直接抛异常。更稳妥的方式是先拿原始字节再手动解码def safe_json(response): try: return response.json() except Exception: body response.body() text body.decode(utf-8, errorsreplace) try: import json return json.loads(text) except Exception: return None拿到 JSON 之后下一步是数据清洗。接口返回结构五花八门常见的有三种最外层是列表直接就是数据数组。最外层是字典数据在data字段里。分页结构数据在data.list或data.records里旁边还有total、page等元信息。可以先写一个通用提取函数def extract_records(payload): if isinstance(payload, list): return payload if isinstance(payload, dict): data payload.get(data) if isinstance(data, list): return data if isinstance(data, dict): for key in (list, records, items, rows): if isinstance(data.get(key), list): return data[key] return []在浏览器开发者工具里先用 Network 面板看一遍接口返回确认数据结构层级再决定extract_records怎么写。这个习惯能帮你节省大量调试时间。3.3 同步等待机制用 expect_response 精准控制时序事件监听适合持续采集场景但更多时候我们需要的是点击按钮后等待某个特定接口返回。这种情况下用page.expect_response()更优雅。expect_response()会在进入with块时注册一个临时监听器然后执行某段触发代码再等待符合条件的响应出现with page.expect_response( lambda resp: /api/list in resp.url and resp.request.resource_type xhr ) as resp_info: page.click(#load-more-btn) response resp_info.value data response.json() print(data)这里有几个很关键的细节。第一匹配函数必须是同步函数返回 True 表示命中。不要在里面写耗时逻辑。第二with块里的触发操作要写在监听的射程范围内。如果点击之前有很长一段其他操作那个目标接口在with之前就已经返回了你就会等不到然后超时。所以我一般把要触发的动作和等待尽量靠在一起。第三默认超时时间是 30 秒可以在expect_response里通过timeout10000调整。如果目标接口在超时时间内没出现会抛出TimeoutError。这时候不要盲目加时间先确认接口是否存在、触发动作是否正确。对比一下page.on(response)更像全局雷达适合持续收集page.expect_response()更像精确制导适合单次操作后的精准等待。两者可以配合使用。4. 数据落地CSV 导出与 SQLite 持久化4.1 数据模型设计先想清楚要存什么很多初学者一上来就写 CSV 列写到一半发现字段不够用了。我的建议是先定一个统一的数据模型每个捕获的响应至少包含元信息层和业务数据层capture_time捕获时间用于调试和追溯。url接口地址出现多个接口时可以区分来源。method请求方法GET 还是 POST。statusHTTP 状态码。api_name从 URL 中提取的接口标识。raw_json原始响应体转成 JSON 字符串保存。为什么一定要保留raw_json因为你当时提取的字段可能只是暂时需要的过几天产品要另一个字段你就得重新跑一遍采集。有了原始响应体直接从库里捞出来重新解析即可。我用一个简单的字典作为行数据def build_row(response, payload): return { capture_time: datetime.now().isoformat(), url: response.url, method: response.request.method, status: response.status, api_name: extract_api_name(response.url), raw_json: json.dumps(payload, ensure_asciiFalse), }4.2 CSV 导出编码与追加写入的坑CSV 是最通用的交付格式但坑也不少。首先是编码。直接open(path, w)默认是用系统编码写文件Windows 下通常是 GBK用 Excel 打开没问题但用 Python 或 Linux 工具读可能乱码。反过来写成 UTF-8用 Excel 打开又会乱码。解决办法是写 UTF-8 带 BOM也就是utf-8-sig编码。其次是追加写入时的表头问题。文件第一次创建时需要写表头后续追加就不应该重复写。判断方式很简单import csv import os CSV_PATH captured_responses.csv FIELD_NAMES [capture_time, url, method, status, api_name, raw_json] def append_row_to_csv(row): file_exists os.path.exists(CSV_PATH) and os.path.getsize(CSV_PATH) 0 with open(CSV_PATH, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesFIELD_NAMES) if not file_exists: writer.writeheader() writer.writerow(row)newline这个参数很多人会忽略。Windows 下如果不设置CSV 写入时每行之间会多一个空行非常难看。注意CSV 是文本格式不支持嵌套结构所以raw_json只能以 JSON 字符串形式存。字段里如果包含逗号、换行、引号csv模块会自动做转义不用担心破坏格式。4.3 SQLite 持久化建表、参数化与去重数据量一上来CSV 的缺点就暴露了没法按条件查询去重麻烦多进程写入还会冲突。这时候用 SQLite 最省事。它是单文件数据库Python 自带sqlite3模块不需要额外安装服务。建表语句和插入逻辑如下import sqlite3 DB_PATH captured.db def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS captured_responses ( id INTEGER PRIMARY KEY AUTOINCREMENT, capture_time TEXT, url TEXT, method TEXT, status INTEGER, api_name TEXT, raw_json TEXT, UNIQUE(url, capture_time) ) ) conn.commit() conn.close() def insert_rows(rows): conn sqlite3.connect(DB_PATH) conn.executemany( INSERT OR IGNORE INTO captured_responses (capture_time, url, method, status, api_name, raw_json) VALUES (?, ?, ?, ?, ?, ?) , [ ( row[capture_time], row[url], row[method], row[status], row[api_name], row[raw_json], ) for row in rows ]) conn.commit() conn.close()这里有两个重点。第一参数化查询。不要用 f-string 拼 SQL因为 raw_json 里有引号、逗号和各种特殊字符拼进去轻则报语法错误重则有注入风险。用?占位符是零成本的规范行为。第二INSERT OR IGNORE配合UNIQUE(url, capture_time)约束实现去重。同一个接口在同一个时刻只可能返回一次数据重复写入时数据库自动忽略。如果你希望覆盖更新而不是忽略可以改成INSERT OR REPLACE。批量插入一定要用executemany别在 Python 循环里逐条 execute。一个是几百条 commit一个是一万条逐条 commit性能差异非常明显。4.4 采集器主流程监听器 队列 批量落地现在把前面的内容串成一个完整的采集器。基本结构是回调只收集数据主流程在页面操作完成后批量落地。from playwright.sync_api import sync_playwright captured [] def on_response(response): if not is_target_response(response): return payload safe_json(response) if payload is None: return captured.append(build_row(response, payload)) with sync_playwright() as p: browser p.chromium.launch(headlessTrue) page browser.new_page() page.on(response, on_response) page.goto(https://example.com/list) page.wait_for_timeout(3000) # 模拟翻页或滚动操作 for i in range(5): page.mouse.wheel(0, 2000) page.wait_for_timeout(1000) browser.close() # 主流程处理清洗 持久化 init_db() insert_rows(captured) for row in captured: append_row_to_csv(row) print(captured:, len(captured))一个很实用的细节是列表captured直接从回调里追加。Python 的 GIL 保证简单列表追加操作是线程安全的而 Playwright 同步 API 的回调本身也是单线程调用所以这里不会出现并发问题。如果你同时开了多个页面并且每个页面都在同一个 context 里触发回调那captured可能被并发访问建议加一个threading.Lock保护。5. 常见问题与排查技巧实录5.1 死活收不到目标响应先别急着调超时我见过太多人一收不到数据就把wait_for_timeout从 3 秒调到 10 秒结果还是没用。问题往往不在时间而在过滤条件。第一resource_type 判断不准。有的接口虽然是从 JavaScript 里发起的但 resource_type 不是xhr也不是fetch而是other。尤其是某些第三方 SDK 发起的请求。排查方法是临时去掉 resource_type 过滤把所有响应都打出来看目标请求到底属于什么类型。第二接口在 iframe 里。page.on(response)其实能收到 iframe 子框架里发出的请求因为page事件覆盖所有子框架。但如果你用page.expect_response()并且触发代码在 iframe 之外调用可能因为主框架和 iframe 的时序问题漏掉。这时候用context.on(response)挂全局监听更稳妥。第三监听器挂晚了。如果你在page.goto()之后才调用page.on(response, ...)页面加载过程中的第一批接口早就返回了自然收不到。监听器应该在goto之前挂好。新开页面、弹窗页面同理要在创建页面后立刻挂。第四别忘了网络级别的问题接口可能被 Service Worker 拦截并缓存或者浏览器根本没发出请求。这种场景少但遇到诡异的丢数据可以先用context.on(request)确认请求是否真的发出了。5.2 JSON 解析失败编码与响应体异常response.json()不是万能的。最常见的异常有两个一是接口的 content-type 不是 JSON。比如有些老系统返回text/plain但内容就是一个 JSON 数组。response.json()会先检查 content-type不匹配就直接抛错。解决办法就是前面写的safe_json函数捕获异常后手动response.body().decode(utf-8)再json.loads。二是个别接口中文乱码。如果服务端用 GBK 编码返回而代码默认用 UTF-8 解码中文字符就全变成乱码。这时需要指定正确的编码body response.body() text body.decode(utf-8, errorsreplace) if \ufffd in text: # 尝试其他常见编码 text body.decode(gbk, errorsreplace)注意errorsreplace不会让你拿到正确数据但至少不会让解析直接崩掉。定位编码问题最好先看响应头里的charset字段那是最可靠的线索。另一个容易被忽略的问题是响应体被截断。Playwright 的response.body()返回的是完整响应体但如果请求发生重定向、连接中断或者响应超过一定大小可能拿到的 body 不完整。遇到这种情况检查一下len(body)是否和响应头里的 content-length 一致。5.3 页面跳转与多页面监听失效我在实战中遇到过这种情况点击某个按钮后页面用target_blank新开了一个标签页然后在新页面里发起了目标接口请求。如果监听器只挂在老页面上新页面的响应你是收不到的。解决方案有两个层面。第一监听器挂在 context 上而不是 page 上context browser.new_context() context.on(response, on_response)context 级别的监听能捕获这个浏览器上下文里所有页面的请求包括新开标签页和弹窗页面。这是做复杂页面采集时最稳妥的做法。第二个层面是响应对象生命周期。Python 回调里的response对象只在回调执行期间有效如果你把它存到队列里等主流程处理完再去调response.json()可能已经拿不到数据了。所以我在回调里直接就把response.body()或json()的结果取出来转成普通 Python 对象再保存。这就是为什么build_row在回调阶段就要执行。5.4 性能优化与合规建议采集脚本跑久了最常见的性能问题是内存无限增长。captured列表一直 append跑半小时可能就攒了几十万条数据。解决办法是设置一个阈值比如每收集 500 条就写入一次 SQLite然后清空列表。if len(captured) 500: insert_rows(captured) captured.clear()另一个性能优化手段是浏览器复用。采集多个页面时不要每次采集都重新启动浏览器而是在同一个 browser 实例里创建多个 context每个 context 里开一个 page。这样能大幅减少浏览器启动开销。但要注意并发页面太多会抢占 CPU 和网络资源建议控制在 3 到 5 个并发页面。关于合规说点实在的。网络层采集本质上和你在浏览器开发者工具里看接口是一回事但能不能采集、允不允许采集是另一回事。我的建议是只采集公开页面数据设置合理访问频率遵守目标网站的 robots 协议和服务条款。不要把这套方案用在高频请求或涉及个人隐私的数据上。爬虫首先是工程问题但边界问题同样重要。技术能力越强越应该克制。最后再分享一个我自己很受用的习惯每次接到采集任务先别写任何解析代码打开浏览器开发者工具的 Network 面板把目标接口的返回数据从头到尾看一遍再写监听器去打印原始 JSON确认结构稳定后才开始写字段提取和持久化逻辑。这个流程看起来多了一步实际上能帮你少走很多弯路。毕竟只有数据源摸清了后面的代码才写得踏实。