التعامل مع اختبار CAPTCHA في تطبيقات Django باستخدام CaptchaAI

يتعامل أي تطبيق Django مع CAPTCHA من اتجاهين متعاكسين، والخلط بينهما هو أكثر ما يربك المطوّرين:

• اتجاه داخل (Inbound): رموز CAPTCHA تصل إلى نماذجك أنت، فتتحقق منها على الخادم لتصدّ الروبوتات عن التسجيل والتعليقات ونماذج التواصل.
• اتجاه خارج (Outbound): أتمتتك تصطدم بـ CAPTCHA على مواقع طرف ثالث، فتحتاج إلى حلّه برمجيًا حتى يكتمل جمع البيانات أو الاختبار.

الاتجاه الأول تتكفّل به أدوات الحماية نفسها (Turnstile وreCAPTCHA) عبر التحقق الخادمي، أمّا الاتجاه الثاني فهو ما يتولّاه CaptchaAI. تخيّل فريقًا في الرياض يبني بوابة مقارنة أسعار: نماذج التسجيل لديه محميّة بـ Turnstile — اتجاه داخل — بينما يحتاج زاحفه إلى قراءة صفحات موردين محميّة بـ CAPTCHA أيضًا — اتجاه خارج. يغطّي هذا الدليل المسارين معًا داخل مشروع Django واحد.

الاتجاه الداخل: التحقق من CAPTCHA على نماذج Django

المبدأ: حين تضيف Turnstile أو reCAPTCHA إلى نموذج Django، لا يكفي عرض الأداة في الواجهة؛ يجب أن يتحقق الخادم من التوكن قبل قبول الإرسال، وإلا صار الحقل مجرّد زينة يتجاوزها أي روبوت. يجري هذا التحقق مباشرة مع خوادم مزوّد الحماية، لا عبر CaptchaAI.

دمج Turnstile في نموذج Django

# forms.py
from django import forms

class ContactForm(forms.Form):
    name = forms.CharField(max_length=100)
    email = forms.EmailField()
    message = forms.CharField(widget=forms.Textarea)
    cf_turnstile_response = forms.CharField(
        widget=forms.HiddenInput(),
        required=True,
    )

# views.py
import requests
from django.conf import settings
from django.shortcuts import render, redirect
from .forms import ContactForm

def contact_view(request):
    if request.method == "POST":
        form = ContactForm(request.POST)
        if form.is_valid():
            # Verify Turnstile token with Cloudflare
            token = form.cleaned_data["cf_turnstile_response"]
            verification = requests.post(
                "https://challenges.cloudflare.com/turnstile/v0/siteverify",
                data={
                    "secret": settings.TURNSTILE_SECRET_KEY,
                    "response": token,
                    "remoteip": request.META.get("REMOTE_ADDR"),
                },
            ).json()

            if verification.get("success"):
                # Process the form
                return redirect("success")
            else:
                form.add_error(None, "CAPTCHA verification failed")
    else:
        form = ContactForm()

    return render(request, "contact.html", {
        "form": form,
        "turnstile_sitekey": settings.TURNSTILE_SITE_KEY,
    })

<!-- templates/contact.html -->
<form method="post">
    {% csrf_token %}
    {{ form.as_p }}
    <div class="cf-turnstile" data-sitekey="{{ turnstile_sitekey }}"></div>
    <button type="submit">Send</button>
</form>
<script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async defer></script>

ملاحظة أمان: بعد أن يعيد siteverify النتيجة success، عالِج النموذج. خزّن TURNSTILE_SECRET_KEY في متغيّرات البيئة فقط، ولا تضعه في الكود أو في التحكّم بالإصدارات.

الاتجاه الخارج: حل CAPTCHA على المواقع الخارجية عبر CaptchaAI

هنا يبدأ دور CaptchaAI: حين يحتاج تطبيق Django إلى موقع طرف ثالث محميّ بـ CAPTCHA — لجمع بيانات أو اختبار أو أتمتة — يرسل الطلب إلى CaptchaAI فيعيد توكنًا يمرّره إلى الموقع الهدف.

يحلّ CaptchaAI أنواع CAPTCHA الأكثر شيوعًا في هذا السياق:

• مدعوم رسميًا: reCAPTCHA v2 وv3 وEnterprise وInvisible، وCloudflare Turnstile وChallenge، وGeeTest v3، وصور OCR والنصوص، والشبكات الصورية، وBLS.
• في مرحلة beta: CaptchaFox وFriendly Captcha وLemin.
• غير مدعوم: لا يحلّ CaptchaAI حاليًا hCaptcha ولا FunCaptcha، وGeeTest v4 غير متاح بعد.

