如何在 RQ 作业中识别验证码并越过 180 秒时限

要在 RQ（基于 Redis 的 Python 作业队列）中识别验证码，就把整个流程放进一个作业：获取页面，用 CapSkip 识别验证码，提交表单。这样一来，RQ 验证码作业只需要改一个设置：它的超时。除非你另行指定，RQ 给每个作业 180 秒。CapSkip SDK 等待 reCAPTCHA token 最多可达 300 秒，所以一次排在繁忙线程后面的识别，会在轮询中途被杀掉。RQ 记下一个 JobTimeoutException，而 CapSkip 完成的识别结果没有人来取。本指南讲这几件事：作业、超时、worker、重试，以及 worker 可以跑在哪里，包括 Windows。

你需要什么

• 在 Windows 机器上运行的 CapSkip。示例识别的是 reCAPTCHA v2，同样的作业结构适用于 CapSkip 支持的所有类型。
• Python 3.10 或更高版本，加上 RQ 2.x、capskip 包和 requests。示例是在 RQ 2.12 上运行的。
• 一台 Redis 服务器，版本 5 或更高，入队作业的代码和 worker 都要能访问到它。RQ 也支持 Valkey 7.2 或更高版本。
• 一台运行 worker 的机器。RQ 的默认 worker 依赖一个 Windows 上没有的 Unix 调用，所以下文有一节专门讲在 Windows 上、在 WSL 下以及在另一台机器上的几种方案。
• 识别工具的地址。在 Local 模式下，CapSkip 只在 127.0.0.1 上响应，仅供本机使用。在 Server 模式下，它监听你的网络地址或公网 IP，这样另一台机器上的 worker 就能通过 API 调用它。两种模式的设置位置都是 连接设置.

pip install rq capskip requests

第 1 步：把作业写在单独的模块里

RQ 存储一个作业时，存的是函数的导入路径加上它的参数，worker 取到作业时再导入这个函数。定义在入队脚本里的函数位于 __main__ 中，worker 无法导入它，所以 RQ 在入队时就会用 ValueError 拒绝它。把作业放进单独的模块：

# tasks.py: the worker imports jobs by module path, so keep them here
# pip install rq capskip requests
import os
from html.parser import HTMLParser

import requests
from capskip import CapSkip

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")),
)


class FormPage(HTMLParser):
    """Collects the reCAPTCHA sitekey and the hidden fields of its form."""

    def __init__(self):
        super().__init__()
        self.sitekey, self.hidden, self._form = None, {}, None

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "form":
            self._form = {}
        if a.get("data-sitekey"):
            self.sitekey = a["data-sitekey"]
            if self._form is not None:
                self.hidden = self._form
        hidden = (a.get("type") or "").lower() == "hidden"
        if tag == "input" and hidden and a.get("name") and self._form is not None:
            self._form[a["name"]] = a.get("value") or ""

    def handle_endtag(self, tag):
        if tag == "form":
            self._form = None


def sign_up(page_url, username):
    # One session per job, so the page's cookies go back with the form.
    session = requests.Session()
    page = FormPage()
    page.feed(session.get(page_url, timeout=30).text)

    # Solve inside the job, right before the post: tokens expire.
    token = solver.recaptcha(page.sitekey, page_url)["code"]

    resp = session.post(page_url, data={
        **page.hidden,
        "username": username,
        "g-recaptcha-response": token,
    }, headers={"Referer": page_url}, timeout=30)
    resp.raise_for_status()
    return resp.status_code

每个作业都打开自己的 requests.Session，这样页面设置的会话 cookie 就会随表单一起发回去。解析器会保留包含小组件的那个表单的隐藏字段，其中就有与该 cookie 对应的 CSRF token，并忽略页面上其他位置的搜索表单或订阅表单。示例会提交回页面 URL；如果表单的 action 指向别处，就提交到那里。加上 Referer 请求头，是因为 requests 既不发送 Referer 也不发送 Origin，而有些框架（Django 就是其中之一）会拒绝两者都没有的 HTTPS 表单提交。

raise_for_status 会把 HTTP 错误（比如 CSRF 校验失败返回的 403）变成一个失败的作业，而不是悄无声息的成功；一旦开启重试，这一点就很重要。如果站点用一个正常的 200 页面拒绝 token，这仍然算作成功，所以还要检查响应里有没有你的站点在成功时显示的内容。

