Клиент CaptchaAI Python с проверкой Pydantic

Если тестовый запрос к CaptchaAI падает с ERROR_WRONG_CAPTCHA_ID уже после того, как вы прождали ответ несколько секунд, причина почти всегда не в API, а в параметре, который вообще не стоило отправлять по сети: пустой sitekey, URL без схемы, base64-строка короче ожидаемой длины. Pydantic ловит такие ошибки на уровне кода, до HTTP-вызова — вместо загадочного кода ошибки от API вы получаете понятное исключение ValidationError с указанием, какое именно поле не прошло проверку.

Зачем добавлять Pydantic в CaptchaAI-клиент на Python

Без валидации ошибка в параметре обнаруживается только после round-trip к API — вы теряете время на ожидание ответа, чтобы получить код ошибки вместо результата. Pydantic переносит проверку на уровень кода, до сети:

Ключевая экономия — не в микросекундах самой валидации, а в отменённых сетевых вызовах: каждый пропущенный round-trip к in.php — это несколько секунд ожидания, которые Pydantic позволяет не тратить.

Модели данных для типов CAPTCHA

Каждому поддерживаемому типу CAPTCHA — reCAPTCHA v2/v3, Cloudflare Turnstile, image/OCR — соответствует своя Pydantic-модель с полями и ограничениями, которые CaptchaAI ожидает на in.php. Метод to_params() конвертирует модель в словарь параметров запроса, а SubmitResponse/PollResponse разбирают ответ res.php в типизированный объект.

# models.py
from pydantic import BaseModel, Field, field_validator, HttpUrl
from enum import Enum
from typing import Optional

class CaptchaMethod(str, Enum):
    RECAPTCHA_V2 = "userrecaptcha"
    RECAPTCHA_V3 = "userrecaptcha"  # Differentiated by version field
    TURNSTILE = "turnstile"
    HCAPTCHA = "hcaptcha"
    IMAGE = "base64"
    GEETEST = "geetest"

class RecaptchaV2Request(BaseModel):
    """Parameters for solving reCAPTCHA v2."""
    sitekey: str = Field(min_length=20, max_length=100, description="Site's reCAPTCHA sitekey")
    pageurl: HttpUrl = Field(description="URL where CAPTCHA appears")
    invisible: bool = False
    cookies: Optional[str] = None

    @field_validator("sitekey")
    @classmethod
    def validate_sitekey(cls, v: str) -> str:
        if v.strip() != v:
            raise ValueError("Sitekey must not have leading/trailing whitespace")
        return v

    def to_params(self) -> dict:
        params = {
            "method": "userrecaptcha",
            "googlekey": self.sitekey,
            "pageurl": str(self.pageurl),
        }
        if self.invisible:
            params["invisible"] = "1"
        if self.cookies:
            params["cookies"] = self.cookies
        return params

class RecaptchaV3Request(BaseModel):
    """Parameters for solving reCAPTCHA v3."""
    sitekey: str = Field(min_length=20, max_length=100)
    pageurl: HttpUrl
    action: str = Field(default="verify", min_length=1, max_length=100)

    def to_params(self) -> dict:
        return {
            "method": "userrecaptcha",
            "version": "v3",
            "googlekey": self.sitekey,
            "pageurl": str(self.pageurl),
            "action": self.action,
        }

class TurnstileRequest(BaseModel):
    """Parameters for solving Cloudflare Turnstile."""
    sitekey: str = Field(min_length=10, max_length=100)
    pageurl: HttpUrl
    action: Optional[str] = None
    cdata: Optional[str] = None

    def to_params(self) -> dict:
        params = {
            "method": "turnstile",
            "sitekey": self.sitekey,
            "pageurl": str(self.pageurl),
        }
        if self.action:
            params["action"] = self.action
        if self.cdata:
            params["data"] = self.cdata
        return params

class ImageRequest(BaseModel):
    """Parameters for solving image/text CAPTCHA."""
    base64_image: str = Field(min_length=100, description="Base64-encoded image")
    case_sensitive: bool = False
    min_length: Optional[int] = Field(default=None, ge=1, le=50)
    max_length: Optional[int] = Field(default=None, ge=1, le=50)

    @field_validator("base64_image")
    @classmethod
    def validate_base64(cls, v: str) -> str:
        # Strip data URI prefix if present
        if v.startswith("data:"):
            parts = v.split(",", 1)
            if len(parts) == 2:
                return parts[1]
        return v

    def to_params(self) -> dict:
        params = {
            "method": "base64",
            "body": self.base64_image,
        }
        if self.case_sensitive:
            params["regsense"] = "1"
        if self.min_length is not None:
            params["min_len"] = str(self.min_length)
        if self.max_length is not None:
            params["max_len"] = str(self.max_length)
        return params