تحقّق دائمًا من نوع الحماية على الموقع الهدف قبل بناء تدفّقك.

يعتمد CaptchaAI تسعيرًا قائمًا على الـ threads لا على عدد عمليات الحل، وهو ما يناسب تطبيقات Django المتزامنة: تبدأ باقة BASIC من $15 شهريًا مع 5 threads، وتصل VIP-3 إلى $7,500 شهريًا مع 5,000 threads، وتتضمّن كل باقة عددًا غير محدود من عمليات الحل لكل thread. عمليًا، عدد الـ threads هو سقف عمليات الحل المتوازية لديك.

فئة خدمة CaptchaAI: تغليف تدفّق الإرسال والاستطلاع

# services/captcha_solver.py
import time
import requests
from django.conf import settings


class CaptchaSolverService:
    """Django service for solving CAPTCHAs via CaptchaAI."""

    API_BASE = "https://ocr.captchaai.com"

    def __init__(self):
        self.api_key = settings.CAPTCHAAI_API_KEY

    def solve_recaptcha_v2(self, sitekey, page_url, invisible=False):
        """Solve reCAPTCHA v2."""
        params = {
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
            "json": 1,
        }
        if invisible:
            params["invisible"] = 1
        return self._submit_and_poll(params)

    def solve_turnstile(self, sitekey, page_url, action=None):
        """Solve Cloudflare Turnstile."""
        params = {
            "key": self.api_key,
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": page_url,
            "json": 1,
        }
        if action:
            params["action"] = action
        return self._submit_and_poll(params)

    def solve_image(self, image_base64):
        """Solve image/text CAPTCHA."""
        return self._submit_and_poll({
            "key": self.api_key,
            "method": "base64",
            "body": image_base64,
            "json": 1,
        })

    def get_balance(self):
        """Check API balance."""
        response = requests.get(f"{self.API_BASE}/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        }, timeout=30)
        return float(response.json().get("request", 0))

    def _submit_and_poll(self, params, timeout=120):
        """Submit task and poll for result."""
        # Submit
        response = requests.post(f"{self.API_BASE}/in.php", data=params, timeout=30)
        response.raise_for_status()
        data = response.json()

        if data.get("status") != 1:
            raise CaptchaSolveError(f"Submit failed: {data.get('request')}")

        task_id = data["request"]

        # Poll
        start = time.time()
        while time.time() - start < timeout:
            time.sleep(5)
            result = requests.get(f"{self.API_BASE}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1,
            }, timeout=30).json()

            if result.get("status") == 1:
                return result["request"]
            if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                raise CaptchaSolveError("CAPTCHA unsolvable")

        raise CaptchaSolveError("Solve timed out")


class CaptchaSolveError(Exception):
    pass

تتبع الفئة تدفّقًا من خطوتين:

1. ترسل المهمة إلى نقطة النهاية in.php فتحصل على معرّف.
2. تستطلع النتيجة من res.php كل 5 ثوانٍ حتى تعود status == 1.

تركّز الدالة _submit_and_poll هذا المنطق في مكان واحد، فتشترك فيه جميع أنواع CAPTCHA، ويصبح CaptchaSolveError نقطة تحكّم موحّدة لالتقاط الأخطاء.

إعدادات Django

# settings.py
CAPTCHAAI_API_KEY = "YOUR_API_KEY"
TURNSTILE_SITE_KEY = "0x4AAAAAAAC3DHQhMMQ_Rxrg"
TURNSTILE_SECRET_KEY = "0x4AAAAAAAC3DHQhYYY_secret"

تشغيل الحل داخل تطبيق Django

تُستدعى الخدمة نفسها بحسب سياق التشغيل: طريقة عرض، أمر إدارة، عرض غير متزامن، أو مهمة Celery.

طريقة عرض لجمع بيانات خارجية

تستقبل طريقة العرض التالية رابط الهدف، تحلّ الـ CAPTCHA عبر الخدمة، ثم تمرّر التوكن الناتج إلى الموقع المحمي لإكمال الطلب:

# views.py
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from .services.captcha_solver import CaptchaSolverService, CaptchaSolveError

