Интеграция HTTPX + CaptchaAI

Если один скрипт то упирается в блокирующий requests, то не успевает за конкурентными задачами в asyncio, дело обычно не в CaptchaAI, а в HTTP-клиенте. httpx — один из немногих клиентов Python, который одинаково хорошо работает в синхронном и асинхронном режиме, совместим по API с requests и умеет HTTP/2 из коробки. Ниже — два готовых класса для решения CAPTCHA через CaptchaAI: синхронный для простых скриптов и асинхронный для конкурентного парсинга, плюс пример HTTP/2 и рабочий сценарий с реальной страницей.

Требования

Прежде чем подключать CaptchaAI, проверьте версии окружения:

pip install httpx

httpx, requests или aiohttp: что выбрать

Прежде чем копировать код клиента ниже, стоит понять, какой инструмент реально нужен для вашего кейса с CaptchaAI.

Если код уже написан на requests и нужно быстро добавить решение CAPTCHA — берите синхронный клиент httpx, миграция займёт минуты. Если планируется отправлять и опрашивать десятки задач CaptchaAI параллельно — сразу пишите на httpx.AsyncClient, а не переходите на него позже.

Синхронный клиент: минимальный рабочий вариант

Для одиночных запросов или скриптов, где параллелизм не нужен, синхронного клиента достаточно. Он отправляет задачу на in.php, затем опрашивает res.php с фиксированным интервалом, пока не придёт готовый токен или не истечёт тайм-аут.

import httpx
import time
import os


class CaptchaAISync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.Client(timeout=30)

    def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = self.client.get(f"{self.base_url}/in.php", params=params)
        text = resp.text

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        task_id = text.split("|")[1]

        # Poll
        deadline = time.time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while time.time() < deadline:
            time.sleep(5)
            result = self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    def get_balance(self):
        resp = self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    def close(self):
        self.client.close()


# Usage
solver = CaptchaAISync(os.environ["CAPTCHAAI_API_KEY"])

token = solver.solve({
    "method": "userrecaptcha",
    "googlekey": "6Le-wvkS...",
    "pageurl": "https://example.com",
})
print(f"Token: {token[:50]}...")
solver.close()

Обратите внимание на time.sleep(5) между опросами: это не магическое число, а разумный баланс между скоростью ответа и лишними запросами к res.php. Уменьшать интервал ниже пары секунд обычно бессмысленно — решение CAPTCHA всё равно занимает время на стороне воркеров.

Асинхронный клиент: решаем капчи параллельно

Как только задач становится больше одной за раз — например, парсер обходит десятки страниц с CAPTCHA — синхронный клиент превращается в очередь. Асинхронная версия на httpx.AsyncClient отправляет и опрашивает несколько задач конкурентно, не блокируя поток на каждом ожидании.

import httpx
import asyncio
import os


class CaptchaAIAsync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.AsyncClient(timeout=30)

    async def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = await self.client.get(
            f"{self.base_url}/in.php", params=params
        )
        text = resp.text

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        task_id = text.split("|")[1]

        # Poll
        deadline = asyncio.get_event_loop().time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)
            result = await self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    async def get_balance(self):
        resp = await self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    async def close(self):
        await self.client.aclose()


