Meneruskan sitekey kosong ke API CaptchaAI membuang waktu — Anda mendapat ERROR_WRONG_CAPTCHA_ID setelah menunggu respons. Pydantic menangkap error ini sebelum HTTP call terjadi: dapatkan ValidationError yang jelas alih-alih kode error API yang samar.
Mengapa Pydantic untuk Klien API CAPTCHA
| Tanpa Pydantic | Dengan Pydantic |
|---|---|
| Sitekey kosong → error API setelah 5 detik | ValidationError langsung |
Parse respons via dict["key"] → KeyError |
Model bertipe dengan default dan validasi |
| Tidak ada autocomplete IDE untuk parameter | Type hint lengkap di semua field |
Model
# 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}")
Pemecahan Masalah
| Masalah | Penyebab | Solusi |
|---|---|---|
ValidationError pada sitekey yang terlihat valid |
Sitekey terlalu pendek (<20 karakter) | Cek panjang sitekey; sesuaikan min_length jika target Anda menggunakan key yang lebih pendek |
ValidationError di pageurl |
Skema URL tidak ada | Sertakan awalan https:// |
| Validasi Base64 gambar gagal | String terlalu pendek atau mengandung awalan data: |
Validator menghapus awalan data: secara otomatis; pastikan konten base64 sebenarnya >100 karakter |
CaptchaAIError: ERROR_ZERO_BALANCE |
Dana tidak mencukupi | Isi ulang di dashboard CaptchaAI |
| Error import Pydantic v1 | Versi Pydantic salah | Gunakan Pydantic v2: pip install 'pydantic>=2.0' |
Pertanyaan Umum
Apakah validasi Pydantic menambah overhead?
Dapat diabaikan — mikrodetik per validasi vs. detik untuk round-trip API. Waktu yang dihemat dari menangkap parameter tidak valid sebelum network call jauh melebihi biaya validasi.
Bisakah saya menggunakan ini dengan async (httpx)?
Ya. Ganti requests dengan httpx.AsyncClient dan buat _submit, _poll, serta metode solver menjadi async. Model Pydantic tetap sama — mereka memvalidasi secara sinkron sebelum HTTP call async.
Bagaimana cara memperluas model untuk jenis CAPTCHA baru?
Buat subkelas BaseModel baru dengan field wajib dan metode to_params(). Tambahkan metode solver yang sesuai di kelas klien yang membuat instance model dan memanggil _solve.
Artikel Terkait
- Panduan Lengkap Python Playwright CaptchaAI
- Referensi Kode Error CaptchaAI
- CaptchaAI JSON API vs Form API
Bangun klien CaptchaAI yang tervalidasi — dapatkan API key Anda dan tambahkan model Pydantic.