@require_POST
def scrape_external_data(request):
    """Solve CAPTCHA and fetch data from external CAPTCHA-protected site."""
    url = request.POST.get("target_url")
    if not url:
        return JsonResponse({"error": "target_url required"}, status=400)

    solver = CaptchaSolverService()

    try:
        # Solve the CAPTCHA
        token = solver.solve_turnstile(
            sitekey="0x4AAAAAAAC3DHQhMMQ_Rxrg",
            page_url=url,
        )

        # Use token to access the protected resource
        import requests as http_requests
        response = http_requests.post(url, data={
            "cf-turnstile-response": token,
        }, timeout=30)

        return JsonResponse({
            "status": "success",
            "data": response.text[:1000],
        })

    except CaptchaSolveError as e:
        return JsonResponse({"error": str(e)}, status=500)

أمر إدارة Django للتشغيل من سطر الأوامر

حين تريد حلّ CAPTCHA يدويًا أو ضمن سكربت تشغيلي، غلّف الخدمة في أمر إدارة يقرأ نوع الحماية والـ sitekey والرابط من الوسائط:

# management/commands/solve_captcha.py
from django.core.management.base import BaseCommand
from myapp.services.captcha_solver import CaptchaSolverService


class Command(BaseCommand):
    help = "Solve a CAPTCHA and print the token"

    def add_arguments(self, parser):
        parser.add_argument("--type", choices=["recaptcha", "turnstile"], required=True)
        parser.add_argument("--sitekey", required=True)
        parser.add_argument("--url", required=True)

    def handle(self, *args, **options):
        solver = CaptchaSolverService()

        self.stdout.write(f"Solving {options['type']} for {options['url']}...")

        if options["type"] == "recaptcha":
            token = solver.solve_recaptcha_v2(options["sitekey"], options["url"])
        else:
            token = solver.solve_turnstile(options["sitekey"], options["url"])

        self.stdout.write(self.style.SUCCESS(f"Token: {token[:50]}..."))

        # Check balance
        balance = solver.get_balance()
        self.stdout.write(f"Remaining balance: ${balance:.2f}")

مثال على التشغيل من الطرفية:

python manage.py solve_captcha --type turnstile --sitekey 0x4AAA... --url https://example.com

طرق العرض غير المتزامنة

منذ الإصدار 4.1، يدعم Django طرق العرض غير المتزامنة، وهي مناسبة حين تخدم عدّة طلبات ويب في آنٍ واحد دون حجب العامل.

تنبيه: لا تستخدم requests المتزامنة داخل عرض غير متزامن؛ استبدلها بـ aiohttp كما في المثال، وإلا جمّدت حلقة الأحداث.

# views.py (async)
import aiohttp
import asyncio
from django.http import JsonResponse

CAPTCHAAI_API_KEY = "YOUR_API_KEY"

async def solve_captcha_async(request):
    """Async view for solving CAPTCHAs."""
    sitekey = request.GET.get("sitekey")
    page_url = request.GET.get("url")

    if not sitekey or not page_url:
        return JsonResponse({"error": "sitekey and url required"}, status=400)

    async with aiohttp.ClientSession() as session:
        # Submit
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": CAPTCHAAI_API_KEY,
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": page_url,
            "json": 1,
        }) as resp:
            data = await resp.json()

        if data.get("status") != 1:
            return JsonResponse({"error": data.get("request")}, status=500)

        task_id = data["request"]

        # Poll
        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": CAPTCHAAI_API_KEY,
                "action": "get",
                "id": task_id,
                "json": 1,
            }) as resp:
                result = await resp.json()

            if result.get("status") == 1:
                return JsonResponse({"token": result["request"]})

    return JsonResponse({"error": "timeout"}, status=504)

المعالجة في الخلفية عبر Celery

للأحمال الطويلة أو الدفعات الكبيرة، انقل الحل إلى الخلفية عبر Celery حتى لا ينتظر المستخدم النتيجة داخل دورة الطلب.

المتطلبات: عامل Celery منفصل، ووسيط رسائل مثل Redis، وضبط max_retries وزمن التأخير ليتولّى العامل التنفيذ وإعادة المحاولة.

# tasks.py
from celery import shared_task
from .services.captcha_solver import CaptchaSolverService, CaptchaSolveError

@shared_task(bind=True, max_retries=2, default_retry_delay=10)
def solve_captcha_task(self, captcha_type, sitekey, page_url):
    """Background CAPTCHA solving with Celery."""
    solver = CaptchaSolverService()

    try:
        if captcha_type == "recaptcha_v2":
            token = solver.solve_recaptcha_v2(sitekey, page_url)
        elif captcha_type == "turnstile":
            token = solver.solve_turnstile(sitekey, page_url)
        else:
            raise ValueError(f"Unknown type: {captcha_type}")

        return {"success": True, "token": token}

    except CaptchaSolveError as e:
        self.retry(exc=e)