class SubmitResponse(BaseModel):
    """Parsed API submit response."""
    status: int
    request: str

    @property
    def success(self) -> bool:
        return self.status == 1

    @property
    def task_id(self) -> str:
        if not self.success:
            raise ValueError(f"No task ID — submission failed: {self.request}")
        return self.request

class PollResponse(BaseModel):
    """Parsed API poll response."""
    status: int
    request: str

    @property
    def ready(self) -> bool:
        return self.request != "CAPCHA_NOT_READY"

    @property
    def success(self) -> bool:
        return self.status == 1

    @property
    def token(self) -> str:
        if not self.success:
            raise ValueError(f"No token — solve failed: {self.request}")
        return self.request

class SolveResult(BaseModel):
    """Result of a successful solve."""
    token: str
    task_id: str
    solve_time: float = Field(description="Solve time in seconds")

Валидатор validate_sitekey в RecaptchaV2Request отсеивает пробелы по краям строки — распространённая причина ERROR_WRONG_CAPTCHA_ID, когда sitekey скопирован из HTML вместе с переносом строки. Похожим образом validate_base64 в ImageRequest сам убирает префикс data:...;base64,, если вы передали изображение как есть из атрибута src.

Класс клиента CaptchaAI

Клиент отвечает за отправку задачи на in.php, опрос res.php с интервалом poll_interval и разбор обоих ответов через модели SubmitResponse/PollResponse. Исключение CaptchaAIError оборачивает код ошибки, который вернул сам API, а ValidationError — исключение Pydantic, которое наступает раньше, ещё при создании объекта запроса.

# client.py
import time
import requests
from pydantic import ValidationError

from models import (
    RecaptchaV2Request,
    RecaptchaV3Request,
    TurnstileRequest,
    ImageRequest,
    SubmitResponse,
    PollResponse,
    SolveResult,
)

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"

class CaptchaAIError(Exception):
    def __init__(self, code: str, message: str = ""):
        self.code = code
        super().__init__(f"{code}: {message}" if message else code)

