如何在 Celery 任务中识别验证码（Python 队列）

Celery 里的验证码任务有一条规则决定了其余的一切：它必须在每一次尝试里都从头识别一遍。Celery 给你的是至少一次投递，所以一个任务可能跑两遍，而 CapSkip 的结果只能读一次。存下一个 captcha id、重试之后再接着用它，你什么也拿不到。识别、用掉 token、结束，全都在同一个任务体里完成。

你需要什么

• Celery 5 加一个 broker，Redis 或 RabbitMQ。broker 的选择会影响后面的一个设置。
• CapSkip 跑在一台 Windows 机器上，worker 镜像里装好 Python 客户端。
• 几乎所有真实部署里都是 Server mode（服务器模式）。worker 通常跑在 Linux 容器里，而识别工具不是。
• sitekey 和页面 URL，作为任务参数传进去，而不是写死在任务里。

# pip install capskip
pip install -U celery[redis] capskip

第 1 步：任务本身

整个识别就是一次调用。SDK 负责提交、轮询并返回 token，所以没有 id 需要在步骤之间传递，也没有任何东西需要持久化。

# pip install capskip
import os
from celery import Celery
from capskip import CapSkip, NetworkException

app = Celery("solves", broker="redis://redis:6379/0")

# CAPSKIP_HOST is the solver machine. Loopback only works if the
# worker runs on the same Windows box as CapSkip.
solver = CapSkip(
    host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.environ.get("CAPSKIP_PORT", "8080")),
    apiKey=os.environ.get("CAPSKIP_API_KEY", "capskip"),
)

@app.task(
    bind=True,
    autoretry_for=(NetworkException,),
    retry_backoff=True,
    max_retries=3,
    soft_time_limit=330,
    time_limit=360,
)
def solve_and_submit(self, sitekey, page_url):
    result = solver.recaptcha(sitekey=sitekey, url=page_url)
    return submit_form(page_url, result["code"])   # token, used here

注意里面没有什么。返回值里没有 captcha id，没有第二个任务去提交 token，也没有把结果存下来留着以后用。token 就在生成它的同一个任务里被用掉。理由是时序，而不是整洁，这在下面讲重试的部分里会说到。

那次调用是 reCAPTCHA v2 的。CapSkip 支持的其他类型形态完全一样：传 invisible 或 enterprise 设为 1，或者 version 设为 v3 并带一个 action，或者改成调用 turnstile 或 geetest。完整的接口见 Python 验证码识别页面.

第 2 步：两个时间限制，以及该把它们设在哪里

Celery 默认没有时间限制，两个设置都没有。对一个要等网络服务的任务来说，这个默认值是错的，因为一次卡住的识别会永远占着一个 worker 槽位。

两个都要设，而且都要设在 SDK 自己允许的时长之上。CapSkip 给 reCAPTCHA、Turnstile 或极验（GeeTest）识别三百秒，给图片验证码一百二十秒，两者都可以在客户端上配置。如果软限制先触发，Celery 会在你的任务里抛出 SoftTimeLimitExceeded，你就丢掉了 SDK 自己的 TimeoutException，而后者才是更有用的信号，因为它告诉你识别工具是连上了、只是没做完。

如果你在同一组 worker 上同时跑图片验证码和 reCAPTCHA，就给它们分别建任务、分别设限制，而不是用一个任务套上那对更大的数值。给一个通常一秒就完成的作业设三百秒的上限，会把真正的故障藏上五分钟。

第 3 步：识别场景特有的那条重试规则

Celery 的自动重试，对一个正在重启的识别工具来说恰好合适，对一个你已经拿在手里的 token 来说恰好相反。这个区别值得说清楚。

把 retry_backoff 设为 True，第一次重试等一秒，然后两秒、四秒、八秒，而且抖动默认开启，所以真实延迟是一个不超过该上限的随机值。封顶值是 retry_backoff_max，默认六百秒。拿它和 reCAPTCHA token 比一比：后者的有效期大约两分钟。

所以，一个已经识别成功、存下 token、在提交那一步失败、然后重试的任务，可能要在比 token 寿命长好几倍的延迟之后才被恢复执行。它会栽在一个生成时完全有效的 token 上，而日志会去怪目标站点。这种失败模式值得读一读指南 reCAPTCHA token 过期.

解决办法就是上面那种任务形态：在重试里面识别，而不是在重试之前。

把 autoretry_for 限制在那些描述传输问题的异常上。NetworkException 意味着 CapSkip 没连上，值得重试。ApiException 和 ValidationException 意味着请求本身有问题，再来一次还是有问题。TimeoutException 要自己拿捏，通常值得重试一次，而不是三次。

第 4 步：acks_late，以及一个结果为什么只能读一次

Celery 默认在运行一条消息之前就确认它，所以一个在任务中途挂掉的 worker 会把这个作业丢掉。打开 task_acks_late 会把确认挪到任务结束之后，于是崩溃的 worker 手上的作业会被重新投递并再跑一次。对那些在别处要花钱的工作来说，这通常正是你想要的，也正是它让只能读一次这条规则变得重要。

CapSkip 的结果只能读一次。如果第一次尝试提交了挑战、读到了 token，然后在确认之前崩溃了，重新投递的那次尝试是没法再读那个 id 的。它必须重新提交。上面那个任务做的正是这件事，因为它在两次尝试之间不保存任何状态，而重新识别的代价只是你自己硬件上的几秒钟，而不是再被收一次费。

