使用Selenium Wire捕获和分析Selenium自动化中的网络请求

pip install selenium-wire

from seleniumwire import webdriver
import json
import time

# 配置WebDriver选项，例如无头模式
options = webdriver.ChromeOptions()
options.add_argument('--headless') # 可选：在无头模式下运行
options.add_argument('--disable-gpu') # 禁用GPU加速，有时在无头模式下需要

# 初始化Selenium Wire WebDriver
# 默认情况下，Selenium Wire 会为每个请求和响应存储完整的body。
# 如果内存成为问题，可以配置为不存储body，或只存储特定请求的body。
# driver = webdriver.Chrome(options=options, seleniumwire_options={'disable_capture': True})
driver = webdriver.Chrome(options=options)

from seleniumwire import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import json
import time

# 1. 初始化Selenium Wire WebDriver
options = webdriver.ChromeOptions()
# options.add_argument('--headless') # 根据需要启用无头模式
options.add_argument('--disable-gpu')
driver = webdriver.Chrome(options=options)

try:
    # 2. 导航到目标网页
    # 替换为你的目标URL，这里使用一个示例
    driver.get("https://www.example.com/some_dynamic_page")
    print("已导航到页面。")

    # 模拟页面加载和元素出现
    # 假设页面上有一个ID为 'dataButton' 的按钮
    # 你可能需要等待元素可点击
    wait = WebDriverWait(driver, 10)
    data_button = wait.until(EC.element_to_be_clickable((By.ID, "dataButton")))

    # 3. 清空之前的请求记录，确保只捕获点击后的请求
    driver.delete_all_requests()
    print("已清空所有历史请求。")

    # 4. 执行页面交互，触发API请求
    data_button.click()
    print("已点击按钮，等待API请求...")

    # 5. 等待一段时间，让API请求完成并被捕获
    # 或者使用更智能的等待方式，例如等待某个特定的API请求出现
    time.sleep(5) # 简单等待5秒，实际应用中应更精确地等待

    # 6. 遍历捕获到的请求，定位目标API
    target_api_url_part = '/api/get_data' # 目标API的URL片段
    found_api_response = None

    for request in driver.requests:
        if request.url.endswith(target_api_url_part) and request.response:
            print(f"捕获到目标API请求: {request.url}")
            print(f"请求方法: {request.method}")
            print(f"响应状态码: {request.response.status_code}")

            # 尝试获取并解析JSON响应体
            try:
                # request.response.body 是字节流，需要解码
                response_body = request.response.body.decode('utf-8')
                json_data = json.loads(response_body)
                found_api_response = json_data
                print("API响应数据:")
                print(json.dumps(json_data, indent=2, ensure_ascii=False))
                break # 找到目标后即可退出循环
            except json.JSONDecodeError:
                print("响应体不是有效的JSON格式。")
                print(f"原始响应体: {response_body[:200]}...") # 打印部分原始响应体
            except Exception as e:
                print(f"处理响应时发生错误: {e}")

    if not found_api_response:
        print(f"未找到包含 '{target_api_url_part}' 的API请求或其响应。")

    # 7. 对获取到的数据进行后续处理
    if found_api_response:
        print("\n成功获取并解析API数据，可以进行进一步处理。")
        # 例如：assert found_api_response['status'] == 'success'
        # 或者提取特定字段：data_items = found_api_response.get('items', [])

except Exception as e:
    print(f"发生错误: {e}")

finally:
    # 8. 关闭浏览器
    driver.quit()
    print("浏览器已关闭。")

# 等待一个URL包含特定字符串的请求
request = driver.wait_for_request('/api/get_data', timeout=10)
if request.response:
    print(f"捕获到请求: {request.url}")
    print(f"响应体: {request.response.body.decode('utf-8')}")

for request in driver.requests:
    if request.url.startswith('https://api.example.com/') and \
       request.method == 'POST' and \
       request.response.status_code == 200:
        # 处理符合条件的请求
        pass