Как решать капчу в APScheduler без пропущенных запусков

Задание APScheduler для решения капчи представляет собой обычную функцию, которая вызывает CapSkip и запускается по расписанию с интервальным триггером. Само решение занимает одну строку. Проблема в настройках по умолчанию вокруг него. APScheduler одновременно выполняет только один экземпляр задания. Синхронный клиент Python SDK CapSkip не задаёт таймаут для своих HTTP-вызовов. Вместе это означает, что одно решение, которое так и не вернулось, блокирует все последующие запуски, и единственным признаком становится предупреждение каждые десять минут. Запуск, который начинается с опозданием больше чем на одну секунду, отбрасывается. А в Windows с Python 3.13 или более ранней версией Ctrl+C ничего не делает до следующего запланированного запуска. В этом руководстве все три проблемы исправлены в задании, которое каждые десять минут входит на сайт, защищённый reCAPTCHA.

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

• CapSkip, запущенный на машине с Windows, с включённым решателем reCAPTCHA.
• Python 3.10 или новее. Код ниже запускался на Python 3.13.7 и 3.14.0.
• APScheduler 3.11, текущая стабильная ветка, а также пакеты capskip и requests. Зафиксируйте версию APScheduler ниже 4: у предварительных выпусков 4.0 другой API, и импорты ниже на них не работают.
• Адрес решателя. В режиме Local CapSkip отвечает на 127.0.0.1 и только для этого устройства. В режиме Server он слушает ваш сетевой адрес или публичный IP, чтобы планировщик на другой машине мог обращаться к нему через API. Оба режима задаются в разделе Настройки подключения.

Шаг 1: задание, которое решает капчу

Заданием APScheduler может быть любая функция. Эта каждые десять минут запрашивает у CapSkip токен reCAPTCHA. Пишите 127.0.0.1, а не localhost: в Windows localhost сначала пробует адрес IPv6, и в наших запусках это удваивало время до ошибки о незапущенном решателе, с 2 до 4 секунд.

# pip install "apscheduler<4" capskip
from apscheduler.schedulers.blocking import BlockingScheduler
from capskip import CapSkip

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


def solve_login_captcha():
    result = solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")
    print("token:", result["code"][:20], "...")


scheduler = BlockingScheduler()
scheduler.add_job(solve_login_captcha, "interval", minutes=10)
scheduler.start()

BlockingScheduler работает в главном потоке и отправляет каждый запуск в пул из 10 рабочих потоков. Первый запуск происходит через один интервал после старта, а не в момент старта. Всё это работает нормально, пока решение не начнёт вести себя неправильно, и именно этому посвящена остальная часть руководства. Короткие примеры для краткости используют BlockingScheduler. В Windows с Python 3.13 или более ранней версией используйте вместо него цикл из полного скрипта ниже.

Почему медленное решение пропускает следующий запуск

У каждого задания max_instances по умолчанию равен 1. Когда подходит время запуска, а предыдущий запуск ещё идёт, APScheduler не запускает второй. Вместо этого он пишет в лог предупреждение уровня WARNING через логгер apscheduler.scheduler:

Execution of job "solve_login_captcha (trigger: interval[0:10:00], next run at: 2026-10-07 13:10:00 +08)" skipped: maximum number of running instances reached (1)

Для задания с капчей это значение по умолчанию правильное. Два одновременных решения для одного и того же входа вам не нужны никогда. Решение, которое длится чуть дольше интервала, стоит одного пропущенного запуска, и задание продолжается в следующее запланированное время. В тесте с интервалом 5 секунд и решением длиной 12.8 секунды на каждое решение пришлось по два пропущенных запуска, а следующий запуск начался в следующее запланированное время, а не сразу после решения.

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

Шаг 2: задаём HTTP-таймаут для SDK

Python SDK CapSkip вызывает requests.post и requests.get без таймаута. Если CapSkip принимает соединение и так и не отвечает, например потому что приложение зависло, вызов ждёт вечно. recaptchaTimeout из SDK (по умолчанию 300 секунд) не помогает, потому что он проверяется только между опросами, а зависший опрос никогда не заканчивается. В нашем тесте задание, застрявшее таким образом, так и осталось зависшим после того, как решатель пришёл в себя, и писало в лог только предупреждения о пропусках.

Два очевидных способа это исправить не работают. socket.setdefaulttimeout ни на что не влияет, потому что requests всегда передаёт вниз собственное значение таймаута, даже если это значение равно None. А объект сессии, который можно было бы настроить, SDK не принимает. Работает другое: подсунуть модулю SDK небольшую замену пакета requests, чьи вызовы post и get всегда передают таймаут.

# pip install capskip requests
import functools
import types

import requests
import capskip.api
from capskip import CapSkip