class CaptchaAI:
    def __init__(self, api_key: str, poll_interval: int = 5, timeout: int = 180):
        if not api_key or len(api_key) < 10:
            raise ValueError("Invalid API key")
        self.api_key = api_key
        self.poll_interval = poll_interval
        self.timeout = timeout

    def _submit(self, params: dict) -> str:
        params["key"] = self.api_key
        params["json"] = 1

        resp = requests.post(SUBMIT_URL, data=params, timeout=30)
        result = SubmitResponse.model_validate(resp.json())

        if not result.success:
            raise CaptchaAIError(result.request, "Submit failed")

        return result.task_id

    def _poll(self, task_id: str) -> str:
        start = time.monotonic()

        while time.monotonic() - start < self.timeout:
            time.sleep(self.poll_interval)

            resp = requests.get(RESULT_URL, params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1,
            }, timeout=15)

            result = PollResponse.model_validate(resp.json())

            if not result.ready:
                continue

            if result.success:
                return result.token

            raise CaptchaAIError(result.request, "Solve failed")

        raise CaptchaAIError("TIMEOUT", f"Task {task_id} timed out after {self.timeout}s")

    def _solve(self, params: dict) -> SolveResult:
        start = time.monotonic()
        task_id = self._submit(params)
        token = self._poll(task_id)
        elapsed = time.monotonic() - start

        return SolveResult(
            token=token,
            task_id=task_id,
            solve_time=round(elapsed, 1),
        )

    def solve_recaptcha_v2(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
        """Solve reCAPTCHA v2 with validated parameters."""
        req = RecaptchaV2Request(sitekey=sitekey, pageurl=pageurl, **kwargs)
        return self._solve(req.to_params())

    def solve_recaptcha_v3(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
        """Solve reCAPTCHA v3 with validated parameters."""
        req = RecaptchaV3Request(sitekey=sitekey, pageurl=pageurl, **kwargs)
        return self._solve(req.to_params())

    def solve_turnstile(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
        """Solve Cloudflare Turnstile with validated parameters."""
        req = TurnstileRequest(sitekey=sitekey, pageurl=pageurl, **kwargs)
        return self._solve(req.to_params())

    def solve_image(self, base64_image: str, **kwargs) -> SolveResult:
        """Solve image/text CAPTCHA with validated parameters."""
        req = ImageRequest(base64_image=base64_image, **kwargs)
        return self._solve(req.to_params())

    def get_balance(self) -> float:
        """Get current account balance."""
        resp = requests.get(RESULT_URL, params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        }, timeout=10)
        result = SubmitResponse.model_validate(resp.json())
        return float(result.request)

Обратите внимание: _solve() — единая точка входа для всех типов CAPTCHA, поэтому логика опроса и обработки ошибок написана один раз, а не дублируется в каждом solve_*-методе.

Пример использования клиента

Ниже — типичный сценарий: успешное решение reCAPTCHA v2, отклонённый Pydantic пустой sitekey и ошибка API при решении Turnstile.

from pydantic import ValidationError
from client import CaptchaAI, CaptchaAIError

client = CaptchaAI("YOUR_API_KEY", timeout=120)

# Valid request — passes validation, calls API
result = client.solve_recaptcha_v2(
    sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
    pageurl="https://staging.example.com/qa-login",
)
print(f"Token: {result.token[:40]}...")
print(f"Solved in {result.solve_time}s")

# Invalid sitekey — caught immediately, no API call
try:
    client.solve_recaptcha_v2(sitekey="", pageurl="https://example.com")
except ValidationError as e:
    print(e)
    # sitekey: String should have at least 20 characters

# Invalid score — caught before API call
try:
    client.solve_recaptcha_v3(
        sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
        pageurl="https://example.com",
    )
except ValidationError as e:
    print(e)

# API error — caught during request
try:
    result = client.solve_turnstile(
        sitekey="0x4AAAAAAADnPIDROrmt1Wwj",
        pageurl="https://example.com",
    )
except CaptchaAIError as e:
    print(f"API error: {e.code}")

Установите зависимости перед первым запуском:

pip install pydantic requests

Пример из практики: пакетная проверка перед парсингом маркетплейса

Команды, которые парсят карточки товаров или объявления — маркетплейсы, агрегаторы бронирования, сервисы мониторинга цен — обычно запускают десятки воркеров параллельно. Если один воркер отправит пустой pageurl из-за бага в конфиге, без валидации это заметят только через минуту, когда истечёт таймаут res.php. С Pydantic такой воркер падает с ValidationError в момент старта, до того как займёт поток в тарифном плане.

Это особенно ощутимо на потоковом тарифе: план BASIC ($15/мес, 5 потоков) даёт всего 5 одновременных задач, и один воркер, зависший на невалидном запросе, — это пятая часть доступной параллельности впустую. На ADVANCE ($90/мес, 50 потоков) относительные потери ниже, но абсолютные секунды решения, которые уходят на пустые попытки, растут вместе с объёмом. Валидация на входе — простой способ не платить за них.

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

Типичные ошибки валидации и их причины

Большинство ValidationError в этом клиенте сводятся к пяти причинам:

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

Замедляет ли Pydantic решение CAPTCHA?

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

Как подключить модели к асинхронному клиенту на httpx?

Замените requests на httpx.AsyncClient, а методы _submit, _poll и solve_* сделайте async def. Модели Pydantic не меняются — они проверяют параметры синхронно, до того как выполнится асинхронный HTTP-вызов.

Почему ValidationError не попадает в except CaptchaAIError?

Это два разных исключения на разных этапах. ValidationError — из Pydantic, возникает при создании объекта запроса (RecaptchaV2Request(...)), ещё до сети. CaptchaAIError — из клиента, оборачивает код ошибки, который вернул сам in.php/res.php. В коде, вызывающем solve_*, стоит ловить оба исключения отдельно, как показано в разделе «Пример использования клиента».

Как заранее поймать нехватку баланса, не дожидаясь ERROR_ZERO_BALANCE?

Вызовите client.get_balance() перед батчем задач и сравните результат с ожидаемым расходом — например, объёмом потоков на вашем тарифе. Параметров у этого запроса нет, поэтому Pydantic-модель запроса тут не нужна, но ответ всё равно разбирается через SubmitResponse, так что структура ответа остаётся типизированной.

Следующие шаги

• Быстрый старт CaptchaAI: первое решение CAPTCHA за 5 минут
• Как решить reCAPTCHA v2 через API: пошаговое руководство
• Как решить Cloudflare Turnstile через API
• Как решить GeeTest v3 через API