Máy khách Python CaptchaAI có xác thực Pydantic

Xác thực tham số bằng Pydantic trước khi gửi request giúp bạn bắt lỗi cấu hình ngay trên máy, thay vì đợi CaptchaAI trả về ERROR_WRONG_CAPTCHA_ID sau một vòng round-trip tốn vài giây. Bài này xây dựng client Python hoàn chỉnh: model Pydantic v2 cho từng loại CAPTCHA (reCAPTCHA v2/v3, Turnstile, ảnh), class client gửi task tới in.php, polling res.php, và trả về token đã kiểm tra kiểu dữ liệu — dành cho dev cần code chạy được ngay.

Bốn bước client thực hiện, mỗi bước có model Pydantic riêng đứng chắn ở đầu vào:

1. Định nghĩa model cho từng loại CAPTCHA — sitekey, pageurl và tham số riêng được validate ngay khi khởi tạo.
2. Gửi task tới in.php với payload đã qua kiểm tra kiểu dữ liệu.
3. Polling res.php theo chu kỳ poll_interval cho đến khi có kết quả.
4. Parse token trả về thành SolveResult — không còn dò dict["key"].

Vì sao nên xác thực tham số bằng Pydantic trước khi gọi CaptchaAI

Không validate ở tầng client, một sitekey rỗng vẫn được gửi đi bình thường — bạn chỉ biết sai sau khi CaptchaAI trả lỗi. Pydantic chặn ba lớp lỗi phổ biến nhất trước khi có lệnh gọi mạng nào:

• Tham số rỗng hoặc sai kiểu — sitekey rỗng, pageurl không phải URL hợp lệ.
• Field thiếu khi parse response — response API thiếu key mong đợi, tránh KeyError khi truy cập trực tiếp bằng dict["key"].
• Nhầm lẫn tham số giữa các loại CAPTCHA — gửi tham số của reCAPTCHA cho request Turnstile.

Với team QA/scraping ở các công ty outsourcing tại TP.HCM hay Hà Nội chạy nhiều thread song song trên plan STANDARD ($30/tháng, 15 thread), một tham số sai lặp lại trên nhiều thread cùng lúc tốn thời gian hơn nhiều so với một request đơn lẻ — validate sớm chặn lỗi đó ngay tại nguồn.

Định nghĩa model Pydantic cho từng loại CAPTCHA

Mỗi loại CAPTCHA có tập tham số khác nhau. Các model dưới đây tách riêng từng loại, xác thực độ dài sitekey, định dạng pageurl, và tự dựng đúng payload mà in.php cần.

Ba điểm mỗi model kiểm tra:

1. Độ dài sitekey (min_length=20 cho reCAPTCHA, min_length=10 cho Turnstile).
2. pageurl hợp lệ — kiểu HttpUrl tự từ chối chuỗi thiếu scheme.
3. Response in.php/res.php parse thành model có status, success, token.

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

Viết client Python gọi API CaptchaAI

Class CaptchaAI bọc toàn bộ vòng đời một task: gửi tới in.php, polling res.php mỗi poll_interval giây, rồi parse kết quả qua các model ở trên. Tham số đầu vào đã qua validation nên phần này chỉ còn lo I/O và lỗi API.

Bốn method public:

• solve_recaptcha_v2() / solve_recaptcha_v3()
• solve_turnstile()
• solve_image()
• get_balance() — kiểm tra số dư

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

Lưu ý: poll_interval mặc định 5 giây; giảm xuống 2–3 giây nếu cần token nhanh hơn, đổi lại gọi res.php nhiều lần hơn.

Ví dụ chạy thử: request hợp lệ và các lỗi bị chặn sớm

Ba tình huống dưới đây minh hoạ đúng giá trị của việc validate sớm:

1. Request hợp lệ — validation pass, gọi API bình thường.
2. Sitekey rỗng — ValidationError chặn trước khi có request nào ra mạng.
3. Lỗi API thật (CaptchaAIError) — tham số đã hợp lệ nhưng CaptchaAI xử lý thất bại.

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

Cài các gói cần thiết trước khi chạy:

pip install pydantic requests

Xử lý lỗi thường gặp khi validate

Trước khi báo lỗi, kiểm tra nhanh:

• Đã cài pydantic>=2.0 chưa (pip show pydantic).
• sitekey/pageurl đúng định dạng site đang test.
• Số dư tài khoản còn đủ không.

Câu hỏi thường gặp về Pydantic và CaptchaAI

Có bắt buộc dùng Pydantic v2 không?

Nên dùng — field_validator và model_validate trong bài này là API của v2. Pydantic v1 thiếu field_validator (dùng validator thay thế); cài pip install 'pydantic>=2.0' để tránh lỗi import.

Model có tự retry khi gặp ERROR_ZERO_BALANCE không?

Không. Model chỉ validate tham số đầu vào — retry, backoff hay quản lý thread song song là logic ở tầng client, tách biệt với tầng validation.

Có tái dùng cho GeeTest v3 hoặc BLS CAPTCHA không?

Có. Tạo thêm BaseModel cho từng loại (GeeTest v3 dùng method: "geetest", BLS dùng method: "bls") theo mẫu RecaptchaV2Request ở trên, rồi thêm hàm solve_* tương ứng vào client.

Validate có ảnh hưởng gì khi chạy nhiều thread song song không?

Không đáng kể — validation là tác vụ CPU thuần, chỉ vài micro giây và không giữ I/O, kể cả trên các plan nhiều thread như STANDARD (15 thread) hay ADVANCE ($90/tháng, 50 thread). Thời gian tốn thực sự vẫn là round-trip HTTP tới in.php/res.php.

Validate bằng Pydantic có tính vào chi phí gọi CaptchaAI không?

Không. Validation chạy hoàn toàn ở phía client, trước khi có request nào gửi tới in.php — CaptchaAI chỉ tính phí trên task thực sự được submit. Một sitekey sai bị ValidationError chặn lại không tốn gì cả, kể cả trên plan trả theo thread như STANDARD.

Bài viết liên quan

• Hướng dẫn đầy đủ dùng CaptchaAI với Playwright cho Python
• Xây dựng pipeline client-CAPTCHA với CaptchaAI
• Xác thực bảo mật webhook/callback của CaptchaAI

Bước tiếp theo

Ghép các model Pydantic ở trên vào project của bạn — lấy API key tại CaptchaAI và chạy thử ví dụ solve_recaptcha_v2 với sitekey thật của trang bạn đang kiểm thử.

• Copy models.py, client.py vào project, pip install pydantic requests.
• Test với sitekey/pageurl thật trước khi đưa vào pipeline CI.

Hướng dẫn liên quan:

• Xây dựng Python wrapper cho API CaptchaAI
• Tham khảo mã lỗi CaptchaAI
• So sánh JSON API và Form API của CaptchaAI