# The SDK calls requests.post/get with no timeout. Hand its module a
# stand-in whose calls always pass one: 5 s to connect, 30 s to answer.
capskip.api.requests = types.SimpleNamespace(
    RequestException=requests.RequestException,
    post=functools.partial(requests.post, timeout=(5, 30)),
    get=functools.partial(requests.get, timeout=(5, 30)),
)  # do this before the first solve
solver = CapSkip(host="127.0.0.1", port=8080)

Этот код заменяет имя requests внутри capskip.api, модуля, который в capskip 1.3.0 отправляет запросы SDK в CapSkip. После обновления SDK проверьте это снова. Выполните код один раз, до первого решения, и каждый вызов получит 5 секунд на подключение и 30 секунд на ответ.

После этого два вида зависания заканчиваются по-разному. Если CapSkip перестаёт отвечать во время отправки задачи, задание падает через 30 секунд с NetworkException с сообщением read timed out. Если он перестаёт отвечать, пока SDK опрашивает результат, SDK принимает таймаут за признак того, что ответ ещё не готов, и продолжает опрос, поэтому задание заканчивается по recaptchaTimeout с TimeoutException. Со значениями по умолчанию в нашем тесте это заняло 333 секунды. Значит, худший случай длится около пяти с половиной минут, что укладывается в десятиминутный интервал, и следующий запуск решает капчу как обычно.

Шаг 3: решаем, что происходит с запоздавшими запусками

Второе значение по умолчанию относится к misfire_grace_time и равно 1 секунде. Запуск, который начинается больше чем через секунду после запланированного времени, отбрасывается, и APScheduler пишет в лог предупреждение о том, что время запуска пропущено, с указанием задержки. Запуски опаздывают чаще, чем можно подумать. Достаточно пула, все потоки которого заняты решением, или ноутбука, который проспал запуск.

# pip install "apscheduler<4" capskip
from datetime import datetime

from apscheduler.schedulers.blocking import BlockingScheduler
from capskip import CapSkip

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


def solve_login_captcha():
    result = solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")
    print("token:", result["code"][:20])


scheduler = BlockingScheduler()
scheduler.add_job(
    solve_login_captcha, "interval", minutes=10,
    next_run_time=datetime.now(),  # first run now, not ten minutes from now
    jitter=30,                     # add 0 to 30 s to each run
    max_instances=1,               # the default: a slow solve skips the next run
    coalesce=True,                 # the default: after a stall, run once
    misfire_grace_time=None,       # default is 1 s; None runs late instead of dropping
)
scheduler.start()

Задайте misfire_grace_time значение None, и запоздавший запуск всё равно состоится, пусть и с опозданием. Оставьте у опции coalesce значение по умолчанию True, чтобы после долгого простоя несколько пропущенных запусков схлопывались в один, а не выполнялись один за другим. Мы приостановили планировщик на 14 секунд при интервале 4 секунды, чтобы сымитировать компьютер в спящем режиме. Со значениями по умолчанию догоняющий запуск был отброшен. С misfire_grace_time, равным None, он выполнился один раз. А когда вдобавок опция coalesce была равна False, он выполнился три раза подряд: три решения капчи для одного входа.

Передайте next_run_time, чтобы задание запустилось сразу при старте. Опция jitter добавляет к каждому запуску случайную задержку до указанного числа секунд, и это распределяет запросы во времени. Каждая задержка прибавляется к времени предыдущего запуска, которое уже включает задержку, поэтому средний период становится равен интервалу плюс половина jitter.

Шаг 4: видим сбои сразу, как они происходят

Исключение, выброшенное в задании, не останавливает планировщик. APScheduler пишет его в лог с уровнем ERROR, с сообщением о том, что задание выбросило исключение, и полным traceback, а задание сохраняет своё расписание. Слушатель получает те же события в виде объектов, так что их можно считать, отправлять по ним оповещения или записывать туда, куда вам нужно.

# pip install "apscheduler<4" capskip
import logging

from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MAX_INSTANCES, EVENT_JOB_MISSED
from apscheduler.schedulers.blocking import BlockingScheduler
from capskip import CapSkip

logging.basicConfig(level=logging.INFO)
solver = CapSkip(host="127.0.0.1", port=8080)


def solve_login_captcha():
    solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")


def on_event(event):
    if event.code == EVENT_JOB_ERROR:
        logging.error("job %s failed: %r", event.job_id, event.exception)
    elif event.code == EVENT_JOB_MAX_INSTANCES:
        logging.warning("job %s still solving, this run skipped", event.job_id)
    else:
        logging.warning("job %s started too late, this run dropped", event.job_id)


scheduler = BlockingScheduler()
scheduler.add_listener(on_event, EVENT_JOB_ERROR | EVENT_JOB_MAX_INSTANCES | EVENT_JOB_MISSED)
scheduler.add_job(solve_login_captcha, "interval", minutes=10, id="login")
scheduler.start()

