如何在 Python 中识别 Capy Puzzle:从 key 到表单提交

solve capy puzzle in python - How to Solve Capy Puzzle in Python, From Key to Form Post

要在 Python 中识别 Capy Puzzle,先从页面上读出站点的 PUZZLE_ key,用这个 key 和页面 URL 调用 capskip 包里的 solver.capy(),然后把它返回的三个值分别填入 capy_captchakey、capy_challengekey 和 capy_answer 表单字段,一起提交,并且立刻提交。获取页面和发送表单要走同一个 requests 会话,这样站点的 cookie 才会随答案一起发回去。让人意外的是结果的形态。大多数验证码类型只返回一个 token;Capy 返回的是三个值,它们只能作为一组使用,而且其中一个很快就会过期。CapSkip 从 1.4.0 版开始支持 Capy Puzzle。本指南讲这几件事:key、调用、有意设置的两秒延迟、提交、重试,以及同时运行大量识别。

你需要什么

  • 在 Windows 机器上运行的 CapSkip 1.4.0 或更高版本。Capy 支持就是在该版本中加入的,同时加入的还有 CaptchaFox 和 Friendly Captcha。
  • Python 3.10 或更高版本,以及 capskip 包 1.3.0 或更高版本,这是第一个带有 capy() 方法的版本。示例还用到了 requests,讲并行识别的那一节用的是 httpx。
  • 目标页面上的两个值:以 PUZZLE_ 开头的 Capy key,以及小组件所在页面的 URL。第 1 步会告诉你 key 在哪里,从同一个地方还能看出你是否需要第三个可选值。
  • 识别工具的地址。在 Local 模式下,CapSkip 只在 127.0.0.1 上响应,仅供本机使用;在 Server 模式下,它监听你的网络地址或公网 IP,这样另一台机器上的脚本就能通过 API 调用它。两种模式的设置位置都是 连接设置中,下文有一节会说明何时该切换。
# Quoted, so cmd.exe does not read >= as a redirect
pip install "capskip>=1.3.0" requests httpx

第 1 步:找到 PUZZLE_ key 和 Capy 主机

你需要的一切都来自目标页面,而且 key 是公开的,对每个访问者都一样。站点会把它放在两个地方。小组件脚本 URL 里的 k 参数,就在 requests 下载到的 HTML 中。小组件在浏览器里运行过之后,key 还会出现在小组件写入表单的一个隐藏 capy_captchakey input 里,DevTools 里看到的就是这个位置。用一个匹配 PUZZLE_ 前缀的正则表达式,两处都能找到它。

import re

import requests

PAGE_URL = "https://example.com/login"
session = requests.Session()
html = session.get(PAGE_URL, timeout=30).text

# The widget script's k= parameter. The capy_captchakey input only
# exists once the widget has run in a browser.
key = re.search(r"PUZZLE_[A-Za-z0-9_-]+", html)

# The widget script's own host; None means the default one.
host = re.search(r"(https://[^\"'\s<>]+?)/puzzle/get_js/", html)
api_server = host.group(1) if host else None
print(key and key.group(0), api_server)

看 script 标签的时候,顺便记下它的主机。script URL 中 /puzzle/get_js/ 之前的部分,就是这个 key 背后的 Capy API,CapSkip 把它叫作 api_server。它的默认值是 https://jp.api.capy.me,也就是线上服务实际运行的地方,所以只有当页面从别处加载小组件时,你才需要它。有个坑来自其他识别工具的文档:其中好几家仍然写着没有地区前缀的 api.capy.me,而这个主机已经无法解析了。如果旧示例里设置了它,就删掉这个选项。

获取页面时特意用了 requests.Session。你在第 3 步提交的表单,往往依赖页面设置的一个会话 cookie,而 session 对象不用任何额外代码就会把它发回去。

第 2 步:调用 capy()

在 Python 中识别 Capy Puzzle 只需要一个方法。它接受 key 和页面 URL,外加可选的关键字参数;对于使用默认主机的页面,有这两个值就够了。

# pip install "capskip>=1.3.0"
from capskip import CapSkip

solver = CapSkip(host="127.0.0.1", port=8080)

result = solver.capy("PUZZLE_YOUR_KEY", "https://example.com/login")

