عميل CaptchaAI Python مع التحقق من صحة Pydantic

يؤدي تمرير مفتاح موقع فارغ إلى CaptchaAI API إلى إهدار رحلة ذهابًا وإيابًا - ستحصل على ERROR_WRONG_CAPTCHA_ID بعد انتظار الرد. يكتشف Pydantic هذه الأخطاء قبل إجراء استدعاء HTTP: مسح أخطاء التحقق من الصحة بدلاً من رموز خطأ واجهة برمجة التطبيقات المشفرة.

لماذا Pydantic لعملاء CAPTCHA API

نماذج

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

العميل

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

الاستخدام

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://example.com/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}")

تثبيت التبعيات:

pip install pydantic requests

استكشاف الأخطاء وإصلاحها

الخطوات التالية

• البدء السريع مع CaptchaAI: حلّ أول كابتشا في 5 دقائق
• كيفية حلّ reCAPTCHA v2 عبر الـ API: دليل خطوة بخطوة
• كيفية حل Cloudflare Turnstile باستخدام واجهة API
• كيفية حل GeeTest v3 باستخدام API

أدلة ذات صلة

• CaptchaAI في الإنتاج: دليل إدارة التكوين
• تسعير CaptchaAI: تفصيل قائم على نموذج Threads