Selenium Wire 请求拦截：验证码识别实战

页面弹了验证码，document.querySelector('[data-sitekey]') 却返回 null——sitekey 多半不在 DOM 里，而在 reCAPTCHA 的 iframe URL 或配置接口的响应中。Selenium Wire 把每条 HTTP 请求记成 Python 对象：从流量里读参数，交给 CaptchaAI 识别，再写回表单。

原生 Selenium 缺的是网络可见性

验证码资源由独立域名异步加载，网络层比 DOM 层看得更早。

能力	标准 Selenium	Selenium Wire
读取请求	不支持	支持
修改请求头	受限	完全可控
拦截响应	不支持	支持
代理支持	基础	含认证代理
网络日志	仅 DevTools	Python 对象
HAR 导出	不支持	支持

国内站点多见 GeeTest（极验）滑块，国际站点以 reCAPTCHA 和 Turnstile 为主。CaptchaAI 覆盖 reCAPTCHA v2/v3、Turnstile 与 GeeTest v3，hCaptcha、FunCaptcha 暂不支持。

安装依赖

pip install seleniumwire selenium webdriver-manager requests

国内直连 PyPI 较慢，可加镜像：pip install -i https://pypi.tuna.tsinghua.edu.cn/simple seleniumwire。

核心类：识别验证码并调用 CaptchaAI

这个类做三件事：启动 Selenium Wire 版 Chrome、扫描 driver.requests 取 sitekey、提交 in.php 后轮询 res.php 拿 token。

from seleniumwire import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import requests
import time
import json

class SeleniumWireCaptchaSolver:
    BASE_URL = "https://ocr.captchaai.com"

    def __init__(self, api_key, proxy=None):
        self.api_key = api_key
        self.proxy = proxy
        self.driver = None

    def create_driver(self, headless=True):
        options = webdriver.ChromeOptions()
        if headless:
            options.add_argument("--headless=new")
        options.add_argument("--no-sandbox")
        options.add_argument("--no-sandbox")
        options.add_argument("--window-size=1920,1080")

        wire_options = {}
        if self.proxy:
            wire_options["proxy"] = {
                "http": self.proxy,
                "https": self.proxy,
            }

        self.driver = webdriver.Chrome(
            service=Service(ChromeDriverManager().install()),
            options=options,
            seleniumwire_options=wire_options,
        )
        return self.driver

    def detect_captcha_from_requests(self):
        """Detect CAPTCHA type from intercepted network requests."""
        for request in self.driver.requests:
            url = request.url

            if "recaptcha/api2/anchor" in url or "recaptcha/enterprise/anchor" in url:
                # Extract sitekey from reCAPTCHA iframe URL
                import re
                match = re.search(r"k=([A-Za-z0-9_-]+)", url)
                if match:
                    return {
                        "type": "recaptcha_v2",
                        "sitekey": match.group(1),
                        "page_url": self.driver.current_url,
                    }

            if "recaptcha/api2/reload" in url or "recaptcha/enterprise/reload" in url:
                return {
                    "type": "recaptcha_v3",
                    "sitekey": self._extract_v3_sitekey(),
                    "page_url": self.driver.current_url,
                }

            if "challenges.cloudflare.com" in url:
                sitekey = self._extract_turnstile_sitekey()
                if sitekey:
                    return {
                        "type": "turnstile",
                        "sitekey": sitekey,
                        "page_url": self.driver.current_url,
                    }

        return None

    def _extract_v3_sitekey(self):
        for request in self.driver.requests:
            if "recaptcha" in request.url and "render=" in request.url:
                import re
                match = re.search(r"render=([A-Za-z0-9_-]+)", request.url)
                if match:
                    return match.group(1)
        return self.driver.execute_script(
            "return document.querySelector('[data-sitekey]')?.getAttribute('data-sitekey')"
        )

    def _extract_turnstile_sitekey(self):
        return self.driver.execute_script(
            "return document.querySelector('.cf-turnstile[data-sitekey]')?.getAttribute('data-sitekey')"
        )

    def solve_captcha(self, captcha_info):
        """Solve detected CAPTCHA via CaptchaAI API."""
        params = {"key": self.api_key, "json": 1}

        if captcha_info["type"] == "recaptcha_v2":
            params.update({
                "method": "userrecaptcha",
                "googlekey": captcha_info["sitekey"],
                "pageurl": captcha_info["page_url"],
            })
        elif captcha_info["type"] == "recaptcha_v3":
            params.update({
                "method": "userrecaptcha",
                "googlekey": captcha_info["sitekey"],
                "pageurl": captcha_info["page_url"],
                "version": "v3",
                "action": "verify",
            })
        elif captcha_info["type"] == "turnstile":
            params.update({
                "method": "turnstile",
                "key": captcha_info["sitekey"],
                "pageurl": captcha_info["page_url"],
            })

        # Submit
        resp = requests.post(f"{self.BASE_URL}/in.php", data=params)
        data = resp.json()
        if data["status"] != 1:
            raise Exception(f"Submit: {data['request']}")

        task_id = data["request"]

        # Poll
        for _ in range(60):
            time.sleep(5)
            resp = requests.get(f"{self.BASE_URL}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1,
            })
            data = resp.json()
            if data["request"] == "CAPCHA_NOT_READY":
                continue
            if data["status"] != 1:
                raise Exception(f"Solve: {data['request']}")
            return data["request"]

        raise Exception("Timeout")

    def inject_token(self, captcha_type, token):
        """Inject solved token into the page."""
        if captcha_type in ("recaptcha_v2", "recaptcha_v3"):
            self.driver.execute_script(f"""
                document.querySelector('#g-recaptcha-response').value = '{token}';
                // Try hidden textareas in iframes
                document.querySelectorAll('[name="g-recaptcha-response"]').forEach(
                    el => el.value = '{token}'
                );
            """)
        elif captcha_type == "turnstile":
            self.driver.execute_script(f"""
                const input = document.querySelector('[name="cf-turnstile-response"]');
                if (input) input.value = '{token}';
            """)

    def close(self):
        if self.driver:
            self.driver.quit()

