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

طبقة تحقّق صغيرة مبنية على Pydantic تفصل بين نوعين مختلفين تماماً من الأخطاء: خطأ في مدخلاتك أنت، وخطأ قادم من الخدمة. الأول — مفتاح موقع فارغ أو رابط صفحة بلا بروتوكول — يجب أن يظهر فوراً وقبل أي اتصال بالشبكة؛ والثاني وحده يستحق أن يستهلك خيط معالجة ورحلة ذهاب وإياب كاملة إلى CaptchaAI API. في هذا الشرح نبني عميل CaptchaAI في Python يتحقق من كل معامل عبر نماذج Pydantic، فيرفض المدخلات الخاطئة برسالة واضحة بدل رمز خطأ غامض مثل ERROR_WRONG_CAPTCHA_ID، ويقرأ استجابة الـ API داخل نماذج مكتوبة بدل التعامل مع قواميس هشّة عرضة لأخطاء KeyError.

ما الذي تضيفه طبقة Pydantic لعميل CAPTCHA؟

الفكرة الأساسية أن تنقل اكتشاف الخطأ من لحظة استجابة الخادم إلى لحظة إنشاء الطلب. الجدول التالي يقارن عميلاً بلا تحقق بآخر مبني على نماذج مكتوبة:

يترجم هذا الفرق إلى ثلاث فوائد ملموسة أثناء العمل:

• اكتشاف الخطأ ينتقل من لحظة رد الخادم إلى لحظة إنشاء الطلب.
• الرسالة تشير إلى الحقل المسؤول بالاسم بدل رمز خطأ شبكي غامض.
• محرّر الأكواد يقدّم إكمالاً تلقائياً وتلميحات نوعية على كل حقل.

لهذا الفصل قيمة عملية مباشرة في نموذج CaptchaAI القائم على خيوط المعالجة: فالفوترة تحسب لكل خيط متزامن لا لكل عملية حل. على خطة BASIC بسعر 15 دولاراً شهرياً وخمسة خيوط معالجة، كل طلب ينطلق بمعاملات خاطئة يشغل خيطاً كان يمكن أن يخدم عملية حل حقيقية حتى يعود الرد بالخطأ. حجب هذه الطلبات قبل انطلاقها يبقي الخيوط متاحة للعمل المفيد، ويحوّل أخطاء الإعداد الصامتة إلى استثناءات مرئية في سجلّاتك.

النماذج: التحقق من المعاملات قبل الإرسال

يبدأ كل شيء من ملف models.py، حيث يمثّل كل نوع من أنواع CAPTCHA نموذج BaseModel مستقلاً يصف حقوله وحدودها وطريقة تحويلها إلى بيانات الطلب عبر الدالة to_params(). أما نماذج الاستجابة SubmitResponse وPollResponse فتعمل بالعكس: تقرأ رد الـ API وتكشف عبر خصائص محسوبة إن كانت النتيجة جاهزة وناجحة، فتسلّم task_id أو الرمز مباشرةً دون قراءة مفاتيح القاموس يدوياً. أبرز قواعد التحقق المدمجة:

1. sitekey لـ reCAPTCHA v2: طول بين 20 و100 حرف، بلا مسافات زائدة في الطرفين.
2. sitekey لـ Cloudflare Turnstile: حد أدنى 10 أحرف فقط لأن مفاتيحه أقصر.
3. pageurl: معرّف بنوع HttpUrl يرفض أي رابط بلا بروتوكول قبل أن يصل إلى الخدمة.
4. base64_image: يقلّم بادئة data: تلقائياً ويشترط 100 حرف على الأقل من محتوى 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")

العميل: منطق الإرسال والاستطلاع حول النماذج

بعد أن تحرس النماذج المدخلات، يبقى دور العميل في client.py تنسيق دورة الحل. لاحظ أن كل دالة حل عامة — مثل solve_recaptcha_v2 — تُنشئ النموذج المناسب أولاً، وبذلك يقع أي ValidationError قبل بلوغ السطر الذي يجري الاتصال الشبكي. الفصل هنا نظيف: النماذج تتحقق، والعميل ينفّذ، وتتوزع مسؤولياته على ثلاثة محاور:

