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 kein.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,
ValidationErrorlangsung sebelum request terkirim. - Parse respons manual —
dict["key"]rawanKeyError; 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:
ValidationErrorpada sitekey yang terlihat valid — sitekey lebih pendek dari 20 karakter; sesuaikanmin_lengthkalau target pakai key lebih pendek.ValidationErrordi pageurl — URL tanpa skema; tambahkan awalanhttps://.- 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.