Как решать капчи в заданиях RQ, не упираясь в лимит 180 секунд

Чтобы решать капчу в RQ, очереди заданий для Python на основе Redis, поместите всю последовательность в одно задание: загрузить страницу, решить капчу с помощью CapSkip, отправить форму. После этого в задании RQ с капчей нужно поменять всего одну настройку, его таймаут. RQ даёт каждому заданию 180 секунд, если не указать иное. SDK CapSkip ждёт токен reCAPTCHA до 300 секунд, поэтому решение, которое стоит в очереди за занятыми потоками, убивается посреди опроса. RQ записывает JobTimeoutException, а CapSkip заканчивает решение, которое уже никому не нужно. В этом руководстве разобраны задание, таймаут, воркеры, повторы и то, где может работать воркер, включая Windows.

Что понадобится

• CapSkip, запущенный на машине с Windows. Примеры решают reCAPTCHA v2, и та же схема задания работает для любого типа, который поддерживает CapSkip.
• Python 3.10 или новее с RQ 2.x, пакетом capskip и requests. Примеры запускались на RQ 2.12.
• Сервер Redis версии 5 или новее, до которого могут достучаться и код, ставящий задания в очередь, и воркеры. RQ также работает с Valkey 7.2 или новее.
• Машина для воркеров. Стандартный воркер RQ опирается на системный вызов Unix, которого нет в Windows, поэтому отдельный раздел ниже разбирает варианты для Windows, для WSL и для другой машины.
• Адрес решателя. В режиме Local CapSkip отвечает на 127.0.0.1 и только для этого устройства. В режиме Server он слушает ваш сетевой адрес или публичный IP, чтобы воркер на другой машине мог обращаться к нему через API. Оба режима задаются в разделе Настройки подключения.

pip install rq capskip requests

Шаг 1: выносим задание в отдельный модуль

RQ хранит задание как путь импорта функции плюс её аргументы, а воркер импортирует эту функцию, когда берёт задание. Функция, определённая в скрипте, который ставит её в очередь, живёт в __main__, а его воркер импортировать не может, поэтому 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, который ставит страница, возвращается вместе с формой. Парсер сохраняет скрытые поля той формы, в которой находится виджет, включая CSRF-токен, привязанный к этому cookie, и игнорирует форму поиска или подписки на рассылку в другом месте страницы. Пример отправляет форму обратно на URL страницы; если action формы указывает в другое место, отправляйте туда. Заголовок Referer нужен потому, что requests не отправляет ни Referer, ни Origin, а некоторые фреймворки, в том числе Django, отклоняют отправку формы по HTTPS, в которой нет ни того, ни другого.

raise_for_status превращает ошибку HTTP, например 403 после неудачной проверки CSRF, в упавшее задание, а не в тихий успех, и это важно, как только включены повторы. Сайт, который отклоняет токен обычной страницей с кодом 200, всё равно засчитывается как успех, поэтому дополнительно проверяйте в ответе то, что ваш сайт показывает при успехе.

Клиент решателя создаётся при импорте модуля. Со стандартным воркером RQ это происходит в новом дочернем процессе (work horse) каждого задания, а с SimpleWorker один раз на воркер. Подходит любой вариант, потому что клиент не держит открытых соединений между вызовами. SDK сам не читает переменные окружения, поэтому код читает CAPSKIP_HOST, CAPSKIP_PORT и CAPSKIP_API_KEY и передаёт их в конструктор. Чтобы направить другой воркер на другую машину с 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 задача может ждать одного из 10 потоков 250 секунд, а затем у неё есть до 250 секунд на загрузку капчи и ещё 250 на решение. Задачу, которая исчерпала любой из этих лимитов, CapSkip завершает с ошибкой, и она приходит как ApiException. Задача, которая какое-то время ждала поток, а затем медленно решается, может раньше выйти за 300 секунд SDK, и тогда всё заканчивается TimeoutException. Однако SDK не ставит таймаут на собственные HTTP-запросы к CapSkip. Если CapSkip перестанет отвечать посреди запроса, стандартный воркер RQ всё равно завершит задание по job_timeout, а SimpleWorker, о котором речь пойдёт ниже, этого сделать не может.