• الإرسال إلى in.php ثم استطلاع النتيجة من res.php كل بضع ثوانٍ حتى تجهز.
• فصل أخطاء الخدمة عن أخطاء التحقق عبر CaptchaAIError، فتعرف مصدر أي استثناء تلتقطه.
• احترام مهلة انتهاء صريحة في _poll تفصل حالة «لم تجهز بعد» عن الفشل الفعلي، مع get_balance لقراءة الرصيد قبل الدفعات الكبيرة.

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

مثال عملي: طلب صحيح وأخطاء ملتقطة مبكراً

لتقريب الصورة، تخيّل فريقاً في القاهرة يبني أداة لمراقبة أسعار المنافسين على موقع يحمي نماذجه بـ reCAPTCHA v2 وCloudflare Turnstile. مع عشرات عمليات التشغيل يومياً، أي مفتاح موقع منسوخ بشكل ناقص من لوحة التحكم يعني رحلة مهدورة إلى الخدمة وخيط معالجة مشغولاً بلا فائدة؛ ومع نماذج Pydantic يتحوّل هذا إلى سطر ValidationError واضح في سجلّات الفريق يذكر اسم الحقل. المثال التالي يجمع ثلاث حالات في ملف واحد، والاختلاف في نوع الاستثناء هو ما يجعل التشخيص سريعاً:

1. طلب صحيح يمرّ عبر التحقق ويصل إلى الـ API فيعيد رمزاً ووقت حل.
2. مفتاح موقع فارغ يُرفض فوراً بـ ValidationError دون أي اتصال بالشبكة.
3. خطأ قادم من الخدمة يُلتقط عبر CaptchaAIError بعد إرسال الطلب.

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

قبل التشغيل، ثبّت الاعتماديتين المطلوبتين — احرص على استخدام Pydantic v2 تحديداً لأن صيغة field_validator وmodel_validate تخصّه:

pip install pydantic requests

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

تندرج أغلب المشكلات تحت أحد سببين: مدخل رفضه التحقق محلياً، أو خطأ عاد من الخدمة. الجدول التالي يربط كل عرض بسببه المحتمل وحله:

الأسئلة الشائعة

ما الفرق بين ValidationError وCaptchaAIError؟

الفصل بين النوعين يخبرك فوراً أين تبحث: في الكود أم في الحساب أو الشبكة.

• ValidationError: خطأ في مدخلاتك تكتشفه نماذج Pydantic محلياً قبل أي اتصال — مفتاح موقع فارغ أو رابط بلا بروتوكول.
• CaptchaAIError: خطأ يعود من الخدمة نفسها بعد إرسال الطلب، ويحمل الرمز الأصلي مثل ERROR_ZERO_BALANCE.

أين أضع مفتاح الـ API بأمان بدل كتابته في الكود؟

اتبع ثلاث خطوات بسيطة لإبقاء المفتاح خارج الشيفرة:

1. اقرأ المفتاح من متغير بيئة عبر os.environ، أو من مدير أسرار في بيئة الإنتاج.
2. مرّره إلى CaptchaAI(api_key=...) عند إنشاء العميل.
3. اعتمد على المُنشئ الذي يرفض أي مفتاح أقصر من 10 أحرف، فيلتقط نسيان ضبط المتغير قبل انطلاق الطلبات.

هل يجب أن أستخدم Pydantic v2 تحديداً؟

نعم. الكود يعتمد على صيغة الإصدار الثاني في field_validator وmodel_validate، وهي غير متوافقة مع الإصدار الأول. إن ظهرت أخطاء استيراد، ثبّت pydantic>=2.0 وتأكد ألا يوجد إصدار قديم في البيئة نفسها.

ما أنواع CAPTCHA التي يغطيها هذا العميل؟

النماذج في المثال تغطي جزءاً مما تدعمه الخدمة، والباقي يُضاف بالنمط نفسه:

• في المثال مباشرةً: reCAPTCHA v2 وreCAPTCHA v3 وCloudflare Turnstile وكابتشا الصور/OCR.
• تدعمها CaptchaAI أيضاً: Cloudflare Challenge وGeeTest v3 وكابتشا الشبكة وBLS.
• بدعم تجريبي (beta): CaptchaFox وFriendly Captcha وLemin.
• لإضافة نوع جديد: أنشئ نموذج BaseModel بحقوله ودالة to_params() ثم دالة حل تستدعي _solve.

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

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

أدلة ذات صلة

• تشغيل CaptchaAI في الإنتاج وإدارة التكوين
• تسعير CaptchaAI القائم على خيوط المعالجة