Integrations

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

Selenium Wire 扩展了 Selenium,使其能够拦截和修改 HTTP 请求,这是标准 Selenium 无法做到的。这解锁了强大的验证码解决工作流程:从 API 调用中提取站点密钥,将token 提交到拦截的响应中,以及通过代理路由请求。

本指南涵盖请求拦截、网络流量验证码检测、代理集成以及使用 CaptchaAI 进行自动解决。


Selenium Wire添加了什么

特征 标准硒 Selenium Wire
读取请求 是的
修改标题 有限的 完全控制
Intercept responses 是的
代理支持 基本的 身份验证代理支持
网络日志记录 仅限开发工具 Python 对象
哈尔出口 是的

设置

pip install seleniumwire selenium webdriver-manager requests

基本集成

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()

完整的工作流程

def automate_login():
    solver = SeleniumWireCaptchaSolver(
        api_key="YOUR_API_KEY",
        proxy="http://user:pass@proxy.example.com: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("user@example.com")
        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

从 API 调用中提取验证码参数

有些站点通过 AJAX 加载验证码参数。 Selenium Wire 捕获了这些:

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

每个请求的代理轮换

from seleniumwire import webdriver

proxies = [
    "http://user:pass@proxy1.example.com:8080",
    "http://user:pass@proxy2.example.com:8080",
    "http://user:pass@proxy3.example.com: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 导出

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
没有捕获请求 页面加载速度太快 导航后添加 time.sleep(3)
代理的 SSL 错误 证书问题 设置 seleniumwire_options['verify_ssl'] = False
内存泄漏 请求存储增长 定期调用del driver.requests
页面加载缓慢 所有流量均已代理 在线路选项中使用 exclude_hosts
未检测到验证码 通过 iframe 加载 检查 driver.requests 的 iframe URL

常问问题

Selenium Wire和标准Selenium Wire有什么区别?

Selenium Wire 包装 Selenium WebDriver 添加 HTTP 请求/response 拦截。它是一个直接替代品——相同的 API 和额外的网络功能。

我什么时候应该使用 Selenium Wire 和 Puppeteer?

对于需要请求拦截的Python项目使用Selenium Wire。如果您已经进入 JavaScript 生态系统,请使用 Puppeteer (Node.js)。

Selenium Wire 是否可以与未检测到的 chromedriver 一起使用?

是的。将 webdriver.Chrome 替换为 unDetected-chromedriver 中的 uc.Chrome,同时保留 Selenium Wire 的拦截功能。

请求拦截如何帮助验证码?

它可以让您从网络流量中检测验证码类型(比 DOM 解析更可靠),从 API 响应中提取参数,并将token 提交到拦截的请求中。


相关指南

  • Python Selenium + CaptchaAI 实用指南
  • Chrome DevTools 协议 + CaptchaAI
  • 代理质量和验证码解决率

验证码自动化的主请求拦截 -获取您的 CaptchaAI 密钥并使用 Selenium Wire 进行构建。

该文章已禁用评论。