【Selenium】实战:利用CDP协议精准捕获与解析异步网络请求
1. 为什么需要CDP协议捕获网络请求
做爬虫开发的朋友应该都遇到过这样的场景:打开一个网页,页面上明明显示了数据,但查看网页源码却找不到对应的内容。这是因为现代网站普遍采用前后端分离的架构,数据通过XHR或Fetch API异步加载。这时候传统的页面解析方法就失效了。
我最近在采集某电商网站价格数据时就遇到了这个问题。页面加载后可以看到完整的商品列表,但查看源码只有个空壳框架。通过浏览器开发者工具的Network面板,发现数据是通过API接口返回的JSON。这时候就需要一种能直接捕获网络请求的方法。
Selenium虽然能模拟浏览器操作,但默认只提供页面源码获取(page_source)。要获取异步请求数据,就需要请出我们今天的主角——Chrome DevTools Protocol(CDP)。这个协议可以直接与浏览器内核对话,获取包括网络请求在内的各种底层数据。
2. CDP协议基础入门
2.1 什么是CDP协议
CDP是Chrome浏览器提供的一套调试协议,它允许开发者通过程序控制浏览器行为。就像我们用开发者工具调试网页一样,CDP提供了更底层的接口。
这个协议有几个关键特点:
- 基于WebSocket通信
- 支持实时监控网络活动
- 可以拦截和修改请求/响应
- 支持获取DOM、执行JavaScript等操作
在Selenium中,我们可以通过execute_cdp_cmd方法调用CDP命令。这相当于把开发者工具的功能集成到了自动化脚本中。
2.2 Selenium与CDP的版本适配
这里有个需要注意的地方:Selenium 3和4对CDP的支持方式不同。
Selenium 3的配置方式:
caps = { "browserName": "chrome", 'goog:loggingPrefs': {'performance': 'ALL'} } browser = webdriver.Chrome(desired_capabilities=caps)Selenium 4改用Options配置:
options = Options() caps = { "browserName": "chrome", 'goog:loggingPrefs': {'performance': 'ALL'} } for key, value in caps.items(): options.set_capability(key, value) browser = webdriver.Chrome(options=options)如果版本不匹配会出现TypeError,这是新手常踩的坑。建议统一使用Selenium 4的写法,兼容性更好。
3. 实战:捕获API请求数据
3.1 开启性能日志记录
第一步要启用浏览器的性能日志,这样才能捕获网络请求:
from selenium import webdriver from selenium.webdriver.chrome.options import Options options = Options() options.set_capability('goog:loggingPrefs', {'performance': 'ALL'}) browser = webdriver.Chrome(options=options)这里performance: 'ALL'表示记录所有性能日志,包括网络请求、时间线等数据。启动后会生成大量日志,所以下一步我们需要过滤。
3.2 过滤无关请求
一个普通网页会加载数十个资源,但通常我们只需要关注API请求。这是我的过滤方法:
def filter_type(_type): # 需要过滤的资源类型 unwanted_types = [ 'application/javascript', 'text/css', 'image/png', 'image/jpeg', 'font/woff2' ] return _type not in unwanted_types这个过滤器会排除JS、CSS、图片等静态资源,只保留API请求。你可以根据实际需求调整过滤条件。
3.3 解析网络请求日志
获取和解析日志的核心代码如下:
performance_log = browser.get_log('performance') for packet in performance_log: message = json.loads(packet['message'])['message'] # 只处理网络响应 if message['method'] != 'Network.responseReceived': continue # 获取请求详情 params = message['params'] request_id = params['requestId'] url = params['response']['url'] mime_type = params['response']['mimeType'] if not filter_type(mime_type): continue # 获取响应体 try: resp = browser.execute_cdp_cmd( 'Network.getResponseBody', {'requestId': request_id} ) print(f"API: {url}") print(f"Data: {resp['body']}") except: pass这段代码做了几件事:
- 获取所有性能日志
- 筛选出网络响应事件
- 提取请求URL和类型
- 通过CDP获取响应内容
4. 常见问题与解决方案
4.1 数据捕获不全怎么办
在实际使用中,可能会遇到这些问题:
- 页面加载完成前请求已经结束
- 滚动加载的内容没有捕获到
- 部分请求返回空数据
我的解决方案是:
- 增加显式等待时间
from time import sleep sleep(3) # 等待异步加载完成- 模拟滚动操作触发加载
browser.execute_script("window.scrollTo(0, document.body.scrollHeight)")- 重试机制处理偶发失败
4.2 处理登录和认证
对于需要登录的网站,可以先手动登录后复用Cookie:
# 添加保存的Cookie browser.get("https://example.com") # 先访问域名 for cookie in saved_cookies: browser.add_cookie(cookie) browser.refresh() # 刷新使Cookie生效这样就能保持登录状态,捕获需要认证的API请求。
5. 高级技巧与优化建议
5.1 请求拦截与修改
CDP不仅能捕获请求,还能拦截和修改。比如我们可以:
- 屏蔽不必要的资源加载
- 修改请求头
- 模拟移动端环境
示例:禁用图片加载
browser.execute_cdp_cmd('Network.enable', {}) browser.execute_cdp_cmd('Network.setBlockedURLs', { 'urls': ['*.jpg', '*.png', '*.gif'] })5.2 性能优化技巧
处理大量请求时,可以优化以下几点:
- 按需启用日志
# 先不记录日志 browser.get(url) # 需要时再开启 browser.execute_cdp_cmd('Network.enable', {})- 使用请求过滤
# 只监听特定URL browser.execute_cdp_cmd('Network.setRequestInterception', { 'patterns': [{'urlPattern': '*api*'}] })- 及时清理日志
browser.get_log('performance') # 获取后会清空日志5.3 与Page Object模式结合
对于大型项目,建议将CDP操作封装成Page Object:
class ProductPage: def __init__(self, browser): self.browser = browser def get_prices(self): # 捕获价格API logs = self._capture_network_logs() return self._parse_prices(logs) def _capture_network_logs(self): # CDP操作封装 ...这样既保持了代码整洁,又便于复用。
