Xử lý CAPTCHA trong ứng dụng Django bằng CaptchaAI

Các ứng dụng Django thường xuyên cần xử lý CAPTCHA trong hai trường hợp: xác minh CAPTCHA trên biểu mẫu của riêng bạn (bảo vệ khỏi bot) và giải CAPTCHA trên các trang web bên ngoài (thu thập dữ liệu, kiểm tra, tự động hóa). Hướng dẫn này bao gồm cả hai mẫu sử dụng CaptchaAI.

Tình huống 1: Xác minh CAPTCHA trên biểu mẫu Django của bạn

Khi thêm Turnstile hoặc reCAPTCHA vào biểu mẫu Django, bạn cần xác minh mã thông báo phía máy chủ.

Thêm Turnstile vào biểu mẫu 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>

Tình huống 2: Giải CAPTCHA trên các trang web bên ngoài

Đây là lúc CaptchaAI xuất hiện - khi ứng dụng Django của bạn cần tương tác với các trang web bên ngoài được bảo vệ bằng CAPTCHA.

Lớp dịch vụ 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

Cài đặt Django

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

Sử dụng dịch vụ trong chế độ xem

Xem để thu thập dữ liệu bên ngoài

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

Lệnh quản lý Django

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

Cách sử dụng:

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

Không đồng bộ Django với CaptchaAI

Django 4.1+ hỗ trợ chế độ xem không đồng bộ:

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

Tích hợp cần tây để giải quyết nền

Để giải CAPTCHA trong thời gian dài, hãy sử dụng Celery:

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

Khắc phục sự cố

Triệu chứng	nguyên nhân	sửa chữa
`CaptchaSolveError` đang được sản xuất	Khóa API không có trong cài đặt	Thêm `CAPTCHAAI_API_KEY` vào cài đặt Django
Nhiệm vụ cần tây thử lại không ngừng	CAPTCHA không thể giải được hoặc khóa trang web sai	Đặt `max_retries` và xác thực đầu vào
Chế độ xem không đồng bộ bị treo	Đồng bộ hóa mã ở chế độ xem không đồng bộ	Sử dụng `aiohttp` thay vì `requests`
Mã thông báo đã hết hạn trước khi gửi biểu mẫu	Giải quyết mất quá nhiều thời gian	Giải quyết kịp thời, không giải quyết trước
Lỗi nhập lệnh quản lý	Dịch vụ không có trong INSTALLED_APPS	Kiểm tra đăng ký ứng dụng

Câu hỏi thường gặp

Việc giải CAPTCHA nên đồng bộ hay không đồng bộ?

Sử dụng Celery cho chế độ xem trên web để người dùng không phải đợi hơn 15 giây. Sử dụng tính năng giải đồng bộ trong các lệnh quản lý và tập lệnh nền.

Làm cách nào để lưu trữ khóa API một cách an toàn?

Sử dụng các biến môi trường hoặc gói django-environ của Django. Không bao giờ cam kết khóa API để kiểm soát phiên bản.

Tôi có thể lưu vào bộ nhớ đệm các mã thông báo đã giải quyết không?

Mã thông báo reCAPTCHA hết hạn sau 120 giây và mã thông báo Turnstile hết hạn sau 300 giây. Bộ nhớ đệm không thực tế - hãy giải quyết ngay trước khi sử dụng.

Tôi nên tạo một phiên bản dịch vụ hay sử dụng một phiên bản dịch vụ?

Lớp CaptchaSolverService không có trạng thái. Tạo một phiên bản mới cho mỗi yêu cầu hoặc sử dụng các mẫu chèn phụ thuộc của Django.

Tóm tắt

Ứng dụng Django tích hợp vớiCaptchaAIthông qua một lớp dịch vụ bao bọc luồng submit/poll. Sử dụng tính năng giải quyết đồng bộ trong các lệnh quản lý, giải quyết không đồng bộ trong chế độ xem không đồng bộ Django 4.1+ và các tác vụ Celery để xử lý nền. Dịch vụ tương tự xử lý reCAPTCHA, Turnstile và CAPTCHA hình ảnh.