识别客户端在模块被导入时创建。使用 RQ 的默认 worker 时，这发生在每个作业新建的 work horse 子进程里；使用 SimpleWorker 时，每个 worker 只创建一次。两种都没问题，因为客户端在两次调用之间不保持任何连接。SDK 不会自己读取环境变量，所以代码会读取 CAPSKIP_HOST、CAPSKIP_PORT 和 CAPSKIP_API_KEY，再把它们传给构造函数。要让另一个 worker 指向另一台 CapSkip 机器，改它的环境变量就行，不用改代码。

第 2 步：入队时设置一个容得下识别的超时

RQ 验证码作业能否扛过一次慢识别，就取决于这个设置。每次入队时都传入 job_timeout，或者给 Queue 对象设一个 default_timeout。这个默认值存在 Python 对象里，而不是 Redis 里，所以每个负责入队的脚本都必须用同样的方式构建它的 Queue：

# enqueue.py
from redis import Redis
from rq import Queue

from tasks import sign_up

# RQ's default is 180 seconds; a reCAPTCHA solve may take 300.
queue = Queue("captcha", connection=Redis(), default_timeout=420)

job = queue.enqueue(sign_up, "https://example.com/signup", "YOUR_USERNAME")
print(job.id)

这个数字来自把最坏情况加起来。SDK 对一次 reCAPTCHA 识别最多轮询 300 秒，也就是它的 recaptchaTimeout，之后抛出 TimeoutException。获取页面和提交表单各自给 requests 传入 timeout=30，它限制的是建立连接以及每一次等待数据的时间，而不是整个请求的时间。加起来大约是 360 秒，设成 420 就为慢页面留出了一分钟的余量。

CapSkip 自己也有一套计时。在它的 reCAPTCHA 设置中，一个任务可以等待 250 秒来获得 10 个线程中的一个，之后最多有 250 秒加载挑战，再有 250 秒完成识别。任务只要耗尽其中任何一项时限，就会被 CapSkip 判定失败，表现为 ApiException。先等了一阵线程、之后又识别得慢的任务，则可能先超过 SDK 的 300 秒，结果是 TimeoutException。不过，SDK 对自己发往 CapSkip 的 HTTP 调用不设任何超时。如果 CapSkip 在一个请求中途停止响应，RQ 的默认 worker 仍会在 job_timeout 到期时结束这个作业；下文介绍的 SimpleWorker 则做不到。

如果用默认值，会发生这样的事。到了 180 秒，RQ 会在作业内部抛出 JobTimeoutException，不管作业当时执行到哪里；对验证码作业来说，几乎总是在 SDK 的轮询循环里。作业会进入失败作业注册表，错误信息是 Task exceeded maximum timeout value (180 seconds)。没有人通知 CapSkip。它会继续做下去，生成一个没人来取的 token，它占用的那个线程就白白耗掉了。我们用一次耗时 200 秒的识别，原样复现了这个过程。

不要在识别外面捕获 Exception

JobTimeoutException 是一个普通的 Exception 子类，会被抛进你的代码里。包在识别外面的宽泛 except Exception 会把它捕获，作业正常返回，RQ 把它记为已完成，哪里都看不到超时的痕迹。在我们的测试中，这样写的作业在识别中途耗尽了时限，worker 记录的却是 Job OK。应该改为捕获 SDK 自己的基类：

import logging

from capskip import CapSkipError

log = logging.getLogger(__name__)

try:
    token = solver.recaptcha(page.sitekey, page_url)["code"]
except CapSkipError as exc:
    # The SDK's errors only. RQ's JobTimeoutException is not one of
    # them, so a job that ran out of time still fails as it should.
    log.warning("solve failed for %s: %s", page_url, exc)
    raise

第 3 步：worker 数量与 CapSkip 的线程数保持一致

一个 RQ worker 一次只运行一个作业，所以对 RQ 验证码作业来说，worker 的数量就是同时在途的识别数量。CapSkip 默认同时识别 10 个 reCAPTCHA，这就是它的 Max. Threads 设置，多出来的任务都要等待空闲线程。超出这个数量的 worker 并不会完成更多工作。它们的任务只会在 CapSkip 内部排队，而排队的时间会蚕食每个作业的超时。每个线程配一个 worker，规模正合适：

# Ten workers for CapSkip's ten reCAPTCHA threads
rq worker-pool captcha -n 10

