如何在 nodriver 中用异步 SDK 识别验证码

nodriver 里的验证码环节还是老三步:从页面上读出 sitekey,发给识别工具,再用 JavaScript 把 token 写回去。不一样的是 nodriver 从头到尾都是异步的。它通过一条 asyncio websocket 驱动 Chrome,所以一次阻塞式识别不只是让你的脚本干等,它会把承载所有 DevTools 消息的那条 socket 一起卡住。配上异步客户端,整个流程才能保持响应。
你需要什么
- Python 3.10 或更高版本,已安装 nodriver 0.50 和 CapSkip SDK。
- 脚本运行的那台机器上装有 Chrome、Chromium、Edge 或 Brave。nodriver 会直接把它启动起来。
- 受保护表单的页面 URL。sitekey 在运行时读取。
- 如果脚本和识别工具在同一台机器上,就让 CapSkip 以 Local 模式运行;如果不在同一台机器上,就用 Server 模式。两种模式都写在 连接设置.
# Both packages, one line. pip install nodriver capskip
nodriver 为什么会改变这件事的做法
nodriver 是 undetected-chromedriver 的官方接班人,作者是同一个人,它的卖点是整条链路里既没有 webdriver,也没有 Selenium。它直接用 DevTools Protocol 跟自己启动的浏览器通信。没有 chromedriver 二进制要打补丁,也没有驱动版本需要跟着 Chrome 走。
对验证码这件事真正重要的是这句话的后半段:它是完全异步的。连接是一条由 asyncio 管理的 websocket,有一个后台任务负责从上面读协议消息。每一次元素查找、每一次跳转、每一个事件处理器,都依赖这个任务被调度到。你要是在中间调用一个同步的识别工具,那么在整个识别期间进程里别的东西都不会跑,而 reCAPTCHA v2 动辄就是十五到四十五秒。
所以这个框架的规则很短:用异步客户端,并且 await 它。
有一点得说清楚:不用 webdriver 不等于不会被检测。去掉驱动只去掉了一个信号,你指纹里的其余部分原封不动。识别挑战和让自己看起来像一个真实浏览器是两件事,本文只讲前者。
第 1 步:从页面上读出 sitekey
sitekey 在宿主文档上,不在小组件的 iframe 里。Google 的标记把它作为 data-sitekey 属性放在一个容器上,而 nodriver 的 select 方法能用 CSS 选择器找到这个容器。注意最后一行的方括号取值,绊倒大家的就是它。
# pip install nodriver
import nodriver as uc
async def main():
browser = await uc.start()
page = await browser.get("https://example.com/page-with-recaptcha")
# select() retries for 10 seconds by default, so it doubles
# as a wait condition for a widget that renders late.
holder = await page.select("div.g-recaptcha")
sitekey = holder.attrs["data-sitekey"]
print(sitekey) # 6Lxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
uc.loop().run_until_complete(main())属性名在元素上完全按照 HTML 里的写法保存,连字符也照留,所以方括号取值是唯一可靠的读法。带点的简写看起来该能用,实际上却悄悄失效:向元素要 data_sitekey 会拿回一个 None 而不是抛错,因为这次查找会落到一个默认值上。这个 None 随后以空 key 的形式传到识别工具,很久之后才以 ERROR_GOOGLEKEY的形式浮出水面,离引发它的那一行已经很远了。真正需要知道的重命名只有一个:class 属性存在 class_ 下面,好避开 Python 关键字。
有些站点根本不在宿主页面上暴露 sitekey,只在小组件 iframe 的 URL 里传。那就改从查询字符串里读。
# Fallback: the k= parameter on the anchor iframe.
from urllib.parse import urlparse, parse_qs
frame = await page.select("iframe[src*='recaptcha/api2/anchor']")
sitekey = parse_qs(urlparse(frame.attrs["src"]).query)["k"][0]第 2 步:识别它,同时别把 socket 卡住
Python SDK 提供两个客户端。CapSkip 是同步的,AsyncCapSkip 是真正的 asyncio 实现,而不是一个别名,这正是这个框架需要的。两者都跟你自己机器 8080 端口上的识别工具通信,也都不按次计费。
# pip install capskip from capskip import AsyncCapSkip solver = AsyncCapSkip(host="127.0.0.1", port=8080) # Same call shape for v3 (version="v3") and Enterprise # (enterprise=1). Invisible v2 takes invisible=1. result = await solver.recaptcha(sitekey=sitekey, url=PAGE_URL) token = result["code"] # the g-recaptcha-response value
因为这个调用是可 await 的,多个标签页可以同时识别,完全不用线程。先把页面打开,再把这些识别一起 gather,然后把每个 token 注入到它所属的标签页里。更完整的模式,包括 SDK 如何让轮询逐步退避而不是按固定间隔 sleep,见 并行识别验证码.
# Three tabs, three solves, one wait.
import asyncio
results = await asyncio.gather(*[
solver.recaptcha(sitekey=k, url=u) for k, u in targets
])Turnstile 和极验(GeeTest)各有自己的方法,两者的调用形态和上面那次一样。各自的完整参数列表见 CapSkip API 文档.
第 3 步:注入 token 并提交
响应用的 textarea 被 display:none 藏起来了,所以在任何自动化工具里都不可能往里打字。你得用 JavaScript 写。nodriver 的 evaluate 方法只接受一个表达式字符串,没法在它旁边再传参数进去,所以 token 必须嵌进那个字符串里,而安全的做法是用 json.dumps,不是 f-string。一个 JSON 字符串字面量就是一个合法的 JavaScript 字符串字面量,引号和转义都算在内。
# json.dumps gives a correctly quoted JS string literal.
import json
await page.evaluate(
"document.getElementById('g-recaptcha-response').value = "
+ json.dumps(token)
)
# Then submit the form the way the page expects.
button = await page.select("button[type=submit]")
await button.click()确认这个值确实写进去了,这件事本身还有一个坑,值得你花十秒钟注意一下。设了 return_by_value 之后,evaluate 只在值为真时才把普通的 Python 值交还给你。空字符串或者 0 会漏过去,你拿到的是一个协议对象。所以别去读长度再判断,因为长度为 0 恰恰就是你想检测的那种情况。要返回一个永远不可能为假的东西。
# String() keeps a zero-length answer truthy, so the check
# reports the real number instead of a protocol object.
length = await page.evaluate(
"String(document.getElementById('g-recaptcha-response').value.length)",
return_by_value=True,
)
print(length) # "0" means the injection did not land如果站点不是在提交时读 textarea,而是定义了一个回调,那就在设置完值之后调用它。函数名因站点而异,所以要从页面自己的标记里读出来,别靠猜。不管哪种方式,这都还是普通的 reCAPTCHA v2:回调改变的是你怎么把 token 交出去,而不是它怎么被识别出来。 reCAPTCHA v2 识别工具页面 把两种提交方式都讲了。
完整可运行示例
上面所有内容,合成一个脚本。识别工具只创建一次并复用,浏览器在 finally 块里关掉,这样一次失败的识别不会留下一个 Chrome 进程。
# pip install nodriver capskip
import json
import nodriver as uc
from capskip import AsyncCapSkip
PAGE_URL = "https://example.com/page-with-recaptcha"
async def main():
solver = AsyncCapSkip(host="127.0.0.1", port=8080)
browser = await uc.start()
try:
page = await browser.get(PAGE_URL)
holder = await page.select("div.g-recaptcha")
sitekey = holder.attrs["data-sitekey"]
if not sitekey:
raise RuntimeError("Widget found but data-sitekey was empty.")
result = await solver.recaptcha(sitekey=sitekey, url=PAGE_URL)
await page.evaluate(
"document.getElementById('g-recaptcha-response').value = "
+ json.dumps(result["code"])
)
button = await page.select("button[type=submit]")
await button.click()
await page.sleep(2)
print(page.target.url) # the page you land on after submitting
finally:
browser.stop()
uc.loop().run_until_complete(main())把识别工具跑在另一台机器上
nodriver 迟早会被搬到服务器上,它在那儿有两个需求:一个 Chromium 二进制,以及一个让 Chrome 有地方可画的桌面会话。无头模式默认是关的,所以一台没有桌面会话的服务器需要显式把无头打开。识别工具倒不必跟着一起搬。
CapSkip 有两种连接模式。Local 模式绑定到 127.0.0.1,只回应该设备本身,你在写脚本的阶段用它正合适。Server 模式绑定到你的内网或公网 IP,这样一台采集用的虚拟机、一台容器宿主机或者第二台工作站,就能通过 API 调用同一个识别工具。固定公网 IP 能让这个地址保持稳定。代码里除了你传进去的 host 之外什么都不用改,成本上也什么都不变,因为硬件还是你自己的。
# Same SDK, same call. Only the host moves. solver = AsyncCapSkip(host="10.0.0.12", port=8080, apiKey="YOUR_API_KEY")
识别工具一旦监听在网络地址上,就把密钥校验打开,并且给每台机器发各自的密钥,这样吊销其中一个不会动到其他机器。 设置指南 把两种模式都走了一遍。
做这件事的时候有一处命名冲突要分清。nodriver 自己的 start 函数也接受 host 和 port,但那两个描述的是你想接管的 Chrome 调试端点,不是识别工具。两个都传进去,nodriver 就干脆不启动浏览器了。识别工具的地址只属于客户端构造函数,不属于别的任何地方。
常见错误及其含义
| 你所看到的 | 原因 | 修复 |
|---|---|---|
| attrs 查找时抛 AttributeError | select() 什么都没找到,交还了 None | 放宽选择器,或者把 select 的超时调大 |
| sitekey 是 None,却完全没有报错 | 带点的访问方式够不着带连字符的属性 | 用方括号从 attrs 里读 |
| ERROR_GOOGLEKEY | 一个空的 sitekey 传到了识别工具 | 在花掉一次识别之前先检查这个值 |
| 整个识别期间脚本都卡住 | 同步客户端把事件循环阻塞了 | 改用 AsyncCapSkip,并 await 这个调用 |
| NetworkException | CapSkip 没在运行,或者 host 填错了 | 启动应用,或者把 host 指向服务器地址 |
| TimeoutException | 识别耗时超过了 recaptchaTimeout | 把它调到默认的 300 秒以上 |
| evaluate 返回的是一个对象,不是字符串 | 这个值是假值,所以没有走返回普通值那条路 | 把长度包进 String() 里,而不是包值本身 |
常见问题
如果我只识别一次,还能继续用同步客户端吗?
可以,而且在一个短脚本里你可能根本察觉不到。你换掉的是识别期间到达的每一条协议消息:跳转事件、加载事件,以及任何事件处理器在等的东西。跑得久了,这会表现为查找莫名其妙地超时。异步客户端的代价就是一个 import 和一个 await,实在没什么理由去做这笔交换。
我需要进到 reCAPTCHA 的 iframe 里吗?
不需要,而且这正是大家过度设计的地方。复选框确实在 iframe 里,但 sitekey 属性和隐藏的响应 textarea 都属于宿主文档。只有当站点不把 sitekey 放在页面上、你必须从 frame 自己的 URL 里读时,你才需要碰 frame。
我正从 undetected-chromedriver 迁移过来,哪些东西能沿用?
那三步原样沿用,因为它们本来就跟驱动无关:读出 sitekey,识别它,把 token 写进 textarea。沿用不了的是它们周围的 API,因为现在每个调用都是可 await 的,而且没有 driver 对象了。nodriver 还提供了一个助手,能把一个正在运行的 undetected-chromedriver 实例转成 browser 对象,让你可以分阶段迁移脚本。早先的那套做法可以参考 undetected-chromedriver 验证码教程.
我的爬虫跑在 VPS 上,识别工具该放哪?
放哪都行,只要两边能互相访问。Server 模式让识别工具监听在一个网络地址上,而不是回环地址,于是 VPS 就像调用任何内部服务那样通过 API 调它。把 host 参数指向那个地址,打开密钥校验,再给这台 VPS 发一个自己的密钥。识别工具不需要显示环境,考虑到浏览器需要,这一点挺方便。
简短版结论
用方括号从 attrs 里读出 sitekey,用 AsyncCapSkip 在 127.0.0.1:8080 上识别,通过 evaluate 配合 json.dumps 把 token 注入进去,然后提交。所有东西都要 await,因为一次同步识别会占住驱动浏览器的那条 socket。想了解 Python 这边更完整的图景,包括 Selenium 和 Playwright,见 Python 验证码识别页面.
在把抓取规模拉大之前,有一个后果值得说清楚。因为 CapSkip 是一个 无限量验证码识别工具 ,跑在你自己的硬件上,所以重试一千个页面的运行,和只重试十个页面的运行,花的钱完全一样。