判断依据是 URL 特征：recaptcha/api2/anchor 里的 k= 就是 sitekey。轮询 5 秒一次，Turnstile 通常 10 秒内返回，reCAPTCHA v2 在 60 秒内。

端到端流程：识别并提交表单

driver.get() 之后留 3 秒：验证码脚本异步加载，读早了请求列表是空的。

def automate_login():
    solver = SeleniumWireCaptchaSolver(
        api_key="YOUR_API_KEY",
        proxy="http://user:[email protected]:8080"
    )

    try:
        driver = solver.create_driver(headless=True)

        # Navigate and wait for CAPTCHA resources to load
        driver.get("https://staging.example.com/qa-login")
        time.sleep(3)

        # Fill form
        driver.find_element("id", "email").send_keys("[email protected]")
        driver.find_element("id", "password").send_keys("password123")

        # Detect CAPTCHA from network requests
        captcha = solver.detect_captcha_from_requests()

        if captcha:
            print(f"Found {captcha['type']} (sitekey: {captcha['sitekey'][:20]}...)")
            token = solver.solve_captcha(captcha)
            solver.inject_token(captcha["type"], token)
            print(f"Token injected: {token[:50]}...")

        # Submit
        driver.find_element("id", "submit").click()
        time.sleep(3)

        print(f"Current URL: {driver.current_url}")

    finally:
        solver.close()

automate_login()

拦截器的常用写法

补充自定义请求头

def add_headers_interceptor(request):
    request.headers["X-Custom-Header"] = "value"
    request.headers["Accept-Language"] = "en-US,en;q=0.9"

driver = solver.create_driver()
driver.request_interceptor = add_headers_interceptor

拦掉统计与广告脚本

def block_trackers(request):
    blocked = ["analytics", "tracking", "ads", "facebook.net", "doubleclick"]
    if any(b in request.url for b in blocked):
        request.abort()

driver.request_interceptor = block_trackers

改写响应内容

def modify_response(request, response):
    if "captcha-config" in request.url:
        # Modify CAPTCHA configuration response
        import json
        body = json.loads(response.body.decode("utf-8"))
        body["difficulty"] = "easy"
        response.body = json.dumps(body).encode("utf-8")

driver.response_interceptor = modify_response

从 AJAX 响应里取验证码参数

有些站点加载完才下发 sitekey，DOM 解析必然扑空：

def extract_captcha_from_api(driver, timeout=10):
    """Watch network for CAPTCHA configuration API calls."""
    start = time.time()

    while time.time() - start < timeout:
        for request in driver.requests:
            if request.response and "captcha" in request.url.lower():
                try:
                    data = json.loads(request.response.body.decode())
                    if "sitekey" in str(data) or "siteKey" in str(data):
                        return data
                except (json.JSONDecodeError, UnicodeDecodeError):
                    pass
        time.sleep(0.5)

    return None

多代理出口：认证代理与节点选择

带认证的代理必须写进 seleniumwire_options，--proxy-server 不支持用户名密码。

from seleniumwire import webdriver

proxies = [
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
]

current_proxy = [0]

def rotate_proxy(request):
    proxy = proxies[current_proxy[0] % len(proxies)]
    request.headers["Proxy-Authorization"] = None  # Reset
    current_proxy[0] += 1

wire_options = {
    "proxy": {
        "http": proxies[0],
        "https": proxies[0],
    }
}

driver = webdriver.Chrome(seleniumwire_options=wire_options)

统计验证码相关流量

