如何在 Windmill 脚本中识别验证码（Python）

Windmill 里的验证码识别，大概是这类集成里最省事的一种。Windmill 会读取你脚本顶部的 import，把它们解析成 PyPI 包并固定在 lockfile 里，所以 CapSkip SDK 不需要安装步骤，也不需要 requirements 文件。剩下的就是一个 main 函数：传入 sitekey，返回 token。真正值得琢磨的不是代码，而是 worker 跑在哪台机器上，因为这决定了识别工具是留在回环地址上，还是必须监听你的网络。

你需要什么

• 一个 Windmill 实例，自托管或者用它们的云都行，外加一个你能部署脚本的 workspace。
• 在一台 Windows 机器上运行的 CapSkip。如果 worker 就跑在同一台机器上，用 Local mode（本地模式）就够了。如果 worker 在容器里或另一台主机上，就切到 Server mode（服务器模式）。
• 你要自动化的站点的 sitekey 和页面 URL。
• 一个存放识别工具密钥的 Windmill 变量，以及一个存放它地址的 worker 环境变量。

为什么那一行 import 就是全部的安装工作

保存脚本时，Windmill 会解析顶层的 import，算出它们对应哪些 PyPI 包，然后启动一个依赖任务写出 lockfile。这个 lockfile 会绑定到该版本的脚本上，所以你测过的那个部署，半年后跑起来还是同一个。没有 requirements 文件需要维护，也没有任何东西要手动装到 worker 上。

解释器也能在同一个地方固定：在脚本头部写一行注释即可。部署时没有指定版本的脚本会跑在 Python 3.11 上。

# py312
# pip install capskip - Windmill resolves this import itself
# and locks the version when the script is deployed.
from capskip import CapSkip

第 1 步：识别脚本

一个 Windmill 脚本就是一个 main 函数。它的参数会变成输入 schema 和 Windmill 渲染出来的表单，所以要写好类型标注。你返回的东西就是脚本结果，下游的流程步骤从那里读取。

# py312
# pip install capskip - resolved from this import on save.
import wmill
from capskip import CapSkip

def main(sitekey: str, page_url: str) -> str:
    # Host and port come from the worker environment. The key is
    # a Windmill variable, so it is stored encrypted and never
    # appears in the script body or in the run logs.
    solver = CapSkip(
        host=wmill.get_variable("u/admin/capskip_host"),
        port=8080,
        apiKey=wmill.get_variable("u/admin/capskip_key"),
    )

    result = solver.recaptcha(sitekey=sitekey, url=page_url)

    return result["code"]   # the token, for the next step

reCAPTCHA v2 的集成到这里就全了。其他每个变体都是同一个方法多加一个关键字参数：invisible 设为 1，enterprise 设为 1，或者 version 设为 v3 并带上 action 名称。Turnstile 和极验有各自的方法，形状完全一样，完整的参数列表见 CapSkip API 文档.

在你动手写轮询循环之前，关于 SDK 有两件事值得知道。它已经替你做了轮询，从 250 毫秒起步并逐步退避，而不是按固定间隔睡眠，这通常比裸 API 文档给出的等待时间更快。另外，它对 reCAPTCHA、Turnstile 和极验的上限是 300 秒，由 recaptchaTimeout 决定。这个数字在你于第 4 步设置脚本超时时很关键。

第 2 步：把密钥放在 Windmill 变量里

Windmill 有一等公民级的变量和密文，上面的脚本就是直接读取的。还有第二种更适合流程的做法：用引用语法把变量作为步骤参数传进去，Windmill 会在运行时以调用者的权限解析它。

这些引用会递归解析，列表里和嵌套对象里的也一样，所以一个接收密钥列表的步骤可以在每个元素里放一个引用。给这个 workspace 单独配一把识别工具密钥，别让你跑的所有东西共用一把。

第 3 步：worker 跑在哪里决定了连接模式

这才是真正决定整套配置的问题，而且很容易搞错，因为两种情况下脚本看起来一模一样。Windmill 的 worker 是一个自治进程，一次只跑一个脚本。它可能是数据库旁边的一个容器，可能是虚拟机上的一个进程，也可能是你自己桌面上的一个进程。不管是哪一种，SDK 调用都是从 worker 发起的一个 socket，所以识别工具必须能从那里访问到，别处都不算。

CapSkip 正是为此提供了两种连接模式。Local mode 绑定 127.0.0.1，只服务本机。Server mode 绑定你的网络地址或公网 IP，这样另一台机器、容器宿主机或托管平台就能通过 API 访问同一台 Windows 机器。两者都设置在 连接设置，而 Server mode 只改变识别工具运行的位置。它仍然是你自己的硬件，也仍然不计次。

Windmill 的 worker 确实可以跑在 Windows 上，而这正是让你能继续用回环地址的那种情况。这里有一个设置需要留意。PID 命名空间隔离在 Linux 上默认为 true，而 Windmill 自己的文档说，Windows worker 要把它设为 false。在 Windows worker 上把名为 ENABLE_UNSHARE_PID 的变量设为 false，它就能正常启动。