print(result["captchakey"])     # goes in capy_captchakey
print(result["challengekey"])   # goes in capy_challengekey
print(result["answer"])         # goes in capy_answer

返回结果是一个普通的 dict。读取那三个具名的键即可。code 键里是服务器返回的完整答案对象,一个包含全部四个字段的 dict,方便记日志;respKey 是一个空字符串,只是为了与其他服务兼容而存在。答案是一个很长的字符串,开头大致像 0xax8ex0xax84x 这样:它是小组件在拼图块移动时本会记录下来的拖动轨迹,而不是一个坐标。

该方法接受的选项

请求发出之前,capy() 方法会丢弃所有值为 None 的关键字参数,所以你可以在每次调用时都传入第 1 步拿到的主机,让正则来决定它是否被发送。

result = solver.capy(
    key.group(0),
    PAGE_URL,
    # None (no get_js script on the page) is dropped, and CapSkip
    # then uses its default host.
    api_server=api_server,
    # Poll every half second; see the next section for why.
    polling_interval=0.5,
)

除了 api_server,capy() 还接受 proxy、proxytype 和 useragent,另外还有以秒为单位的单次调用 timeout 和 polling_interval。如果你设置了 user agent,它会随 CapSkip 拉取拼图时发出的那唯一一个请求一起发送,不过你很少会用到它。还有一个 version 选项,但只接受 puzzle。Capy 的另一个系列 avatar 是另一种挑战,背后是另一个接口,所以 SDK 会用 ValidationException 拒绝它,而不是返回一个站点会拒收的答案。未知的关键字、为空的 key 或页面 URL,或者 HTTP、HTTPS、SOCKS5、SOCKS5H 之外的代理类型,都会在发送任何内容之前抛出同样的异常。

为什么一次 Capy 识别大约要两秒

检测本身很快。CapSkip 拉取拼图图片,用一些像素运算找到缺口,再生成拖动轨迹,整个过程不涉及浏览器,也不涉及模型。然后它会有意地等一等。

Capy 会测量从下发拼图到收到答案之间的时间,凡是比真人拖动拼图块还快的答案,一律拒绝。CapSkip 自己针对 Capy 做的测试显示,这个下限大约是一秒;而且这种拒绝带的提示和答错时一模一样,所以一个正确但送得太快的识别结果,看起来和识别工具坏了没有任何区别。因此,CapSkip 会把每个 Capy 结果压住,直到拉取拼图满两秒之后才放行。这段等待只是 sleep,所以只增加延迟,不消耗 CPU。

为什么在默认设置下一次调用要接近四秒:SDK 一把任务发给 CapSkip 就立刻轮询一次,四分之一秒后再轮询一次,之后不断把间隔翻倍,直到达到它的 pollingInterval,默认是 5 秒。各次轮询落在 0、0.25、0.75、1.75 和 3.75 秒,所以两秒时就绪的答案要到第五次轮询才被取回。如果在调用时传入 polling_interval=0.5,或者在客户端识别的主要是 Capy 时在构造函数里设置 pollingInterval=0.5,调用会在任务发出后两秒多一点返回,我们测试时大约是 2.3 秒。提交表单前不要自己再加延迟,也不要想办法把这段等待省掉:答案能通过,靠的正是它。

第 3 步:在一个请求中提交全部三个值

这三个值要填进小组件本来会自己写入表单的那几个字段里。把它们一起发送,和表单的其余内容放在同一个请求里,并且通过获取页面的那个 session 发送。

# session is the one Step 1 used, so the page's cookies go back too.
resp = session.post(
    PAGE_URL,  # or wherever the form's action attribute points
    data={
        "username": "YOUR_USERNAME",
        "capy_captchakey": result["captchakey"],
        "capy_challengekey": result["challengekey"],
        "capy_answer": result["answer"],
    },
    timeout=30,
)
print(resp.status_code)

答案要按返回的原样提交。站点后端会拿它和当时下发的拼图进行比对,所以截断它、重新拼装它,或者以任何方式整理它,都会让它失效。requests 库会对表单请求体做百分号编码,这没有问题,因为服务器在比对之前会先解码。然后要尽快提交。CapSkip 每次识别都会生成一个新的挑战 key,拼图就绑定在它上面,所以这个 key 只能用一次,而且寿命很短,一次识别只对应一次提交。