# Usage in views
from .tasks import solve_captcha_task

def start_solve(request):
    result = solve_captcha_task.delay("turnstile", "0x4AAA...", "https://example.com")
    return JsonResponse({"task_id": result.id})

def check_solve(request, task_id):
    from celery.result import AsyncResult
    result = AsyncResult(task_id)
    if result.ready():
        return JsonResponse(result.get())
    return JsonResponse({"status": "pending"})

متى تختار كل نمط: متزامن، غير متزامن، أو Celery

الأنماط الثلاثة تستدعي الخدمة نفسها، والفرق في مكان تنفيذ الحل وتوقيته. اختر بحسب سياق الاستدعاء:

قاعدة عملية: إن كان الطلب قادمًا من مستخدم ينتظر استجابة، فلا تحلّ الـ CAPTCHA بشكل متزامن داخل الطلب؛ ادفعه إلى Celery وأعِد للمستخدم معرّف مهمة يستعلم عنه لاحقًا.

الأسئلة المتداولة

هل يمكن حلّ hCaptcha أو FunCaptcha بهذه الطريقة؟

لا. لا يدعم CaptchaAI هذين النوعين حتى الآن، ولا GeeTest v4. يقتصر الدعم على reCAPTCHA v2/v3، وCloudflare Turnstile وChallenge، وGeeTest v3، وصور OCR والشبكات وBLS، مع CaptchaFox وFriendly Captcha وLemin في مرحلة beta. تحقّق من نوع الحماية على الموقع الهدف قبل كتابة الكود.

كم يكلّف الحل على نطاق واسع، وكيف تُحسب الـ threads؟

الفوترة قائمة على الـ threads لا على عدد عمليات الحل:

• كل thread يمثّل عملية حل واحدة متزامنة، وعمليات الحل غير محدودة داخل الباقة.
• تبدأ BASIC من $15 شهريًا مع 5 threads، وترتفع الباقات كلما زاد التزامن المطلوب.
• اربط عدد الـ threads بعدد عمّال Celery حتى لا تتجاوز حصّتك.

ما الفرق بين "التحقق" و"الحل" في هذا الدليل؟

التفريق بينهما جوهري في التصميم:

• التحقق (الاتجاه الداخل): يجري على الخادم مع مزوّد الحماية لقبول إرسال نموذجك، ولا يمرّ عبر CaptchaAI.
• الحل (الاتجاه الخارج): استدعاء CaptchaAI للحصول على توكن يفتح موقعًا خارجيًا محميًا.

أين أضع مفتاح CAPTCHAAI_API_KEY بأمان؟

في متغيّرات البيئة، أو عبر حزمة django-environ، ولا تضعه أبدًا في الكود أو في التحكّم بالإصدارات. الأمر نفسه ينطبق على TURNSTILE_SECRET_KEY.

هل يتعامل نفس الكود مع reCAPTCHA وTurnstile معًا؟

نعم. فئة CaptchaSolverService عديمة الحالة (stateless)، وتوفّر دوالّ منفصلة لكل نوع فوق التدفّق نفسه. أنشئ نسخة جديدة لكل طلب، أو اعتمد على أنماط حقن التبعية في Django.

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

ملخص

باختصار، يفصل هذا الدليل اتجاهَي CAPTCHA في Django: التحقق الخادمي على نماذجك، والحل الخارجي عبر CaptchaAI بفئة خدمة واحدة تغلّف تدفّق الإرسال والاستطلاع. اعتمد الحل المتزامن في أوامر الإدارة والسكربتات، وطرق العرض غير المتزامنة في Django 4.1+، ومهام Celery للأحمال الطويلة — وتبقى الخدمة نفسها صالحة لـ reCAPTCHA وTurnstile وصور CAPTCHA دون تكرار الكود.

مقالات ذات صلة

• الفرق بين GeeTest وCloudflare Turnstile
• معالجة خطأ 403 في Turnstile بعد إصلاح التوكن
• أوضاع عمل أداة Cloudflare Turnstile

أدلة ذات صلة

• حل CAPTCHA في اختبارات التكامل المستمر
• بناء ناقل أحداث CAPTCHA باستخدام Node.js وCaptchaAI