Как решать капчи параллельно с помощью Python asyncio

Если вам нужно решать капчи параллельно из Python, используйте AsyncCapSkip с помощью asyncio.gather. Python — единственный CapSkip SDK, где асинхронный клиент является настоящей асинхронной реализацией, а не псевдонимом, поэтому пакет из десяти reCAPTCHA завершается примерно за время самой медленной из них, а не за сумму всех десяти. В этом руководстве — рабочий код, способ ограничить параллелизм, чтобы не перегрузить сервис, и приём, который не даёт одной ошибке погубить весь пакет.

Чем интересен именно клиент для Python

Все четыре SDK экспортируют нечто под названием AsyncCapSkip. Но только в одном из них это отдельная реализация.

SDK	Что такое AsyncCapSkip —	Как выполняется параллельная работа
Python	Настоящий асинхронный клиент	await asyncio.gather(...)
Node.js	Псевдоним для CapSkip	await Promise.all([...])
.NET	Псевдоним для CapSkipClient	await Task.WhenAll(...)
PHP	Только псевдоним, ради единообразия исходников	Синхронно, без параллелизма

Node и .NET и так неблокирующие, поэтому псевдоним там ничего не стоит. PHP синхронный, и псевдоним не даёт вообще ничего. В Python же разница принципиальна: обычный CapSkip клиент блокирует цикл событий во время опроса, поэтому помещение его в корутину даёт параллелизм на бумаге и последовательное время на практике.

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

• Python 3.10 или новее
• pip install capskip
• Запущенный CapSkip с включённым API-сервером, слушающим на 127.0.0.1:8080
• Список sitekey и URL страниц для обработки

Ничего не покидает вашу машину, поэтому нет ни лимитов, которые надо согласовывать, ни счётчика оплаты за решения, пока вы экспериментируете. Полные сигнатуры методов для каждого типа есть на странице SDK для распознавания капчи .

Последовательная версия и её цена

Вот с чего обычно начинают. Код корректный — и медленный.

# pip install capskip
from capskip import CapSkip

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

targets = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

# Each call blocks until that one CAPTCHA comes back.
for sitekey, url in targets:
    result = solver.recaptcha(sitekey=sitekey, url=url)
    print(result["code"][:40])   # the token, truncated for the log

Решение reCAPTCHA — это в основном ожидание. Ваш процесс простаивает, пока работает сервис распознавания, затем переходит к следующей задаче и простаивает снова. Три решения занимают время трёх решений. Тридцать — тридцати.

Решаем несколько капч одновременно с asyncio.gather

Замените клиент, добавьте await к вызовам и передайте их все в gather. Типы можно смешивать: reCAPTCHA, Turnstile и GeeTest в одном пакете — это нормально.

# pip install capskip
import asyncio
from capskip import AsyncCapSkip

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)

    # gather starts all three now and waits for the slowest.
    results = await asyncio.gather(
        solver.recaptcha(sitekey="SITEKEY_A", url="https://example.com/page-a"),
        solver.turnstile(sitekey="SITEKEY_B", url="https://example.com/page-b"),
        solver.normal("captcha.png"),
    )

    for r in results:
        print(r["code"][:40])   # token for widgets, text for images

asyncio.run(main())

Каждый метод возвращает dict с одними и теми же основными полями: captchaId и code. Turnstile дополнительно возвращает userAgent, который нужно вернуть вместе с токеном при отправке решения страницы-проверки. GeeTest добавляет challenge, validate и seccode, а сырой JSON кладёт в code.

Ограничьте параллелизм семафором

Не стоит запускать двести решений на локальный демон и надеяться на лучшее. Распознавание — это работа процессора на вашей собственной машине, поэтому после определённой ширины вы просто становитесь в очередь сами к себе, и каждое отдельное решение замедляется. Семафор удерживает фиксированное число задач в работе.

import asyncio
from capskip import AsyncCapSkip

# Start at 4 or 5, then measure. More is not automatically faster.
sem = asyncio.Semaphore(5)