照搬真实表单发送的内容。大多数页面带有自己的隐藏字段,比如防伪造 token;有些页面则通过脚本用 JSON 请求体提交,而不是表单提交。打开 DevTools 手动提交一次,把请求复制下来,然后把三个 Capy 值放到页面放它们的位置。

第 4 步:识别失败时重试

只要在 Python 中识别 Capy Puzzle 有一定的量,就总会有几次识别失败。失败的识别会以 ApiException 的形式返回,其消息中包含 ERROR_CAPTCHA_UNSOLVABLE;而通过 SDK 轮询的接口 res.php,这一个错误码涵盖了两种不同的情况。区分它们的依据是出现的频率。

发生了什么表现该怎么做
CapSkip 没有定位到缺口偶尔出现,下一次尝试通常就会成功重试。每次尝试都会拿到一张基于不同照片的全新拼图
Capy API 拒绝了这个 key这一个 key 每次尝试都失败,CapSkip 里的 Capy 任务列表显示 Invalid captcha key检查 key 和 api_server。CapSkip 从不重试被拒的 key,因为它会以同样的方式再被拒一次

失手很少见,所以重试一次几乎就能覆盖所有情况。你可以在 CapSkip 设置的 Capy 部分设置 Retries,它默认为 0,每个任务最多允许重试三次;也可以在自己的代码里重试:

from capskip import ApiException


def solve_capy(key, url, attempts=3, **options):
    for attempt in range(1, attempts + 1):
        try:
            return solver.capy(key, url, **options)
        except ApiException as exc:
            # A missed hole reads as unsolvable, and the next try
            # draws a new puzzle. Anything else is final.
            if "UNSOLVABLE" not in str(exc) or attempt == attempts:
                raise

capy() 方法按客户端的 defaultTimeout 轮询,即 120 秒,因为一次 Capy 识别只是一次拉取加上一些运算,而不是一个浏览器会话。CapSkip 自己也有一套计时:在 Capy 部分,一个任务最多可以等待 250 秒(Wait Timeout)来获得 10 个线程(Max. Threads)中的一个,而单次尝试有 60 秒(Row Timeout)。一次正常的识别,在这些时限内都绰绰有余。

同时识别大量 Capy 拼图

在 Python 中,AsyncCapSkip 是真正的 asyncio 客户端,而不是一个别名,所以 asyncio.gather 能让多个识别并行运行。要让每次识别都和它自己的提交配对。先收齐一百个识别结果、之后再统一提交,会让最早拿到的挑战 key 在等最后几次识别完成时不断变旧。

import asyncio

import httpx
from capskip import AsyncCapSkip

solver = AsyncCapSkip(host="127.0.0.1", port=8080, pollingInterval=0.5)
slots = asyncio.Semaphore(10)  # CapSkip's Capy Max. Threads


async def solve_and_submit(key, url, form):
    # One client per job, so each form session keeps its own cookies.
    async with slots, httpx.AsyncClient(
            timeout=30, follow_redirects=True) as client:
        await client.get(url)
        result = await solver.capy(key, url)
        return await client.post(url, data={
            **form,
            "capy_captchakey": result["captchakey"],
            "capy_challengekey": result["challengekey"],
            "capy_answer": result["answer"],
        })


async def main(jobs):
    return await asyncio.gather(
        *(solve_and_submit(*job) for job in jobs), return_exceptions=True)

这个信号量的容量与 CapSkip 的线程数一致,所以不会有任务一边在 CapSkip 的队列里排着,一边占着一个会话不放;它来自 Python 标准库,文档条目是 asyncio.Semaphore。量大时还要加代理。每次识别都会从 Capy API 拉取一张新拼图,而从同一个地址源源不断地拉取,正是限流机制要抓的那种模式。在 CapSkip 的 Capy 部分配置代理池,或者在每次调用时传入一个带 type 和 uri 的 proxy 字典;代理只作用于拉取拼图这一步,也就是一次 Capy 识别唯一会发出的请求。更多 asyncio 用法见 在 Python 中并行识别验证码的指南.

把识别工具跑在另一台机器上