worker-pool 负责启动 worker，并重启任何挂掉的 worker。它启动的每个 worker 都开启了调度器，其中一个会为该队列实际运行调度器，下一步的重试就依赖于它。如果你在 CapSkip 中调高 Max. Threads，就把 worker 池也相应调大，并且要把调用同一个 CapSkip 的每台机器上的每个 worker 都算进去。

第 4 步：重试，以及为什么重试需要调度器

RQ 默认什么都不重试。失败的作业会进入失败作业注册表，并一直待在那里。要让 RQ 再试一次，就在入队时传入一个 Retry：

from rq import Retry

job = queue.enqueue(
    sign_up, "https://example.com/signup", "YOUR_USERNAME",
    # Up to two more runs: 30 seconds, then 120 seconds after a failure.
    retry=Retry(max=2, interval=[30, 120]),
)

带间隔的重试不会被直接放回队列。RQ 会为它排期，而只有运行着调度器的 worker 才会把它移回队列。worker-pool 总会运行一个调度器。用普通的 rq worker 启动的单个 worker 则不会，这时每个被重试的作业都会在已调度注册表里永远等下去，而且不报任何错误。我们亲眼看到过这种情况，而一旦启动了一个带调度器的 worker，同一个作业几秒内就运行了。像这样启动单个 worker：

rq worker captcha --with-scheduler

一次重试会把整个作业重新运行一遍：重新获取页面、重新识别、重新提交。这正是你想要的，因为 reCAPTCHA token 只会被接受一次，所以没有什么可以接着继续的。这也是作业应该在提交处结束的原因。提交成功之后任何可能失败的步骤，都会让重试再提交一次。

把识别和提交放在同一个作业里

把工作拆开很诱人：一个作业负责识别并返回 token，再由第二个作业或入队第一个作业的代码稍后提交它。不要这样做。reCAPTCHA token 在签发后大约两分钟就会过期，而 RQ 不保证下一个作业何时开始。负载高时，它要排在队列里所有其他作业后面。RQ 默认也只把作业的返回值保留 500 秒。在作业内部、紧挨着提交之前完成识别，能让 token 的整个生命周期都处在一次函数调用之内。 reCAPTCHA token 能保持有效多久 一文详细讲解了过期规则。

RQ worker 可以跑在哪里：Windows、WSL 或另一台机器

RQ 的默认 worker 会为每个作业 fork 一个新进程。Windows 没有 os.fork，所以在 Windows 上，worker 处理第一个作业时就会因 AttributeError 而停止。RQ 的文档说 worker 需要 fork，并让 Windows 用户改用 WSL，尽管它介绍 worker 类的那一页又为 Windows 提供了 SpawnWorker。在 RQ 2.12 上，SpawnWorker 同样会失败，因为它用 os.wait4 等待每个作业，而 Windows 也没有这个调用。这样就剩下三条路。

在 CapSkip 所在的电脑上，使用 SimpleWorker。 它在 worker 进程内部运行每个作业，而不是在子进程里。在 Windows 上配合 RQ 2.12，它运行了我们的作业和调度器，worker-pool 也能并行启动多个 SimpleWorker。请把这看作我们在这里测试过，而不是上游正式支持。它用计时器来执行 job_timeout，但计时器只会在 Python 语句之间触发。我们让一次对 CapSkip 的调用在 8 秒的 job_timeout 下卡住 40 秒，结果作业在 40 秒时才失败，而不是 8 秒。一次永远不返回的调用，会一直占住那个 worker，直到你重启它。CapSkip 保持 Local 模式即可，因为 worker 在同一台机器上。Redis 没有官方的 Windows 版本，但在 WSL 下运行的 Redis 会在 127.0.0.1 上响应 Windows 程序，我们就是这样测试这套环境的：

# On Windows: no fork, so run jobs inside the worker process
rq worker-pool captcha -n 10 -w rq.worker.SimpleWorker

在同一台电脑的 WSL 2 下，把 worker 当作另一台机器。 在 WSL 2 默认的 NAT 网络模式下，Linux 里的 127.0.0.1 指的是 Linux 虚拟机，而不是 Windows，所以那里的 worker 无法访问处于 Local 模式的 CapSkip。我们检查过，连接确实失败。把 CapSkip 切换到 Server 模式，并把 CAPSKIP_HOST 指向 Linux 所看到的 Windows 主机地址。 Microsoft 的 WSL 网络指南 介绍了如何找到这个地址，还说明了 Windows 11 22H2 及更高版本上的镜像网络模式，在这种模式下，Linux 可以通过 127.0.0.1 访问 Windows 上的服务。