Для задания с капчей хватает трёх событий. EVENT_JOB_ERROR несёт само исключение, например ApiException с сообщением ERROR_CAPTCHA_UNSOLVABLE. EVENT_JOB_MAX_INSTANCES срабатывает один раз на каждый пропущенный запуск, и зависшее решение становится видно по растущему счётчику. EVENT_JOB_MISSED срабатывает на каждый отброшенный запоздавший запуск.

Полное задание APScheduler для решения капчи

Этот скрипт загружает страницу входа через сессию requests, читает sitekey и поле CSRF, решает reCAPTCHA с помощью CapSkip и отправляет форму с токеном. Ошибки CapSkip он перехватывает внутри задания, поэтому неудачное решение оборачивается одним предупреждением, а следующий запуск пробует снова.

# pip install "apscheduler<4" capskip requests
import functools
import logging
import re
import time
import types
from datetime import datetime

import requests
import capskip.api
from apscheduler.schedulers.background import BackgroundScheduler
from capskip import CapSkip, CapSkipError

LOGIN_URL = "https://example.com/login"
USERNAME, PASSWORD = "YOUR_USERNAME", "YOUR_PASSWORD"

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
log = logging.getLogger("login-job")


# The SDK's HTTP calls have no timeout; give them one so a stuck CapSkip
# cannot hang the job.
capskip.api.requests = types.SimpleNamespace(
    RequestException=requests.RequestException,
    post=functools.partial(requests.post, timeout=(5, 30)),
    get=functools.partial(requests.get, timeout=(5, 30)),
)
solver = CapSkip(host="127.0.0.1", port=8080)


def field(html, pattern):
    match = re.search(pattern, html)
    if not match:
        raise ValueError(f"not found on the login page: {pattern}")
    return match.group(1)


def login():
    with requests.Session() as session:
        page = session.get(LOGIN_URL, timeout=30)
        sitekey = field(page.text, r'data-sitekey="([^"]+)"')
        # The form's hidden CSRF field (Django names it csrfmiddlewaretoken).
        csrf = field(page.text, r'name="csrfmiddlewaretoken" value="([^"]+)"')
        try:
            token = solver.recaptcha(sitekey=sitekey, url=LOGIN_URL)["code"]
        except CapSkipError as exc:
            log.warning("CapSkip could not solve: %s: %s", type(exc).__name__, exc)
            return
        reply = session.post(LOGIN_URL, timeout=30, headers={"Referer": LOGIN_URL}, data={
            "csrfmiddlewaretoken": csrf,
            "username": USERNAME,
            "password": PASSWORD,
            "g-recaptcha-response": token,
        })
        # Many sites answer a failed login with 200 and the form again;
        # check for something only a logged-in page shows.
        if reply.ok:
            log.info("logged in (HTTP %s)", reply.status_code)
        else:
            log.warning("login refused (HTTP %s): %s", reply.status_code, reply.text[:200])


scheduler = BackgroundScheduler()
scheduler.add_job(
    login, "interval", minutes=10,
    next_run_time=datetime.now(),  # first run now, not ten minutes from now
    max_instances=1,               # never two solves for the same account at once
    coalesce=True,                 # after a stall, run once, not once per missed slot
    misfire_grace_time=None,       # a late run still runs instead of being dropped
)
scheduler.start()
try:
    while True:
        time.sleep(1)  # Ctrl+C lands here at once, on any Python version
except KeyboardInterrupt:
    log.info("stopping; a solve in flight finishes first")
    scheduler.shutdown()

Он использует BackgroundScheduler и цикл со sleep вместо BlockingScheduler, и причина проявляется только в Windows. С BlockingScheduler на Python 3.13 или более ранней версии главный поток ждёт на блокировке, которую Ctrl+C не может прервать, поэтому KeyboardInterrupt приходит только при следующем запланированном пробуждении. При десятиминутном интервале это может занять десять минут. В Python 3.14 это исправили для Windows, но цикл работает на любой версии: Ctrl+C сразу прерывает time.sleep.

Остановка всё равно ждёт решения, которое уже идёт. Прежде чем процесс завершится, Python ждёт, пока закончат работу рабочие потоки пула, поэтому идущее решение сначала завершается. Так происходит независимо от того, вызываете ли вы scheduler.shutdown(), и с таймаутом из шага 2 ожидание длится не больше чем примерно пять с половиной минут. Второе нажатие Ctrl+C его не сокращает.

Сессия отправляет форму с заголовком Referer, потому что requests сам его не отправляет, а фреймворки вроде Django отклоняют отправку формы по HTTPS без него. Кроме того, сессия хранит cookie, который соответствует полю CSRF. Решайте капчу прямо перед отправкой: токен reCAPTCHA истекает через две минуты после выдачи. Невидимые виджеты, ключи Enterprise и другие варианты разобраны в руководстве по решению reCAPTCHA v2 на Python.