只要脚本和 CapSkip 在同一台 Windows 电脑上,127.0.0.1 就是对的。一旦 Python 代码挪到 VPS、容器、CI runner 或托管 notebook 上,回环地址就指向了错误的机器,第一次识别就会抛出 NetworkException。把 CapSkip 切换到 Server 模式,它就会监听你的网络地址或公网 IP,上面这些环境都能通过同一套 API 调用它。如果链路要经过公网,请使用静态公网 IP,打开 API 密钥校验,并用一条 Windows Firewall 规则把端口限制在你预期的地址上。识别工具仍然是你自己的 Windows 机器,识别也仍然不计量。

SDK 不会自己读取环境变量。像完整示例那样,在你的代码里读取 CAPSKIP_HOST、CAPSKIP_PORT 和 CAPSKIP_API_KEY 并传给构造函数,这样同一个脚本在你的桌面机和服务器上都能运行。

完整可运行示例

# pip install "capskip>=1.3.0" requests
import os
import re

import requests
from capskip import (ApiException, CapSkip, CapSkipError,
                     TimeoutException, ValidationException)

PAGE_URL = "https://example.com/login"

solver = CapSkip(
    apiKey=os.getenv("CAPSKIP_API_KEY", "capskip"),
    host=os.getenv("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.getenv("CAPSKIP_PORT", "8080")),
    # Capy answers after a two second hold; poll often enough to catch it.
    pollingInterval=0.5,
)
session = requests.Session()

html = session.get(PAGE_URL, timeout=30).text
key = re.search(r"PUZZLE_[A-Za-z0-9_-]+", html)
if not key:
    raise SystemExit("No PUZZLE_ key in the HTML; find it in DevTools.")
host = re.search(r"(https://[^\"'\s<>]+?)/puzzle/get_js/", html)

try:
    for attempt in range(1, 4):
        try:
            result = solver.capy(key.group(0), PAGE_URL,
                                 api_server=host.group(1) if host else None)
            break
        except ApiException as exc:
            # A missed hole reads as unsolvable; the next try draws a new puzzle.
            if "UNSOLVABLE" not in str(exc) or attempt == 3:
                raise
            print(f"attempt {attempt}: {exc}")
except ValidationException as exc:
    raise SystemExit(f"not sent: {exc}")
except TimeoutException:
    raise SystemExit("gave up waiting; defaultTimeout is 120 seconds")
except CapSkipError as exc:
    # A third miss, a refused key, or CapSkip unreachable.
    raise SystemExit(f"solve failed: {exc!r}")

# All three together, straight away: the challenge key is single-use.
resp = session.post(
    PAGE_URL,  # or wherever the form's action attribute points
    data={
        "username": "YOUR_USERNAME",
        "capy_captchakey": result["captchakey"],
        "capy_challengekey": result["challengekey"],
        "capy_answer": result["answer"],
    },
    timeout=30,
)
print(resp.status_code)

如果三次尝试全部失败,先检查 key,因为被拒的 key 每次都以同样的方式失败,而失手几乎不会连续发生三次。如果页面是用打包后的脚本构建小组件的,正则什么都找不到,就打开 Network 标签页,从小组件自己发出的请求里复制 key 和主机。这个方法背后的原始接口,详见 API 参考文档的 Capy 部分.

常见错误及其含义

你所看到的原因修复
capy() 返回了全部三个值,站点却拒绝了提交答案被改动过,或者只有其中一两个值到达了表单把 result["answer"] 原样连同另外两个值,放在一个请求里提交
站点拒绝了提交,并提示会话已过期页面是用 requests.get() 获取的,表单却用了一个新的请求,所以页面的 cookie 始终没有发回去通过同一个 requests.Session 获取页面并提交表单
第一次成功的提交,第二次却失败挑战 key 只能用一次,而且寿命很短每次提交都重新识别,并且识别完立刻提交
某个 key 每次尝试都抛出 ApiExceptionCapy API 拒绝了这个 key,或者 api_server 指向了错误的主机重新复制 key,连前缀一起,并检查 script 标签的主机
偶尔出现包含 ERROR_CAPTCHA_UNSOLVABLE 的 ApiExceptionCapSkip 没能在那张拼图里定位到缺口重试;下一次尝试会拿到另一张拼图
从其他服务复制示例之后,每次识别都失败示例把 api_server 设成了 api.capy.me,而这个主机已经无法解析删掉这个选项,使用默认主机
还没发送任何内容就抛出 ValidationExceptionkey 或页面 URL 为空、version 设成了 avatar、代理类型不是 HTTP、HTTPS、SOCKS5 或 SOCKS5H,或者传了这个方法不接受的关键字确认 key 不是空字符串,并修正或去掉错误信息中指出的那个参数
抛出带有 ERROR_WRONG_USER_KEY 或 ERROR_KEY_DOES_NOT_EXIST 的 ApiExceptionCapSkip 打开了 API 密钥校验,而脚本发送的是默认密钥,或者根本没有发送把 CAPSKIP_API_KEY 设为 CapSkip 中配置好的密钥
高并行负载下抛出 TimeoutException排队的任务多于 10 个线程在 120 秒内能处理完的数量,或者 CapSkip 在识别中途重启了用信号量限制并发数,或者调高 defaultTimeout
第一次调用时抛出 NetworkExceptionCapSkip 没有在运行,或者主机和端口不对启动 CapSkip,然后确认它应该处于 Local 模式还是 Server 模式

常见问题

为什么 capy() 返回的是三个值而不是一个 token?

因为 Capy 表单提交的就是这些。整个过程中,从来没有哪个服务器签发过 token。小组件自己生成挑战 key,拉取属于它的拼图,再记录拖动过程;之后站点用自己的私钥,把挑战 key 和答案发给 Capy 进行核验。CapSkip 扮演的是小组件的角色,所以它返回的,就是小组件本来会写进表单的内容。

python-requests 的 user agent 会导致答案被拒吗?

Capy 不会因此拒绝。Capy 的答案不绑定浏览器,CapSkip 不会为它返回 user agent,而你可以传入的那个可选 user agent,也只影响拉取拼图的那个请求。不过站点本身可能出于自己的原因,拒绝自报为 python-requests 的请求。如果遇到这种情况,就在 session 上设置一次浏览器的 User-Agent,之后从它发出的每个请求,包括提交在内,都会带上这个请求头。

能在 Selenium 或 Playwright 会话中使用这个答案吗?

可以。用浏览器当前打开的页面 URL 进行识别,然后用一小段脚本把三个值写进表单的 capy_captchakey、capy_challengekey 和 capy_answer 隐藏 input 中,小组件还没创建的 input 就自己补上,再按页面平时的方式提交表单。不要同时在浏览器里拖动拼图块,因为那会产生第二个不同的答案。如果浏览器只是为了过这道拼图,上面的 requests 路线更简单,也更快。

托管平台上的 Python 脚本能连到识别工具吗?

可以。在连接设置里把 CapSkip 切换到 Server 模式,让它监听网络地址而不是回环地址,然后在你的脚本里从 CAPSKIP_HOST 读取这个地址,并传给 CapSkip()。VPS、容器宿主机、CI runner 和托管的 notebook 都通过同一套 HTTP API 连接。如果链路要经过公网,请使用静态公网 IP 并配上防火墙规则。识别工具始终运行在你自己的硬件上,因此识别次数的计算方式没有任何变化。同样的流程在 .NET 中怎么写,见 C# Capy Puzzle 指南.

简短版结论

要在 Python 中识别 Capy Puzzle,先用 requests 会话获取页面,用一个正则取出 PUZZLE_ key,并记下小组件脚本的主机,以防它不是默认主机。用 key 和真实的页面 URL 调用 solver.capy(),让它花上那几秒钟,设了 polling_interval=0.5 时大约是 2.3 秒。通过同一个 session,把 captchakey、challengekey 和 answer 放进三个 capy_ 字段,原样、一起、立刻提交。偶尔失手就重试,用量增长时加上代理,一旦脚本离开识别工具所在的机器,就切换到 Server 模式。

关于重试,最后再说一点。每次尝试都会拿到一张不同的拼图,所以遇到失手,再试一次就是正确的解决办法;而当 无限量验证码识别工具 就在你自己的机器上时,第二次尝试只多花几秒钟,而不是再多一次计费的识别。