对于另外两行的情况，地址应该放在 worker 的环境里，而不是脚本里。Windmill 默认并不会把主机上的每个变量都交给任务，所以要在 worker 上的 WHITELIST_ENVS 变量里列出你需要的那些，用逗号分隔。worker 组也可以带自己的静态和动态环境变量，在 UI 里设置；当只有一部分 worker 靠近识别工具时，这种做法更整洁。

第 4 步：超时与重试

Windmill 在脚本的运行时设置里放了一个 Timeout 字段，就在 Cache 和 Concurrency 限制旁边。把它设得高于你最慢的一次识别，而不是低于。reCAPTCHA v2 的勾选框通常远不到一分钟就搞定，但 Turnstile 的挑战页面和极验要更久，而 SDK 最多会轮询 300 秒才放弃并抛出 TimeoutException。脚本超时如果低于这个值，一次慢识别就会变成一个被杀掉的任务，日志里什么有用的都没有。

一旦这个脚本成为流程里的一步，你就多了一层保障。Windmill 的流程步骤有两种重试形态，其中指数那种适合偶尔繁忙的识别工具。

指数形态的延迟等于乘数乘以底数的尝试次数次方，所以底数 3、乘数 2、一共五次尝试，等待时间会从 6 秒一路拉到 486 秒。另外还有一个 Continue on error 设置，它让流程在重试耗尽后继续往下走，并把错误当作该步骤的结果传下去，你就是靠它来搭一个失败时降级、而不是让整次运行失败的分支。

完整可运行示例

一个脚本同时完成识别和提交，这样 token 就不会闲在那里等下一步。最后这一点不是风格问题。一个 reCAPTCHA token 大约只有两分钟有效期，而一个先在某一步识别、再等待审批、然后在另一步提交的流程，是丢掉它最可靠的办法。更多内容见指南 reCAPTCHA token 过期.

# py312
# pip install capskip requests - both resolved from these imports.
import os
import requests
import wmill
from capskip import CapSkip
from capskip.exceptions import TimeoutException, NetworkException

SITE = "https://example.com/page-with-recaptcha"

def main(sitekey: str, username: str) -> dict:
    # CAPSKIP_HOST is set on the worker and allowed through by
    # WHITELIST_ENVS. It falls back to the loopback address so the
    # same script still runs on a worker that sits next to CapSkip.
    solver = CapSkip(
        host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
        port=8080,
        apiKey=wmill.get_variable("u/admin/capskip_key"),
    )

    try:
        result = solver.recaptcha(sitekey=sitekey, url=SITE)
    except TimeoutException:
        # Let the flow's retry policy decide what happens next.
        raise
    except NetworkException:
        raise RuntimeError("CapSkip is unreachable from this worker")

    # Submit immediately. The token is short lived, and the field
    # name below is the one the page's own form posts.
    posted = requests.post(
        SITE,
        data={
            "username": username,
            "g-recaptcha-response": result["code"],
        },
        timeout=30,
    )

    return {"status": posted.status_code, "captcha_id": result["captchaId"]}

上面最后那个脚本返回的是验证码任务 id 而不是 token，这是有意为之。以后你翻 Windmill 的运行历史时，id 是有用的，token 则没用：那时它早就过期了，而且把它放进存下来的任务结果里，等于把它写进了你的日志。

同时识别多个

一个 Windmill worker 一次只跑一个脚本，独占它所在的整台机器。所以这里的并发取决于你跑多少个 worker，而不取决于你的脚本怎么写。有两种办法，而且可以叠加使用。

• 多跑几个 worker。worker 组可以独立扩缩容，任务会被任何一个空闲的 worker 领走。
• 在一个脚本里批量识别。Python SDK 提供的是真正的异步客户端，所以一个任务里可以同时有好几个识别在跑。当一次运行需要十个 token 而不是一个时，这么做很值得，具体模式写在 用 Python 并行识别验证码的指南.

如果脆弱的一方是目标站点，就给脚本设一个并发上限。CapSkip 本身不计次，所以多跑几次并不会更贵，但你正在自动化的那个站点很可能会察觉。

常见错误及其含义

常见问题

简短版结论

在 Windmill 脚本顶部 import SDK，让依赖任务替你固定版本。把识别工具的密钥放进 Windmill 变量，把它的地址放进由 WHITELIST_ENVS 放行的 worker 环境变量。判断连接模式时要问 worker 跑在哪里，而不是你人坐在哪里：跑在识别工具那台 Windows 机器上的 worker 可以继续用 Local mode，其他情况都需要 Server mode。把脚本超时设到 300 秒以上，在流程步骤上加指数退避，并且在产生 token 的同一个任务里就把它提交掉。

• Python SDK 本身的说明见 Python 验证码识别页面.
• 勾选框挑战的说明见 reCAPTCHA v2 识别页面.
• Node.js、PHP 和 C# 里的等价调用列在 验证码识别 SDK 页面.

在你把它设成每五分钟跑一次之前，有一点值得掂量：CapSkip 是一款 本地验证码识别工具 ，运行在你已经拥有的硬件上，所以不停触发的流程和偶尔触发的流程，花费完全相同。