Pydantic Doğrulamalı CaptchaAI Python İstemcisi

CaptchaAI API'sine boş bir site anahtarı iletmek gidiş-dönüş israfına neden olur - yanıt bekledikten sonra ERROR_WRONG_CAPTCHA_ID alırsınız.Pydantic bu hataları HTTP çağrısı gerçekleşmeden önce yakalar: şifreli API hata kodları yerine doğrulama hatalarını temizleyin.

CAPTCHA API İstemcileri için Neden Pydantic

Modeller

# 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")

Müşteri

# 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)

Kullanım

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}")

Bağımlılıkları yükleyin:

pip install pydantic requests

Sorun giderme

SSS

Pydantic doğrulama ek yük getiriyor mu?

İhmal edilebilir - doğrulama çağrısı başına mikrosaniye ve API gidiş dönüşleri için saniye. Ağ aramalarından önce geçersiz parametrelerin yakalanmasıyla kazanılan zaman, doğrulama maliyetinden çok daha fazladır.

Bunu eşzamansız (httpx) ile kullanabilir miyim?

Evet. requests'yi httpx.AsyncClient ile değiştirin ve _submit, _poll ve çözücü yöntemlerini async yapın. Pydantic modelleri aynı kalır; eşzamansız HTTP çağrısından önce eşzamanlı olarak doğrulanırlar.

Modelleri yeni CAPTCHA türleri için nasıl genişletirim?

Gerekli alanlar ve to_params() yöntemiyle yeni bir BaseModel alt sınıfı oluşturun. Modeli başlatan ve _solve'yi çağıran istemci sınıfına karşılık gelen bir çözücü yöntemi ekleyin.

İlgili Makaleler

• Python Oyun Yazarı Captchaai Tam Kılavuzu
• İstemci Captcha Boru Hatları Oluşturma Captchaai
• Captchaai Webhook Güvenliği Geri Arama Doğrulaması

Sonraki Adımlar

Doğrulanmış bir CaptchaAI istemcisi oluşturun —API anahtarınızı alınve Pydantic modellerini ekleyin.

İlgili kılavuzlar:

• CaptchaAI API için Python Sarmalayıcı Oluşturma
• CaptchaAI Hata Kodları Referansı
• CaptchaAI JSON API ve Form API karşılaştırması