Pydantic Doğrulamalı CaptchaAI Python İstemcisi

Doğru kurgulanmış bir istemci, CAPTCHA hatalarını üretimden editörünüze taşır. CaptchaAI API'si geçersiz bir sitekey ya da şemasız bir pageurl aldığında bunu ancak saniyeler sonra ERROR_WRONG_CAPTCHA_ID gibi şifreli bir kodla bildirir; Pydantic ise aynı hatayı istek hiç gönderilmeden, modeli kurarken yakalar. Bu rehberde reCAPTCHA v2/v3, Cloudflare Turnstile ve görüntü CAPTCHA'ları için parametreleri doğrulayan, yanıtları tiplenmiş modellere ayrıştıran üretime hazır bir Python istemcisi kuracaksınız.

Kuracağınız istemci üç işi tek yerde toplar:

• Gönderim öncesi parametre doğrulaması (sitekey, pageurl ve tür bazlı alanlar).
• API yanıtlarının tiplenmiş modellere ayrıştırılması — dict["key"] ve KeyError yok.
• Gönder, sorgula ve sonucu döndür adımlarını saran temiz bir çözüm arayüzü.

Neden Pydantic ile doğrulama?

Kısa cevap: doğrulama katmanı, çalışma zamanı hatalarını yazma zamanına çeker. CaptchaAI'ye giden her istek bir HTTP çağrısıdır ve parametrelerin hatalı olduğunu gönderdikten sonra öğrenmek hem süre hem thread kaybettirir. Pydantic modelleri isteği daha oluştururken doğrular; googlekey boş ya da pageurl şemasız olduğunda kod ağ katmanına hiç ulaşmaz. Şifreli ERROR_* kodlarını okumak yerine, alan adını ve nedeni söyleyen okunaklı doğrulama hatalarıyla çalışırsınız.

Müşteri otomasyonu geliştiren bir ekip için bu fark doğrudan maliyete yansır. CaptchaAI thread bazlı faturalandırır — örneğin BASIC ($15/ay, 5 thread) planında aynı anda 5 CAPTCHA çözebilirsiniz. Yerelde doğrulama iki yerde kazandırır:

• Thread kullanımı: boş bir sitekey yüzünden boşa giden her istek, o thread'i saniyelerce meşgul eder.
• Hata ayıklama: hatalar ERROR_* kodları yerine alan adını söyleyen okunaklı mesajlar olarak gelir.

Parametreleri CI aşamasında doğrulamak bu israfı erkenden keser; freelance otomasyon işlerinde teslim tarihine yetişirken bu güvenlik ağı özellikle değerlidir.

İstek ve yanıt modelleri

Her CAPTCHA türü kendi BaseModel alt sınıfıyla temsil edilir. Alan kısıtlarını (min_length, max_length) doğrudan Field içinde tanımlar, özel kuralları field_validator ile eklersiniz. to_params() yöntemi doğrulanmış modeli CaptchaAI'nin beklediği form parametrelerine çevirir; böylece method, googlekey, sitekey ve pageurl gibi parametre adları tek bir yerde toplanır ve çağrı tarafına dağılmaz.

Modeller iki gruba ayrılır:

• İstek modelleri: RecaptchaV2Request, RecaptchaV3Request, TurnstileRequest ve ImageRequest — her biri kendi alanlarını doğrular.
• Yanıt modelleri: SubmitResponse, PollResponse ve SolveResult — ham JSON'u tiplenmiş nesnelere dönüştürür ve success, token gibi türetilmiş özellikleri açar.

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

İstemci sınıfı

İstemci sınıfı çözüm akışının üç adımını sarmalar:

• _submit — görevi in.php uç noktasına gönderir ve görev kimliğini döndürür.
• _poll — sonucu poll_interval aralığıyla periyodik sorgular; timeout süresi dolarsa zaman aşımı hatası verir.
• _solve — bu ikisini birleştirir ve toplam çözüm süresini ölçer.

Her genel yöntem — solve_recaptcha_v2, solve_turnstile ve diğerleri — önce ilgili modeli kurar; yani doğrulama her zaman ağ çağrısından önce çalışır. API tarafında bir sorun çıktığında CaptchaAIError yükselir ve hata kodunu tek bir tipte toplarsınız.

# 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 örneği

Aşağıdaki örnek istemcinin üç davranışını bir arada gösterir:

• Geçerli istek: doğrulamayı geçer ve API'yi çağırarak token'ı döndürür.
• Boş sitekey: yerelde ValidationError ile durur, hiç ağ çağrısı yapılmaz.
• API hatası: sorun API tarafındaysa CaptchaAIError yükselir.

staging.example.com/qa-login gibi bir test hedefi kullanmak, üretim trafiğine dokunmadan akışı doğrulamanızı sağlar.

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

Sık sorulan sorular

Pydantic doğrulaması isteği yavaşlatır mı?

Pratikte hayır. Doğrulama çağrı başına mikrosaniyelerle ölçülürken bir API gidiş-dönüşü saniyeler sürer. Geçersiz parametreleri ağ çağrısından önce yakalayarak kazandığınız süre, doğrulama maliyetinin çok üzerindedir.

Bu istemciyle hangi CAPTCHA türlerini çözebilirim?

Örnek modeller reCAPTCHA v2, reCAPTCHA v3 ve Cloudflare Turnstile ile görüntü/OCR CAPTCHA'sını kapsar. CaptchaAI ayrıca GeeTest v3, grid ve BLS CAPTCHA'larını da çözer; CaptchaFox (beta), Friendly Captcha (beta) ve Lemin (beta) desteklenir. hCaptcha ve FunCaptcha (Arkose Labs) şu anda desteklenmez, GeeTest v4 ise çok yakında.

Geçersiz bir sitekey neden API'ye ulaşmadan hata veriyor?

Çünkü doğrulama, Pydantic modelinin kurucusunda çalışır. RecaptchaV2Request(sitekey="") çağrısı min_length=20 kısıtını ihlal ettiği için ValidationError fırlatır ve to_params() hiç çağrılmaz — hatalı istek in.php'ye gitmez.

Pydantic v1 ile import hatası alıyorum, nasıl çözerim?

Bu istemci Pydantic v2 API'sini (field_validator, model_validate) kullanır. Ortamınızda v1 varsa yükseltin: pip install 'pydantic>=2.0'. v1 ve v2 aynı projede karışırsa doğrulama davranışı beklenmedik biçimde değişir.

Aynı istemciyi async (httpx) ile kullanabilir miyim?

Evet. requests yerine httpx.AsyncClient kullanın ve _submit, _poll ile çözücü yöntemleri async yapın. Pydantic modelleri aynı kalır — HTTP çağrısından önce senkron olarak doğrularlar.

İlgili makaleler

• Python ve Playwright ile CaptchaAI entegrasyonu
• İstemci taraflı CAPTCHA işlem hatları kurma
• CaptchaAI webhook güvenliği ve callback doğrulama

Sonraki adımlar

Doğrulanmış bir CaptchaAI istemcisini bugün kurun — API anahtarınızı alın ve yukarıdaki Pydantic modellerini projenize ekleyin.

İlgili kılavuzlar:

• CaptchaAI API için Python sarmalayıcı kütüphanesi
• CaptchaAI hata kodları başvurusu
• CaptchaAI JSON API ile Form API karşılaştırması