API Tutorials

Klien Python CaptchaAI dengan Validasi Pydantic

Skrip scraping Anda jalan mulus di laptop, lalu gagal diam-diam di produksi karena sitekey ikut ter-strip spasi saat disalin dari .env.

Intinya: Pydantic v2 memvalidasi sitekey, pageurl, dan base64 gambar sebelum satu byte pun terkirim ke CaptchaAI, jadi kesalahan ketahuan instan lewat ValidationError — bukan setelah menunggu round-trip penuh ke in.php.

Tanpa validasi klien, kesalahan seperti ini baru ketahuan setelah menunggu respons dari server CaptchaAI — entah dari region AWS ap-southeast-1 maupun GCP asia-southeast2 yang dipakai tim automation Indonesia.

Pydantic vs Dict Mentah untuk Klien API CAPTCHA

Tiga skenario yang paling sering bikin klien API CAPTCHA gagal diam-diam:

  • Sitekey kosong — tanpa Pydantic baru ketahuan setelah error API; dengan Pydantic, ValidationError langsung sebelum request terkirim.
  • Parse respons manualdict["key"] rawan KeyError; model bertipe punya default dan validasi otomatis.
  • Parameter API — tanpa autocomplete IDE untuk field request; dengan Pydantic, type hint lengkap di semua field.

Di plan ADVANCE ($90/bulan, 50 thread), request gagal-validasi tak pernah keluar dari mesin Anda.


Model Pydantic per Jenis CAPTCHA

Tiap tipe CAPTCHA yang didukung CaptchaAI punya model sendiri di models.py, siap disalin:

  • reCAPTCHA v2 dan v3
  • Cloudflare Turnstile
  • Gambar/OCR (base64)
# 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")

Klien

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)

## Penggunaan

Instal dependensi:

```bash
pip install pydantic requests
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")

# Sitekey tidak valid — tertangkap langsung, tanpa 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

# Skor tidak valid — tertangkap sebelum API call
try:
    client.solve_recaptcha_v3(
        sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
        pageurl="https://example.com",
    )
except ValidationError as e:
    print(e)

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

Error yang Sering Muncul dan Cara Memperbaikinya

Lima penyebab umum kalau error tetap muncul setelah model terpasang:

  • ValidationError pada sitekey yang terlihat valid — sitekey lebih pendek dari 20 karakter; sesuaikan min_length kalau target pakai key lebih pendek.
  • ValidationError di pageurl — URL tanpa skema; tambahkan awalan https://.
  • Validasi base64 gambar gagal — string pendek atau masih membawa awalan data:; validator otomatis menghapusnya, pastikan konten aslinya >100 karakter.
  • CaptchaAIError: ERROR_ZERO_BALANCE — saldo tidak cukup; isi ulang di dashboard CaptchaAI.
  • Error import Pydantic v1 — versi salah; pakai Pydantic v2: pip install 'pydantic>=2.0'.

Pertanyaan Umum Seputar Validasi Pydantic dan CaptchaAI

Apakah validasi Pydantic bikin klien lebih lambat?

Praktis tidak terasa — validasi makan waktu mikrodetik, satu round-trip ke in.php/res.php makan waktu detik.

Kenapa ada HCAPTCHA di enum CaptchaMethod kalau CaptchaAI tidak mendukung hCaptcha?

Entri itu hanya menjaga kelengkapan definisi method, bukan tanda dukungan aktif — tidak ada model request atau solve_hcaptcha() di klien ini. CaptchaAI menyelesaikan reCAPTCHA v2/v3, Turnstile, GeeTest v3, dan gambar; untuk hCaptcha perlu layanan lain.

Apakah HttpUrl di model ini menerima pageurl localhost untuk testing?

Ya. HttpUrl Pydantic v2 hanya mensyaratkan skema dan host valid, tanpa TLD publik, jadi http://localhost:8000/login lolos validasi.

Bisakah client ini dipakai async, misalnya di endpoint FastAPI?

Bisa. Ganti requests dengan httpx.AsyncClient, lalu ubah _submit, _poll, dan tiap solve_* jadi async def. Model Pydantic-nya tidak berubah.

Kenapa RecaptchaV3Request di atas tidak punya field min_score?

Karena API CaptchaAI tidak punya parameter target skor minimum — Google yang menentukan skor secara otomatis. Yang bisa Anda atur hanya sitekey, pageurl, dan action.


Artikel Terkait


Salin model dan klien di atas, lalu ambil API key CaptchaAI Anda dan jalankan solve pertama yang sudah tervalidasi.

Komentar dinonaktifkan untuk artikel ini.