Selenium 发送 Fetch 请求及异步处理方法

本教程详细阐述了如何利用 Selenium 的 execute_async_script 方法在浏览器环境中执行 JavaScript fetch 请求，并获取其异步响应。内容涵盖了如何设置自定义 HTTP 头、处理 GET 和 POST 请求、解析 JSON 响应，以及配置浏览器代理，从而有效解决 Selenium 自动化测试中复杂的网络请求场景。

在自动化测试和网页抓取场景中，Selenium 通常用于模拟用户与网页的交互。然而，有时我们需要在浏览器上下文中直接发起 HTTP 请求，例如利用当前浏览器会话的 Cookies、LocalStorage 或绕过 CORS 限制等。JavaScript 的 fetch API 是执行此类请求的强大工具，但由于其异步特性，通过 Selenium 的 execute_script 方法直接调用会遇到挑战。本教程将深入探讨如何结合 Selenium 的 execute_async_script 方法，在 Python 中正确执行 fetch 请求并获取其响应。

1. execute_async_script 的核心机制

fetch 请求是异步的，这意味着它不会立即返回结果，而是返回一个 Promise 对象。Selenium 的 execute_script 方法只会返回 JavaScript 代码的同步执行结果，因此无法直接获取 Promise 最终解决的值。

为了解决这个问题，Selenium 提供了 execute_async_script 方法。该方法允许执行异步 JavaScript 代码，并通过一个回调函数来接收最终结果。在 execute_async_script 中，JavaScript 代码会接收一个特殊的 arguments[0] 参数，它是一个由 Selenium 注入的回调函数。当异步操作完成时，JavaScript 代码需要调用这个回调函数，并将结果作为参数传递，Selenium 才能在 Python 端接收到这个结果。

2. 使用 execute_async_script 发送 GET 请求并获取 JSON 响应

以下是一个使用 fetch 发送 GET 请求，并带有自定义 HTTP 头，最终获取 JSON 响应的示例：

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import json
import time

# 1. 配置 WebDriver
chrome_options = Options()
# 可以添加无头模式等其他选项
# chrome_options.add_argument("--headless")
# chrome_options.add_argument("--disable-gpu")

driver = webdriver.Chrome(options=chrome_options)
# 设置异步脚本的超时时间，以防fetch请求长时间未响应
driver.set_script_timeout(10) # 10秒超时

try:
    # 2. 导航到一个页面 (可以是任何页面，因为fetch请求是独立发起的)
    driver.get("https://www.example.com") # 或者任何你希望执行脚本的页面

    # 3. 构造 JavaScript fetch 请求
    # 注意：arguments[0] 是 Selenium 提供的回调函数
    # .then(r => r.json()) 将响应体解析为 JSON
    # .then(callback) 将解析后的 JSON 数据传递给 Python
    # .catch(error => callback({ error: error.message })) 处理错误
    js_script = """
    let callback = arguments[0];
    fetch('https://httpbin.org/get', {
        method: 'GET',
        headers: {
            'Accept': 'application/json',
            'X-Custom-Header-1': 'Value1',
            'X-Custom-Header-2': 'Value2'
        }
    })
    .then(response => {
        if (!response.ok) {
            throw new Error('Network response was not ok ' + response.statusText);
        }
        return response.json();
    })
    .then(data => callback(data))
    .catch(error => callback({ error: error.message }));
    """

    # 4. 执行异步脚本并获取结果
    print("正在执行 GET 请求...")
    response_data = driver.execute_async_script(js_script)

    # 5. 处理 Python 端的结果
    if isinstance(response_data, dict) and "error" in response_data:
        print(f"Fetch 请求发生错误: {response_data['error']}")
    else:
        print("GET 请求成功，响应数据:")
        print(json.dumps(response_data, indent=2))
        # 验证自定义头是否被发送
        if 'headers' in response_data and 'X-Custom-Header-1' in response_data['headers']:
            print(f"自定义头 'X-Custom-Header-1' 已发送: {response_data['headers']['X-Custom-Header-1']}")

except Exception as e:
    print(f"发生异常: {e}")

finally:
    # 6. 关闭浏览器
    driver.quit()

3. 使用 execute_async_script 发送 POST 请求

发送 POST 请求与 GET 类似，但需要指定 method: 'POST' 和 body 参数。body 通常是一个 JSON 字符串。

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import json
import time

chrome_options = Options()
driver = webdriver.Chrome(options=chrome_options)
driver.set_script_timeout(10)