# Usage
async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])

    # Solve multiple concurrently
    tasks = [
        solver.solve({
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": f"https://example.com/page{i}",
        })
        for i in range(5)
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Page {i}: FAILED - {r}")
        else:
            print(f"Page {i}: solved ({len(r)} chars)")

    await solver.close()

asyncio.run(main())

asyncio.gather(..., return_exceptions=True) здесь принципиален: одна упавшая CAPTCHA не должна обрушивать весь пакет — вы получаете список результатов вперемешку с исключениями и разбираете их по отдельности. Сколько задач реально можно держать в работе одновременно, зависит от тарифа CaptchaAI: план BASIC ($15/мес, 5 потоков) подходит для тестов и небольших пакетов, а для стабильного пула из полусотни конкурентных задач нужен как минимум план уровня ADVANCE ($90/мес, 50 потоков) — считайте по своей нагрузке, а не наугад.

HTTP/2: меньше накладных расходов при массовом опросе

HTTPX поддерживает HTTP/2, и это не просто галочка в таблице: мультиплексирование запросов по одному TCP-соединению заметно снижает накладные расходы, когда клиент опрашивает res.php десятки раз подряд.

pip install httpx[http2]

client = httpx.AsyncClient(http2=True, timeout=30)

Для команд, чья инфраструктура развёрнута в Европе или Казахстане, а серверы CaptchaAI — в другом регионе, HTTP/2 особенно заметен на нестабильных или мобильных каналах: меньше повторных TCP-хендшейков означает меньше шансов словить лишний тайм-аут при опросе.

Парсинг с решением CAPTCHA: рабочий пример

Ниже — сценарий, с которого обычно начинают: страница отдаёт HTML с reCAPTCHA, скрипт вытаскивает sitekey, решает CAPTCHA через CaptchaAI и отправляет форму с готовым токеном.

import httpx
import re
import os

async def scrape_with_captcha(url, solver):
    async with httpx.AsyncClient() as client:
        # Fetch page
        resp = await client.get(url)
        html = resp.text

        # Check for reCAPTCHA
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        if not match:
            return html

        site_key = match.group(1)
        token = await solver.solve({
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

        # Submit form with token
        resp = await client.post(url, data={
            "g-recaptcha-response": token,
        })
        return resp.text


async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])
    content = await scrape_with_captcha("https://example.com", solver)
    print(f"Got {len(content)} chars")
    await solver.close()

asyncio.run(main())

Если скрипт в процессе парсинга собирает данные, которые можно отнести к персональным (контакты, профили пользователей), для читателей из РФ действуют требования 152-ФЗ «О персональных данных», а для трансграничных проектов — обычная GDPR-осторожность: собирайте только те данные, на обработку которых у вас есть основание, и не храните лишнего. Это вопрос организации пайплайна, а не самого решения CAPTCHA.

Частые вопросы

httpx или requests: что выбрать для проекта с CaptchaAI?

Для нового проекта — httpx: API почти идентичен requests, но добавляет асинхронность и HTTP/2 без смены библиотеки в будущем. Для существующего кода на requests миграция синхронного клиента, как в примере выше, обычно занимает не больше часа.

aiohttp или httpx.AsyncClient быстрее при решении сотен CAPTCHA?

На чисто асинхронной нагрузке aiohttp может показывать немного меньшие накладные расходы. Но при опросе CaptchaAI через HTTP/2 и в смешанном sync/async-коде httpx на практике удобнее и не уступает в скорости.

Как настроить тайм-аут и повторные попытки при опросе res.php?

В примерах выше timeout=300 — общий бюджет времени на задачу, а time.sleep(5) / asyncio.sleep(5) — интервал между опросами res.php. Если задачи регулярно не укладываются в 300 секунд, сначала проверьте тип CAPTCHA и её ожидаемое время решения, а не просто увеличивайте тайм-аут вслепую.

Сколько потоков CaptchaAI нужно для пула AsyncClient?

Один поток CaptchaAI — это одна CAPTCHA, решаемая параллельно; освободившийся поток сразу берёт следующую задачу. Число конкурентных корутин в asyncio.gather имеет смысл ограничивать под фактический тариф — держать 50 задач одновременно на тарифе с 5 потоками бессмысленно, лишние запросы просто встанут в очередь.

Работает ли httpx с HTTP/2 без дополнительных зависимостей?

Нет — базовая установка pip install httpx даёт только HTTP/1.1. Для HTTP/2 нужен отдельный экстра-пакет pip install httpx[http2], как показано в разделе выше.

Похожие материалы

• aiohttp + CaptchaAI: асинхронное решение CAPTCHA
• Axios + CaptchaAI без браузера
• Парсинг с CAPTCHA на Python: полное руководство