从数据抓取到性能监控：一个Playwright page.on()方法，搞定三种副业场景-平芜编程栈

从数据抓取到性能监控：Playwright的page.on()方法跨界实战指南

如果你正在寻找一种能同时解决数据采集、自动化办公和前端性能分析的"瑞士军刀"级工具，Playwright的page.on()方法绝对值得深入探索。这个看似简单的监听器接口，实际上能帮你打开三个完全不同领域的效率之门——不需要学习三种工具，掌握这一个API就能让你在副业开发中游刃有余。

1. 数据抓取：精准拦截动态内容的秘密武器

传统爬虫最头疼的问题莫过于动态加载内容。当页面通过Ajax异步加载数据时，常规的HTML解析往往束手无策。而page.on('response')监听器能让你直击数据传输的"咽喉要道"。

假设我们需要抓取某电商网站的价格波动数据，页面上的价格是通过接口动态获取的。下面这个实战脚本展示了如何精确捕获目标API响应：

async def intercept_api_responses(): browser = await playwright.chromium.launch() page = await browser.new_page() target_api = "https://api.example.com/product/prices" async def handle_response(response): if target_api in response.url: json_data = await response.json() print(f"捕获价格数据: {json_data}") # 这里可以添加数据存储逻辑 page.on('response', handle_response) await page.goto('https://www.example.com/product-page') await page.wait_for_timeout(5000) # 留出足够时间捕获响应 await browser.close()

关键技巧：

使用response.url过滤目标接口
await response.json()直接解析JSON格式响应
设置合理的等待时间确保捕获完整数据流

对比传统爬虫方案，这种方法有三大优势：

方法	动态内容支持	反爬抵抗性	资源消耗
传统HTML解析	差	低	高
无头浏览器渲染	好	中	极高
`page.on('response')`监听	极好	高	中

提示：遇到接口参数加密的情况，可以配合page.on('request')监听器记录完整的请求参数，逆向分析加密逻辑。

2. 自动化办公：智能处理弹窗与文件下载

日常办公自动化中，网页弹窗和文件下载是最常见的痛点。page.on()的dialog和download事件监听能让你轻松掌控这些"意外情况"。

2.1 弹窗自动应答系统

想象一下需要批量处理数百个网页表单，每个提交后都可能出现确认弹窗。手动操作简直是一场噩梦。下面的解决方案能自动识别并处理各种弹窗类型：

async def auto_handle_dialogs(): browser = await playwright.chromium.launch(headless=False) page = await browser.new_page() async def dialog_handler(dialog): print(f"检测到 {dialog.type} 弹窗: {dialog.message}") if dialog.type == 'confirm': await dialog.accept() # 自动确认 elif dialog.type == 'prompt': await dialog.accept("默认输入值") # 自动填充 page.on('dialog', dialog_handler) # 示例：自动填写并提交表单 await page.goto('https://example.com/form') await page.fill('#username', 'test_user') await page.click('#submit-btn') await browser.close()

2.2 文件下载管家

自动化下载文件时，最大的挑战是何时开始下载、下载是否完成。这个方案能精准捕获下载事件并管理文件：

async def download_manager(): browser = await playwright.chromium.launch() context = await browser.new_context(accept_downloads=True) page = await context.new_page() download_path = '/path/to/downloads' download_status = {} def handle_download(download): file_id = download.url.split('/')[-1] print(f"开始下载: {file_id}") download_status[file_id] = 'started' # 等待下载完成 download_path = download.path() download_status[file_id] = 'completed' print(f"下载完成: {download_path}") page.on('download', handle_download) # 触发下载操作 await page.goto('https://example.com/download-page') await page.click('#download-button') # 检查下载状态 while not download_status.get('target_file') == 'completed': await page.wait_for_timeout(1000) await browser.close()

进阶技巧：

使用context.expect_download()可以更精确地等待特定下载
结合download.failure()方法处理下载失败情况
通过download.suggested_filename获取原始文件名

3. 前端性能监控：打造轻量级分析工具

前端开发者经常需要分析页面性能，但专业工具往往过于复杂。利用page.on()监听网络活动，你可以自制一个轻量级性能监控器。

3.1 关键性能指标采集

这个脚本能捕获页面加载过程中的关键时间节点和资源加载情况：

async def performance_monitor(): browser = await playwright.chromium.launch() page = await browser.new_page() performance_data = { 'requests': [], 'timings': {} } # 记录所有请求 def log_request(request): performance_data['requests'].append({ 'url': request.url, 'method': request.method, 'size': request.headers.get('content-length') }) # 记录关键时间点 def log_load(): performance_data['timings']['load'] = time.time() page.on('request', log_request) page.on('load', log_load) start_time = time.time() await page.goto('https://example.com') performance_data['timings']['total'] = time.time() - start_time # 输出性能报告 print(f"完整加载时间: {performance_data['timings']['total']:.2f}s") print(f"总请求数: {len(performance_data['requests'])}") await browser.close()

3.2 资源加载瀑布流分析

更深入的分析可以生成类似Chrome DevTools的瀑布流图：

async def waterfall_analysis(): browser = await playwright.chromium.launch() page = await browser.new_page() resources = [] async def on_request(request): resources.append({ 'url': request.url, 'start': time.time(), 'end': None }) async def on_response(response): for res in resources: if res['url'] == response.url: res['end'] = time.time() res['duration'] = res['end'] - res['start'] break page.on('request', on_request) page.on('response', on_response) await page.goto('https://example.com') await page.wait_for_load_state('networkidle') # 生成排序后的资源加载时间表 sorted_resources = sorted(resources, key=lambda x: x['start']) for idx, res in enumerate(sorted_resources, 1): print(f"{idx}. {res['url']} - {res['duration']:.2f}s") await browser.close()

性能优化建议：

重点关注耗时超过1秒的资源
检查同一域名下的并行请求数量
识别未使用的CSS/JS资源
监控第三方脚本的加载时间

4. 高级技巧与最佳实践

掌握了基础应用后，让我们深入一些能大幅提升效率的高级技巧。

4.1 事件监听器的精细控制

page.on()的强大之处在于它的灵活性，但也需要注意管理：

# 添加监听器 def request_logger(request): print(f"请求: {request.url}") page.on('request', request_logger) # 移除特定监听器 page.off('request', request_logger) # 一次性监听器 async with page.expect_event('request') as event_info: await page.click('#trigger-button') request_event = await event_info.value

4.2 多事件协同工作流

结合多个事件类型可以实现更复杂的自动化：

async def complex_workflow(): page.on('request', lambda req: print(f"请求开始: {req.url}")) page.on('response', lambda res: print(f"响应到达: {res.status}")) page.on('load', lambda: print("页面完全加载")) await page.goto('https://example.com') # 等待特定条件组合 async def wait_for_conditions(): await page.wait_for_event('load') await page.wait_for_selector('#content-loaded') return True await wait_for_conditions()

4.3 错误处理与调试

完善的错误处理能让你的脚本更加健壮：

async def robust_listener(): try: page.on('response', async lambda res: ( print(await res.json()) if res.ok else print(f"错误响应: {res.status}") )) await page.goto('https://example.com') except Exception as e: print(f"监听器错误: {e}") finally: # 清理监听器 page.remove_all_listeners('response')

性能考量：