async def solve_one(solver, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def run(targets):
    solver = AsyncCapSkip()
    tasks = [solve_one(solver, k, u) for k, u in targets]
    return await asyncio.gather(*tasks)

Подбирайте число замерами, а не на глаз. Прогоните одни и те же 20 целей при 2, 5 и 10 и оставьте то значение, которое быстрее завершилось на вашем железе. Правильный ответ зависит от вашего процессора, а не от SDK.

Один клиент, а не по клиенту на задачу

Создайте один AsyncCapSkip и разделяйте его между корутинами, как показано выше. Создавать по клиенту на задачу расточительно и бессмысленно: клиент хранит конфигурацию, а не состояние отдельного решения.

Не дайте одной ошибке погубить весь пакет

По умолчанию gather пробрасывает первое же исключение, и вы теряете результаты всего остального, что было в работе. Передайте return_exceptions=True , и исключения придут обычными элементами списка результатов, так что вы сможете отделить удачные от неудачных.

from capskip import (
    AsyncCapSkip, ApiException, NetworkException,
    TimeoutException, ValidationException,
)

results = await asyncio.gather(*tasks, return_exceptions=True)

for target, r in zip(targets, results):
    if isinstance(r, TimeoutException):
        print("timed out, worth retrying:", target)
    elif isinstance(r, ApiException):
        print("api rejected this one:", target, r)
    elif isinstance(r, NetworkException):
        print("solver unreachable, stop the run:", target)
    elif isinstance(r, Exception):
        raise r
    else:
        print("ok:", r["code"][:40])

Четыре типа исключений одинаковы во всех SDK CapSkip, и все они наследуются от базового CapSkipError , если вам удобнее ловить один тип. ValidationException означает, что ваши параметры неверны и повтор завершится точно так же. NetworkException обычно означает, что приложение не запущено, а это проблема всего запуска, а не отдельной цели.

Таймауты и опрос, которые ведут себя по-разному для разных типов

Действуют два отдельных таймаута, и пакет со смешанными типами подчиняется обоим.

Параметр	По умолчанию	Применяется к
defaultTimeout	120 секунд	Графические капчи
recaptchaTimeout	300 секунд	reCAPTCHA, Turnstile, GeeTest
pollingInterval	5 секунд	The максимальный интервал между опросами

pollingInterval стоит понять до того, как его настраивать. SDK не опрашивает с постоянным интервалом. Он начинает с 250 мс и постепенно увеличивает паузу до заданного вами значения — именно поэтому решение через SDK обычно возвращается раньше, чем самописный цикл, построенный по совету сырого API «подождите, затем опрашивайте каждые пять секунд». Увеличение значения заставляет быстрые решения приходить позже. Уменьшение добавляет запросов без всякой пользы.

Задачи GeeTest истекают, поэтому не собирайте пакет заранее

Это особенно больно бьёт именно при переходе к параллельной работе. Значение gt у GeeTest постоянно для сайта, а вот challenge одноразовый и истекает примерно за минуту. Если сначала собрать пятьдесят задач и только потом начать решать, те, что в конце очереди, умрут ещё до отправки. Получайте каждую задачу непосредственно перед тем решением, которое её использует.

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

# pip install capskip
import asyncio
from capskip import AsyncCapSkip, ApiException, TimeoutException

TARGETS = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

async def solve_one(solver, sem, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)
    sem = asyncio.Semaphore(5)

    tasks = [solve_one(solver, sem, k, u) for k, u in TARGETS]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    tokens = {}
    for (sitekey, url), r in zip(TARGETS, results):
        if isinstance(r, (ApiException, TimeoutException)):
            print("failed:", url, r)
        else:
            tokens[url] = r["code"]

    print(len(tokens), "of", len(TARGETS), "solved")
    return tokens

asyncio.run(main())

Вот и весь приём: один общий клиент, семафор, return_exceptions=True, и словарь токенов в конце. Вставьте это в парсер, и шаг с капчей перестанет быть узким местом. Страница распознавание капчи для веб-скрейпинга описывает, как это встраивается в более широкий конвейер.

Та же идея в остальных SDK

Если вы переносите этот код, примитив параллелизма меняется, а форма — нет. Node и так неблокирующий, поэтому обычного клиента достаточно.

// npm install capskip
const { CapSkip } = require('capskip');

const solver = new CapSkip({ host: '127.0.0.1', port: 8080 });

// AsyncCapSkip here is just an alias. Promise.all does the work.
const results = await Promise.all([
  solver.recaptcha('SITEKEY_A', 'https://example.com/page-a'),
  solver.turnstile('SITEKEY_B', 'https://example.com/page-b'),
]);

console.log(results.map(r => r.code));

В .NET всё то же самое с Task.WhenAll, а PHP не предлагает параллелизма вовсе. Если вам нужно параллельное решение и язык выбираете вы, Python — тот, где есть специально сделанный клиент. Страница Решатель капчи для Python описывает остальную часть интерфейса.

Частые ошибки

Ошибка	Что происходит	Исправить
Использование CapSkip внутри корутин	Блокирует цикл событий, поэтому время остаётся последовательным	Использовать AsyncCapSkip
Нет семафора	Каждое решение замедляется, как только очередь становится глубокой	Ограничьте число задач в работе, начните примерно с 5
Обычный gather	Одна ошибка отбрасывает все остальные результаты	return_exceptions=True
Предварительная выборка задач GeeTest	Поздние истекают до отправки	Получайте каждую прямо перед решением
Увеличение pollingInterval	Быстрые решения возвращаются позже, а не раньше	Оставьте значение по умолчанию
Прокси для решения графической капчи	Для графических капч не поддерживается	Прокси применимы только к reCAPTCHA, Turnstile и GeeTest

Сырой запрос и ответ для каждого типа, если вы хотите увидеть, что именно отправляет SDK, есть на странице Документация по API. См. документацию Python: справочник по задачам asyncio охватывает gather подробно описывает семантику.

Часто задаваемые вопросы

Сколько капч я могу решать одновременно?

Квоты, в которую можно упереться, нет, поэтому предел — ваше собственное железо. Распознавание идёт локально, так что параллелизм ограничен процессором, а не тарифом аккаунта. Начните с пяти задач в работе, замерьте фиксированный пакет и подстройтесь по результату.

Можно ли смешивать типы капчи в одном вызове gather?

Да. recaptcha, turnstile, geetest и normal — это корутины на одном и том же клиенте, и их можно ожидать вместе. Помните, что графические решения используют таймаут 120 секунд, а остальные — 300.

Может, лучше использовать потоки?

Только если окружающий код уже многопоточный. Работа здесь — это ожидание ввода-вывода, ровно то, для чего создан asyncio, и один цикл событий дешевле пула потоков. Если вы застряли в синхронной кодовой базе, пул потоков вокруг обычного CapSkip клиента тоже подойдёт.

Нужно ли закрывать AsyncCapSkip?

В SDK не документированы ни метод close, ни асинхронный контекстный менеджер, поэтому создайте один клиент, используйте его весь запуск и дайте ему выйти из области видимости при завершении процесса.

Сводка

Использовать AsyncCapSkip, разделяйте один клиент, ограничьте число одновременных решений семафором и передайте return_exceptions=True , чтобы одна плохая цель не отбросила весь пакет. Так очередь капч превращается из последовательного узкого места в одно ожидание.

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