def analyze_captcha_traffic(driver):
    """Analyze all CAPTCHA-related network traffic."""
    captcha_requests = []

    for request in driver.requests:
        if any(k in request.url for k in ["recaptcha", "turnstile", "hcaptcha", "geetest"]):
            captcha_requests.append({
                "url": request.url,
                "method": request.method,
                "status": request.response.status_code if request.response else None,
                "content_type": request.response.headers.get("Content-Type") if request.response else None,
                "size": len(request.response.body) if request.response and request.response.body else 0,
            })

    print(f"\nCAPTCHA Network Traffic ({len(captcha_requests)} requests):")
    for req in captcha_requests:
        print(f"  [{req['status']}] {req['method']} {req['url'][:80]}...")

    return captcha_requests

导出 HAR 复盘失败请求

导出 HAR 丢进 Chrome DevTools 回放，请求头、状态码和时序一目了然。

from seleniumwire.utils import decode

def export_har(driver, filename="captcha_traffic.har"):
    """Export HAR file for debugging CAPTCHA issues."""
    import json

    har = {
        "log": {
            "version": "1.2",
            "entries": []
        }
    }

    for request in driver.requests:
        if not request.response:
            continue

        entry = {
            "request": {
                "method": request.method,
                "url": request.url,
                "headers": [{"name": k, "value": v} for k, v in request.headers.items()],
            },
            "response": {
                "status": request.response.status_code,
                "headers": [{"name": k, "value": v} for k, v in request.response.headers.items()],
            },
        }
        har["log"]["entries"].append(entry)

    with open(filename, "w") as f:
        json.dump(har, f, indent=2)

    print(f"HAR exported: {filename} ({len(har['log']['entries'])} entries)")

常见故障与处理方式

现象	原因	处理方式
`seleniumwire` 导入报错	没装依赖	`pip install seleniumwire`
`driver.requests` 为空	读得太早	导航后加 `time.sleep(3)`
代理下 SSL 报错	证书校验失败	关掉 `verify_ssl`
内存持续增长	请求堆在内存里	定期 `del driver.requests`
页面明显变慢	全部流量过代理	配 `exclude_hosts`
识别不到验证码	在 iframe 内加载	按 URL 关键字查请求
reCAPTCHA 资源超时	脚本由 Google 托管	换可直连的 CI 环境

常见问题

Selenium Wire 会拖慢脚本吗？

会有开销：流量都过它的本地代理，请求还留在内存里。长任务用 exclude_hosts 排除静态资源，并定期清理请求列表。

抓不到 reCAPTCHA 请求怎么办？

多半是读得太早，先等 iframe 出现。跨域 iframe 里的验证码要按 URL 关键字找，别按当前域名过滤。

并发跑多个浏览器实例该买哪个套餐？

CaptchaAI 按并发线程计费，套餐内识别次数不限。调试用 BASIC（$15/月，5 线程），十几路并发选 STANDARD（$30/月，15 线程），再往上看 ADVANCE（$90/月，50 线程）。

Selenium Wire + CaptchaAI：请求拦截以解决验证码

原生 Selenium 缺的是网络可见性

安装依赖

核心类：识别验证码并调用 CaptchaAI

端到端流程：识别并提交表单

拦截器的常用写法

补充自定义请求头

拦掉统计与广告脚本

改写响应内容

从 AJAX 响应里取验证码参数

多代理出口：认证代理与节点选择

统计验证码相关流量

导出 HAR 复盘失败请求

常见故障与处理方式

常见问题

Selenium Wire 会拖慢脚本吗？

抓不到 reCAPTCHA 请求怎么办？

并发跑多个浏览器实例该买哪个套餐？

相关阅读

使用验证码处理自动提交表单

注册流程测试中的验证码处理

持续集成测试中的验证码处理

Web 表单中 API 端点测试的 CAPTCHA 解决方案

Bright Data + CaptchaAI：完整的代理集成指南

使用验证码解决进行学术研究网络抓取

原生 Selenium 缺的是网络可见性

安装依赖

核心类：识别验证码并调用 CaptchaAI

端到端流程：识别并提交表单

拦截器的常用写法

补充自定义请求头

拦掉统计与广告脚本

改写响应内容

从 AJAX 响应里取验证码参数

多代理出口：认证代理与节点选择

统计验证码相关流量

导出 HAR 复盘失败请求

常见故障与处理方式

常见问题

Selenium Wire 会拖慢脚本吗？

抓不到 reCAPTCHA 请求怎么办？

并发跑多个浏览器实例该买哪个套餐？

相关阅读

相关文章

使用验证码处理自动提交表单

注册流程测试中的验证码处理

持续集成测试中的验证码处理

Web 表单中 API 端点测试的 CAPTCHA 解决方案

Bright Data + CaptchaAI：完整的代理集成指南

使用验证码解决进行学术研究网络抓取