А вот что происходит со значением по умолчанию. На 180-й секунде RQ выбрасывает JobTimeoutException внутри задания, где бы оно ни находилось, а для задания с капчей это почти всегда цикл опроса SDK. Задание попадает в реестр упавших заданий с ошибкой Task exceeded maximum timeout value (180 seconds). CapSkip об этом не узнаёт. Он продолжает работу, выдаёт токен, который никто не заберёт, и занятый им поток потрачен впустую. Мы воспроизвели ровно это на решении, которое заняло 200 секунд.

Не перехватывайте Exception вокруг решения

JobTimeoutException представляет собой обычный подкласс Exception, который выбрасывается прямо в ваш код. Широкий except Exception вокруг решения его перехватывает, задание завершается нормально, и RQ записывает его как выполненное, без каких-либо следов таймаута. В нашем тесте задание, написанное таким образом, исчерпало свой лимит посреди решения, а воркер записал в лог 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: запускаем столько воркеров, сколько потоков у CapSkip

Воркер RQ выполняет одно задание за раз, поэтому для заданий RQ с капчей число воркеров равно числу решений в работе. По умолчанию CapSkip решает 10 reCAPTCHA одновременно (это его настройка Max. Threads), а любую следующую задачу заставляет ждать свободного потока. Воркеры сверх этого числа не выполняют больше работы. Вместо этого их задачи встают в очередь внутри CapSkip, и ожидание съедает таймаут каждого задания. Правильный размер пула равен одному воркеру на поток:

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

worker-pool запускает воркеры и перезапускает те, что упали. У каждого запущенного им воркера включён планировщик, и один из них запускает его для очереди, а от этого зависят повторы из следующего шага. Если вы поднимаете Max. Threads в CapSkip, увеличьте пул соответственно и считайте каждый воркер на каждой машине, которая обращается к тому же CapSkip.

Шаг 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-pool всегда запускает такой воркер. Одиночный воркер, запущенный простой командой rq worker, его не запускает, и тогда каждое повторяемое задание навсегда остаётся в реестре запланированных заданий без всякой ошибки. Мы наблюдали ровно это, и то же задание выполнилось за считаные секунды, как только запустился воркер с планировщиком. Одиночные воркеры запускайте так:

rq worker captcha --with-scheduler

Повтор выполняет всё задание заново: новая страница, новое решение и новая отправка. Это именно то, что нужно, потому что токен reCAPTCHA принимается только один раз и продолжать нечего. По этой же причине задание должно заканчиваться отправкой. Всё, что может упасть после успешной отправки, заставит повтор отправить форму второй раз.

Держите решение и отправку в одном задании

Работу соблазнительно разделить. Одно задание решает капчу и возвращает токен, а второе задание или код, поставивший первое в очередь, позже его отправляет. Не делайте так. Токен reCAPTCHA истекает примерно через две минуты после выдачи, а RQ ничего не обещает о том, когда начнётся следующее задание. Под нагрузкой оно ждёт за всем остальным, что стоит в очереди. Кроме того, по умолчанию RQ хранит возвращаемое значение задания всего 500 секунд. Решение внутри задания, прямо перед отправкой, укладывает всю жизнь токена в один вызов функции. Статья о сроке жизни токена reCAPTCHA подробно описывает правила его истечения.

Где может работать воркер RQ: Windows, WSL или другая машина

Стандартный воркер RQ создаёт через fork новый процесс для каждого задания. В Windows нет os.fork, поэтому там воркер останавливается с AttributeError на первом же задании. Документация RQ говорит, что воркерам нужен fork, и отправляет пользователей Windows в WSL, хотя на её странице о классах воркеров для Windows предлагается SpawnWorker. На RQ 2.12 SpawnWorker тоже падает, потому что ждёт каждое задание через os.wait4, которого в Windows тоже нет. Остаются три пути.