try:
    driver.get("https://www.example.com")

    post_payload = {
        "name": "Selenium User",
        "age": 30,
        "city": "Testville"
    }

    js_script_post = f"""
    let callback = arguments[0];
    fetch('https://httpbin.org/post', {{
        method: 'POST',
        headers: {{
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'X-Auth-Token': 'your_auth_token_here'
        }},
        body: JSON.stringify({json.dumps(post_payload)})
    }})
    .then(response => {{
        if (!response.ok) {{
            throw new Error('Network response was not ok ' + response.statusText);
        }}
        return response.json();
    }})
    .then(data => callback(data))
    .catch(error => callback({{ error: error.message }}));
    """

    print("\n正在执行 POST 请求...")
    response_data_post = driver.execute_async_script(js_script_post)

    if isinstance(response_data_post, dict) and "error" in response_data_post:
        print(f"Fetch POST 请求发生错误: {response_data_post['error']}")
    else:
        print("POST 请求成功，响应数据:")
        print(json.dumps(response_data_post, indent=2))
        # 验证请求体和自定义头是否被发送
        if 'json' in response_data_post and response_data_post['json'] == post_payload:
            print("请求体数据正确发送。")
        if 'headers' in response_data_post and 'X-Auth-Token' in response_data_post['headers']:
            print(f"自定义头 'X-Auth-Token' 已发送: {response_data_post['headers']['X-Auth-Token']}")

except Exception as e:
    print(f"发生异常: {e}")

finally:
    driver.quit()

4. 配置代理服务器

如果希望通过代理服务器发送 fetch 请求，需要为 Selenium WebDriver 配置代理。fetch 请求在浏览器环境中执行，因此它会遵循浏览器本身的代理设置。

from selenium import webdriver
from selenium.webdriver.chrome.options import Options

proxy_server_url = "127.0.0.1:8888" # 你的代理服务器地址和端口

chrome_options = Options()
chrome_options.add_argument(f'--proxy-server={proxy_server_url}')
# 如果代理需要认证，可能需要额外的扩展或配置
# chrome_options.add_argument("--ignore-certificate-errors") # 如果代理有自签名证书，可能需要忽略证书错误

driver = webdriver.Chrome(options=chrome_options)
driver.set_script_timeout(10)

try:
    driver.get("https://www.example.com")
    print(f"浏览器已启动，并配置代理: {proxy_server_url}")

    # 此时，通过 execute_async_script 发起的 fetch 请求应该会通过配置的代理服务器
    js_script_proxy_test = """
    let callback = arguments[0];
    fetch('https://api.ipify.org?format=json') // 这是一个返回当前IP地址的API
    .then(response => response.json())
    .then(data => callback(data))
    .catch(error => callback({ error: error.message }));
    """
    print("正在通过代理执行 fetch 请求以获取外部 IP 地址...")
    ip_response = driver.execute_async_script(js_script_proxy_test)

    if isinstance(ip_response, dict) and "error" in ip_response:
        print(f"Fetch 请求发生错误: {ip_response['error']}")
    else:
        print(f"通过代理获取的 IP 地址: {ip_response.get('ip')}")
        # 如果你正在运行一个本地代理（如 Burp Suite, Fiddler），
        # 此时你应该能在代理工具中看到这个请求。

except Exception as e:
    print(f"发生异常: {e}")

finally:
    driver.quit()

注意事项：

• 确保代理服务器正在运行且可访问。
• 如果代理需要认证，直接通过 add_argument 可能不够，可能需要使用带有认证的代理扩展。
• fetch 请求会使用浏览器配置的代理，而不是 Selenium 驱动本身的 HTTP 客户端代理。

5. 注意事项与最佳实践

1. 错误处理： 在 JavaScript 的 fetch 请求中，务必使用 .catch() 来捕获网络错误或响应处理错误，并将错误信息传递回 Python，以便进行适当的日志记录或异常处理。
2. 超时设置： driver.set_script_timeout() 是至关重要的。如果 fetch 请求长时间没有响应或回调，Selenium 会抛出 TimeoutException。根据你的网络环境和预期响应时间合理设置超时。
3. 数据序列化： Python 和 JavaScript 之间传递复杂数据时，需要进行 JSON 序列化和反序列化。例如，Python 中的字典需要 json.dumps() 转换为 JSON 字符串才能在 JavaScript 中作为 body 使用，而 JavaScript 返回的 JSON 数据在 Python 中会自动解析为字典或列表。
4. 响应类型： fetch 响应有多种处理方式，如 response.json() (解析 JSON)、response.text() (解析文本)、response.blob() (解析二进制数据) 等。根据实际需求选择合适的方法。
5. 适用场景：
   • 何时使用： 当你需要利用当前浏览器会话的上下文（如登录状态、Cookies、LocalStorage、JS 变量）发起请求时，或者需要模拟浏览器环境下的复杂网络行为（如绕过 CORS）。
   • 何时不使用： 对于简单的、不需要浏览器上下文的 HTTP 请求，Python 的 requests 库通常更简洁、高效。
6. 安全性： 在 execute_async_script 中注入的 JavaScript 代码拥有与当前页面相同的权限。请确保注入的代码是安全的，避免潜在的跨站脚本（XSS）风险。

总结

通过 execute_async_script 方法，Selenium 能够有效地桥接 Python 与浏览器内部的异步 JavaScript fetch API。这使得在自动化测试和数据抓取场景中，可以灵活地执行复杂的网络请求，利用浏览器上下文的优势，并获取精确的异步响应数据。掌握这一技术，将极大地扩展 Selenium 的应用范围，使其能够处理更广泛、更复杂的自动化任务。