Много аккаунтов по одному расписанию

Хорошо работает схема с одним заданием на аккаунт и одним общим клиентом CapSkip, который безопасно использовать из нескольких потоков. Сделайте пул потоков таким же большим, как число аккаунтов.

# pip install "apscheduler<4" capskip
import logging
from datetime import datetime

from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.schedulers.blocking import BlockingScheduler
from capskip import CapSkip

logging.basicConfig(level=logging.INFO)
solver = CapSkip(host="127.0.0.1", port=8080)  # one client is safe to share across threads
ACCOUNTS = [f"user{i:02d}" for i in range(20)]


def login(account):
    token = solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")["code"]
    logging.info("%s got a token %s", account, token[:12])


scheduler = BlockingScheduler(executors={"default": ThreadPoolExecutor(max_workers=20)})
for account in ACCOUNTS:
    scheduler.add_job(login, "interval", minutes=10, args=[account], id=account,
                      next_run_time=datetime.now(), misfire_grace_time=None)
scheduler.start()

В пуле по умолчанию 10 потоков. Когда одновременно подошло время 20 заданий, 10 запусков ждали свободного потока, стартовали с опозданием 7.8 секунды и были отброшены из-за допуска по умолчанию в 1 секунду. Половина аккаунтов так и не вошла в систему, и сообщали об этом только предупреждения о пропущенных запусках. Это исправляет пул из 20 потоков или misfire_grace_time, равный None.

У CapSkip есть собственный лимит. Его настройки reCAPTCHA по умолчанию разрешают 10 одновременных решений (Max. Threads), а следующие задачи ждут свободного потока. В нашем тесте с 20 потоками 10 решений заняли 7.8 секунды, а остальные 10 заняли 12.8, потому что 5 секунд ждали CapSkip. Поднимите Max. Threads в CapSkip, если ресурсов вашей машины хватает, а если нет, распределите аккаунты во времени с помощью jitter.

С asyncio вместо потоков

Если остальной ваш код асинхронный, используйте AsyncIOScheduler с AsyncCapSkip, нативным асинхронным клиентом SDK. Задания-корутины выполняются в цикле событий, а max_instances работает точно так же, как раньше.

# pip install "apscheduler<4" capskip
import asyncio

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from capskip import AsyncCapSkip

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


async def solve_login_captcha():
    result = await solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")
    print("token:", result["code"][:20], "...")


async def main():
    scheduler = AsyncIOScheduler()
    scheduler.add_job(solve_login_captcha, "interval", minutes=10)
    scheduler.start()
    await asyncio.Event().wait()  # keep the event loop running


asyncio.run(main())

Отличий два. AsyncCapSkip построен на httpx, у которого таймаут по умолчанию 5 секунд, поэтому зависший CapSkip заканчивается исключением и без патча из шага 2: через 5 секунд, если зависла отправка задачи, и по recaptchaTimeout, если завис опрос. NetworkException от него может прийти с пустым сообщением, поэтому записывайте в лог и тип исключения. А остановка не ждёт идущее решение, а отменяет его: тогда задание пишет в лог CancelledError как своё исключение. В traceback над ним также виден NetworkException, но это всего лишь сигнал SDK о том, что ответ ещё не готов, а не сетевой сбой. Ctrl+C завершает программу за доли секунды.

Запуск CapSkip на другой машине

Планировщику и решателю не обязательно работать на одном ПК. Переключите CapSkip в режим Server в настройках подключения, и он будет слушать ваш сетевой адрес или публичный IP. Затем направьте клиент на этот адрес.

# Replace the solver line with this when CapSkip runs in Server mode
# on another Windows machine on your network.
solver = CapSkip(apiKey="YOUR_API_KEY", host="192.168.1.50", port=8080)

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

Частые ошибки при решении капчи в APScheduler

FAQ

Коротко

Заданию APScheduler для решения капчи нужно больше, чем само решение. Задайте таймаут HTTP-вызовам SDK, потому что при max_instances, равном 1, из-за одного зависшего решения пропускаются все последующие запуски. Задайте misfire_grace_time значение None и оставьте coalesce включённым, чтобы запоздавшие запуски выполнялись один раз, а не отбрасывались. Подгоните пул потоков под свои задания, перехватывайте CapSkipError внутри задания и используйте BackgroundScheduler с циклом sleep, если останавливаете планировщик через Ctrl+C в Windows с Python 3.13 или более ранней версией. Руководство пользователя APScheduler 3.x описывает каждый параметр, использованный здесь.

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

Задание, которое запускается каждые десять минут, делает больше четырёх тысяч решений в месяц. Когда распознавание капчи выполняется на вашей собственной машине с Windows, это число никогда не превращается в счёт.