На самом ПК с CapSkip используйте SimpleWorker. Он выполняет каждое задание внутри процесса воркера, а не в дочернем процессе. В Windows с RQ 2.12 он выполнял наши задания и планировщик, а worker-pool запускал несколько таких воркеров параллельно. Считайте это проверенным нами, а не официально поддерживаемым в самом RQ. Он соблюдает job_timeout с помощью таймера, но таймер срабатывает только между инструкциями Python. Когда мы заставили вызов к CapSkip зависнуть на 40 секунд при job_timeout в 8 секунд, задание упало на 40-й секунде, а не на 8-й. Вызов, который так и не вернётся, держит этот воркер, пока вы его не перезапустите. CapSkip остаётся в режиме Local, потому что воркер находится на той же машине. У Redis нет официальной сборки для Windows, но Redis, запущенный под WSL, отвечает программам Windows на 127.0.0.1, и именно так мы тестировали эту схему:

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

Под WSL 2 на том же ПК считайте воркер другой машиной. В сетевом режиме NAT, который WSL 2 использует по умолчанию, 127.0.0.1 внутри Linux указывает на виртуальную машину Linux, а не на Windows, поэтому воркер там не может достучаться до CapSkip в режиме Local. Мы проверили: подключение не проходит. Переключите CapSkip в режим Server и направьте CAPSKIP_HOST на адрес хоста Windows в том виде, в каком его видит Linux. Руководство Microsoft по сетям в WSL показывает, как найти этот адрес, а также описывает зеркальный сетевой режим (mirrored) в Windows 11 22H2 и новее, в котором Linux может обращаться к серверам Windows по 127.0.0.1.

На Linux-сервере, хосте контейнеров или платформе используйте режим Server. Тогда CapSkip слушает ваш сетевой адрес или публичный IP, и каждый воркер подключается через один и тот же API. Если маршрут идёт через интернет, используйте статический публичный IP и включите проверку API-ключа. При запуске режима Server CapSkip открывает свой порт для всех адресов, поэтому, чтобы доступ к нему был только у ваших воркеров, добавьте в Windows Firewall правило, блокирующее остальные адреса. Решателем по-прежнему остаётся ваша собственная машина с Windows, и решения по-прежнему не тарифицируются, сколько бы воркеров к ней ни обращалось.

Полный рабочий пример

Два файла и одна команда. В 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, потому что именно оттуда воркеры его импортируют. rq info показывает для каждой очереди число ожидающих, выполняемых, завершённых и упавших заданий, а также воркеры. Реестр запланированных заданий, где ждут повторы с интервалом, он не учитывает. По умолчанию упавшее задание хранит свой traceback в течение года: получите задание через Job.fetch и прочитайте его exc_info.

Частые ошибки и что они означают

FAQ

Коротко

Для задания RQ с капчей поместите загрузку страницы, решение и отправку в одну функцию задания в отдельном модуле. Ставьте его в очередь с job_timeout в 420 секунд или задайте очереди такое значение по умолчанию, потому что собственные 180 секунд RQ короче, чем может длиться решение reCAPTCHA. Перехватывайте вокруг решения CapSkipError и никогда Exception. Запускайте через worker-pool столько воркеров, сколько потоков у CapSkip: он же запускает планировщик, который нужен повторам с интервалом. В Windows используйте SimpleWorker; из WSL или с другой машины переключите CapSkip в режим Server.

• Все типы капчи, которые поддерживает пакет для Python, перечислены на странице сервиса распознавания капч для Python.
• Как устроено само решение reCAPTCHA v2, рассказано на странице сервиса распознавания reCAPTCHA v2.

Десять воркеров, круглосуточно решающих капчу, создают ровно ту нагрузку, за которую сервис с поштучной оплатой берёт деньги за каждую тысячу решений. На вашей собственной машине распознавание капчи справляется с ней без платы за каждое решение.