[Python] Selenium监控网络请求

  Selenium监控网络有两种方式，第一种使用代理，第二张是使用CDP（Chrome DevTools Protocol）协议，下面直接进入主题分别介绍如何使用代理和CDP协议监控网络请求。

  一、使用Selenium-Wire设置代理拦截处理请求。

  Selenium-Wire是基于Selenium开发的抓包工具，基本使用方式如下：

from selenium import webdriver
from seleniumwire import webdriver as webdriverwire
from seleniumwire.request import Request, Response


def request_interceptor(request: Request):
    """处理请求"""
    print("request_interceptor", request.method, request.url)


def response_interceptor(request: Request, response: Response):
    """处理请求响应"""
    print("response_interceptor", request.method, request.url, response.status_code)

options = webdriver.ChromeOptions()
options.add_argument("--ignore-certificate-errors")

# url正则表达式集合
scopes = set()
scopes.add("https?://.+/confirm")
ignore_http_methods = [
    "OPTIONS",
    "HEAD",
    "CONNECT",
    "TRACE",
    "PATCH",
]
seleniumwire_options = {
    # 过滤域名
    "exclude_hosts": ["www.exclude.com"],
    # 过滤请求方法
    "ignore_http_methods": ignore_http_methods,
    "verify_ssl": False,  # 不验证证书
    "enable_logging": True,
    "request_storage": "memory",  # 缓存到内存
    # "request_storage_base_dir": request_storage_base_dir,  # 设置请求缓存的目录
    "request_storage_max_size": 100,  # Store no more than 100 requests in memory
}

driver = webdriverwire.Chrome(
    options=options,
    seleniumwire_options=seleniumwire_options,
)
driver.request_interceptor = request_interceptor
driver.response_interceptor = response_interceptor
driver.scopes = list(scopes)

driver.get('https://www.baidu.com')

  解释：

  scopes：通过正则表达式过滤网络请求，过滤后的请求更方便查找，注意不符合正则表达式的URL依然使用代理。

  exclude_hosts：过滤不使用代理的域名。

  request_interceptor：拦截处理请求。

  response_interceptor：拦截处理请求响应结果。

  注意：

  （1）要把https证书放程序目录的seleniumwire文件夹，若网页提示https不安全，需要导入seleniumwire https证书到受信任的根目录。

  （2）部分网站无法使用seleniumwire打开，会提示502，需要设置代理，下面示范使用Fiddler代理。

  使用Fiddler代理抓包：

  1. 安装并运行Fiddler Classic，设置开启HTTPS抓包，保证软件开启就行，不需要实时抓包，Fiddler默认使用8888端口代理。

  2. 设置seleniumwire_options使用Fiddler代理：

# 设置代理
seleniumwire_options["proxy"] = {
    "https": "https://127.0.0.1:8888",
    "http": "http://127.0.0.1:8888",
    "no_proxy": "localhost,127.0.0.1",  # 不使用代理的地址
}

  二、使用CDP协议监控网络请求。

"""CDP监控网络请求"""

import json
import re
import time
from typing import Dict, List, TypeVar
from selenium import webdriver
from selenium.webdriver.chrome.webdriver import WebDriver

from log import log


class CdpNetworkMonitor:
    """CDP监控网络请求"""

    _S = TypeVar("_S", bound="CdpNetworkMonitor")

    static_monitor: _S = None

    def __init__(self, driver: WebDriver, url_patterns: List[str]):
        """
        :param url_patterns url正则表达式集合
        """
        self.driver = driver
        self.url_patterns = url_patterns
        """url正则表达式集合"""
        self.last_log_timestamp = 0
        """最新日志时间戳"""
        self.requests: Dict[str, dict] = {}
        """请求集合"""
        self.last_flush_time: float = 0
        """上次读取性能日志时间"""
        CdpNetworkMonitor.static_monitor = self

    def start(self):
        """启用网络监控"""
        try:
            self.driver.execute_cdp_cmd("Network.enable", {})
        except Exception as ex:
            log("启用网络监控", type(ex), ex)

    def stop(self):
        """禁用网络监控"""
        try:
            self.driver.execute_cdp_cmd("Network.disable", {})
        except Exception as ex:
            log("禁用网络监控", type(ex), ex)

    def flush(self):
        """CDP读取性能日志"""
        timestamp = time.time()
        if timestamp - self.last_flush_time < 3:
            return
        self.last_flush_time = timestamp
        # 获取性能日志
        try:
            cdp_logs: List[dict] = self.driver.get_log("performance")
            for cdp_log in cdp_logs:
                log_timestamp: int = cdp_log.get("timestamp")
                if log_timestamp <= self.last_log_timestamp:
                    # 日志已读取
                    continue
                self.last_log_timestamp = log_timestamp
                log_json: dict = json.loads(cdp_log["message"])
                message: dict = log_json.get("message")
                if not message:
                    log("CDP性能日志无法获取message")
                    continue
                method: str = message.get("method")
                if not method:
                    log("CDP性能日志无法获取method")
                    continue
                if method == "Network.requestWillBeSent":
                    # 发起请求
                    request: dict = message["params"]["request"]
                    url = request.get("url")
                    if any(
                        re.search(url_pattern, url) for url_pattern in self.url_patterns
                    ):
                        # 符合url正则表达式
                        request_id = message["params"]["requestId"]
                        log(request_id, request)
                        self.requests[request_id] = request
                elif message.get("method") == "Network.responseReceived":
                    # 响应请求
                    request_id = message["params"]["requestId"]
                    if request_id in self.requests:
                        response: dict = message["params"]["response"]
                        # 获取响应内容
                        response_body = self.driver.execute_cdp_cmd(
                            "Network.getResponseBody", {"requestId": request_id}
                        )
                        response["body"] = response_body
                        log(request_id, response)
                        self.requests[request_id]["response"] = response
        except Exception as ex:
            log("CDP读取性能日志", type(ex), ex)

    def clear(self):
        """清理请求缓存"""
        try:
            self.requests.clear()
            self.driver.execute_cdp_cmd("Log.clear", {})
        except Exception as ex:
            log("清理请求缓存", type(ex), ex)


options = webdriver.ChromeOptions()
options.set_capability("goog:loggingPrefs", {"performance": "ALL"})  # 启用性能日志
driver = webdriver.Chrome(options=options)
scopes = ["https?://www.baidu.com"]
monitor = CdpNetworkMonitor(driver, scopes)
monitor.start()

driver.get('https://www.baidu.com')

time.sleep(10)
monitor.flush()

for request_id in monitor.requests:
    print(monitor.requests[request_id])

   解释：

  Python版本的CDP网络请求事件无法通过事件回调机制监听，只能轮询性能日志，这里我封装了CdpNetworkMonitor工具类过滤要监控的请求。

  CDP也可以拦截修改请求，但需要轮询性能日志并通过execute_cdp_cmd处理请求，若不及时处理拦截的请求会导致请求失败，这与driver.get()打开页面有冲突，所以得出结论Python的CDP无法拦截处理请求，但可以仅查看。