在 Linux 服务器、容器宿主机或托管平台上，使用 Server 模式。 这时 CapSkip 会监听你的网络地址或公网 IP，每个 worker 都通过同一套 API 连接。如果链路要经过公网，请使用静态公网 IP，并打开 API 密钥校验。Server 模式启动时，CapSkip 会向所有地址开放端口，所以要想只让你的 worker 访问，就为其他地址添加一条 Windows Firewall 阻止规则。识别工具仍然是你自己的 Windows 机器，不管有多少个 worker 调用它，识别也仍然不计量。

完整可运行示例

两个文件加一条命令。tasks.py 里是作业本身，带有第 2 步中那种只捕获特定异常的处理；enqueue.py 负责设置超时和重试。

# tasks.py
# pip install rq capskip requests
import logging
import os
from html.parser import HTMLParser

import requests
from capskip import CapSkip, CapSkipError

log = logging.getLogger(__name__)
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")),
)


class FormPage(HTMLParser):
    def __init__(self):
        super().__init__()
        self.sitekey, self.hidden, self._form = None, {}, None

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "form":
            self._form = {}
        if a.get("data-sitekey"):
            self.sitekey = a["data-sitekey"]
            if self._form is not None:
                self.hidden = self._form
        hidden = (a.get("type") or "").lower() == "hidden"
        if tag == "input" and hidden and a.get("name") and self._form is not None:
            self._form[a["name"]] = a.get("value") or ""

    def handle_endtag(self, tag):
        if tag == "form":
            self._form = None


def sign_up(page_url, username):
    session = requests.Session()
    page = FormPage()
    page.feed(session.get(page_url, timeout=30).text)
    if not page.sitekey:
        raise ValueError(f"no data-sitekey on {page_url}")

    try:
        token = solver.recaptcha(page.sitekey, page_url)["code"]
    except CapSkipError as exc:
        # RQ's JobTimeoutException is not a CapSkipError, so it passes.
        log.warning("solve failed for %s: %s", page_url, exc)
        raise

    resp = session.post(page_url, data={
        **page.hidden,
        "username": username,
        "g-recaptcha-response": token,
    }, headers={"Referer": page_url}, timeout=30)
    resp.raise_for_status()
    return resp.status_code

# enqueue.py
from redis import Redis
from rq import Queue, Retry

from tasks import sign_up

queue = Queue("captcha", connection=Redis(), default_timeout=420)

for user in ["YOUR_USERNAME_1", "YOUR_USERNAME_2"]:
    job = queue.enqueue(
        sign_up, "https://example.com/signup", user,
        retry=Retry(max=2, interval=[30, 120]),
    )
    print(job.id)

# Linux or WSL; on Windows add -w rq.worker.SimpleWorker
rq worker-pool captcha -n 10

在存放 tasks.py 的文件夹里运行 worker 池，因为 worker 要从那里导入它。rq info 会显示每个队列中等待、执行中、已完成和失败的作业数，以及各个 worker。它不统计已调度注册表，而带间隔的重试正是在那里等待。失败的作业默认会把 traceback 保留一年：用 Job.fetch 取出作业，读取它的 exc_info 即可。

常见错误及其含义

常见问题

简短版结论

对于 RQ 验证码作业，把获取页面、识别和提交放进单独模块里的一个作业函数。入队时把 job_timeout 设为 420 秒，或者给队列设这个默认值，因为 RQ 自己的 180 秒比一次 reCAPTCHA 识别可能花的时间要短。在识别外面捕获 CapSkipError，绝不要捕获 Exception。用 worker-pool 运行与 CapSkip 线程数相同数量的 worker，它还会运行带间隔的重试所需要的调度器。在 Windows 上使用 SimpleWorker；从 WSL 或另一台机器调用时，把 CapSkip 切换到 Server 模式。

• Python 包能处理的所有验证码类型： Python 验证码识别页面.
• reCAPTCHA v2 识别本身是如何运作的： reCAPTCHA v2 识别页面.

十个 worker 全天候不停地识别，正是按量计费的服务按千次收费的那种负载。而运行在你自己机器上的 验证码识别工具 承接这样的负载，不会产生按次计费的账单。