# Redelivery is safe here because the task resolves rather
# than resuming. Pair it with reject_on_worker_lost so a
# killed worker requeues instead of dropping the job.
app.conf.task_acks_late = True
app.conf.task_reject_on_worker_lost = True

# Long tasks and a prefetch of 4 means idle workers sit on
# queued jobs. Drop it to 1 for solve queues.
app.conf.worker_prefetch_multiplier = 1

最后那一项是大多数识别队列里那个悄无声息的性能问题。预取倍数默认是四，所以每个 worker 进程会预先占住四条消息。对那些几毫秒就跑完的任务来说，这是好事。对那些要等一次识别的任务来说，四条里有三条被压在一个除了轮询什么都不干的作业后面，与此同时另一个 worker 却闲着没活干。

第 5 步：会让识别重复执行的那个 broker 设置

如果你的 broker 是 Redis，那还有一个数字要管。Redis 没有原生确认机制，所以 Celery 用一个 visibility timeout 来模拟：它等待一个 worker 确认任务的秒数，超过就把消息重新投递给另一个 worker。它默认是一小时，而且它住在 broker_transport_options 里面，并不是一个独立的设置项。

一小时远远高于三百秒的识别，所以默认值是安全的。麻烦从有人为了让失败作业更快恢复而把它调小开始，因为 Celery 自己的文档就警告过：执行时间超过 visibility timeout 的任务会被再次执行，然后一遍又一遍地循环下去。于是每一次慢识别都至少跑两遍。

# Keep this above your hard time limit, not near it.
# 3600 is the default and it is fine. If you must lower it,
# stay well clear of the 360 second time_limit above.
app.conf.broker_transport_options = {"visibility_timeout": 3600}

RabbitMQ 是原生确认的，没有对应的设置，这也是慢任务多的队列更该选它的一个理由。

第 6 步：把识别工具跑在 worker 够得着的地方

Celery worker 通常跑在集群里的 Linux 容器上。CapSkip 跑在 Windows 上。所以实际上 worker 和识别工具在两台不同的机器上，回环地址不是答案。

CapSkip 为此提供两种连接模式。Local mode（本地模式）绑定 127.0.0.1，只服务本机。Server mode 绑定你的内网地址或公网 IP，于是另一台机器、一个容器宿主机或者一个托管平台，都能通过 API 访问同一台 Windows 机器。两种模式都说明在 连接设置，而 Server 模式只改变识别程序监听在哪个地址上。硬件还是你自己的，识别也依然不计量。

正是最后这一点，才让一个不停触发的队列成为一件跑得起的事。

host 和密钥要从环境变量里读，而不是写在代码里。Python 客户端不会自己读取 CAPSKIP_HOST、CAPSKIP_PORT 或 CAPSKIP_API_KEY，所以第 1 步里的任务会读取它们并传进去。这样一来，worker 容器只需要这几个变量，别的什么都不需要。

同时识别很多个

有两种做法，适合不同形态的工作。一个验证码一个任务，靠 worker 并发度来实现并行，这是通常的答案，也正是上面那条预取提示所针对的情况。而对于成批一起到达的任务，Python 客户端有一个货真价实的异步实现，所以一个任务可以在它内部就把一整批收拢起来一起处理。

import asyncio
import os
from capskip import AsyncCapSkip

# AsyncCapSkip in Python is a real async client, not an alias.
async def solve_batch(pairs):
    solver = AsyncCapSkip(
        host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"), port=8080)
    return await asyncio.gather(*[
        solver.recaptcha(sitekey=k, url=u) for k, u in pairs
    ])

@app.task(soft_time_limit=330, time_limit=360)
def solve_many(pairs):
    return [r["code"] for r in asyncio.run(solve_batch(pairs))]

在你把这套东西照搬到另一门语言之前值得知道：Node 和 .NET 客户端里也有一个叫 AsyncCapSkip 的类，但在那里它只是个别名，而不是第二套实现。只有在 Python 里它才真有意义。完整的对比见指南 在 Python 中并行识别验证码.

常见错误及其含义

密钥相关的错误值得单独读一读，因为同一个响应既覆盖密钥缺失，也覆盖密钥单纯写错了的情况： 如何修复 ERROR_WRONG_USER_KEY.

常见问题

简短版结论

把识别和用到 token 的那部分放进同一个任务。把软限制设成 330、硬限制设成 360，好让客户端自己的超时先报出来。只在 NetworkException 上重试，并且让重试重新识别而不是接着上次继续，因为一次退避可能远远长过一个 token 的寿命，而且一个结果只能读一次。打开延迟确认，把预取倍数降到 1，并让 Redis 的 visibility timeout 远高于硬限制。只要 worker 不在识别工具本机上，就让 CapSkip 以 Server mode 运行。

• reCAPTCHA v2 复选框本身的说明见 reCAPTCHA v2 识别页面.
• 客户端背后的原始端点，文档见 CapSkip API 文档.

在给队列定规模之前，最后还有一点值得掂量：CapSkip 是一款 验证码识别工具 ，跑在你已经拥有的硬件上，所以一百个 worker 猛砸它，和一个 worker 慢慢磨完同样这些活，花的钱一样都是零。