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 进行构建。