Penanganan CAPTCHA pada Aplikasi Django dengan CaptchaAI

Ada dua kebutuhan CAPTCHA di Django yang sering tertukar, padahal solusinya berbeda. Yang pertama adalah memverifikasi token CAPTCHA yang masuk dari formulir Anda sendiri — di sini Cloudflare-lah yang memvalidasi, bukan CaptchaAI. Yang kedua adalah menyelesaikan CAPTCHA di situs pihak ketiga saat aplikasi Anda mengambil data atau menjalankan pengujian otomatis — di sinilah CaptchaAI berperan. Panduan ini memisahkan keduanya dengan tegas, lalu membungkus pola kedua ke dalam satu service class yang bisa dipakai ulang dari view, management command, async view, sampai task Celery.

Untuk memilih pola yang tepat, kenali dulu posisi Anda:

• Pemilik formulir. Anda memasang widget Turnstile atau reCAPTCHA di halaman sendiri dan hanya perlu memvalidasi token yang masuk. Validatornya adalah siteverify Cloudflare, bukan CaptchaAI.
• Klien situs pihak ketiga. Aplikasi Anda mengambil data atau menguji halaman yang dilindungi CAPTCHA milik sumber lain. Di sinilah CaptchaAI menyelesaikan tantangan dan mengembalikan token siap pakai.

Skenario 1: memverifikasi Turnstile pada formulir Django Anda

Ketika Anda memasang Cloudflare Turnstile atau reCAPTCHA di formulir Django, token yang dikirim browser harus divalidasi di sisi server sebelum data diproses. Validasi ini memanggil endpoint siteverify milik Cloudflare — bukan CaptchaAI — dan cukup satu request HTTP biasa.

Menambahkan field Turnstile ke Django form

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

Field tersembunyi cf-turnstile-response diisi otomatis oleh widget, lalu view membacanya dari cleaned_data. Selama respons success bernilai false, formulir ditolak — pola yang sama berlaku untuk reCAPTCHA, hanya endpoint dan nama field-nya yang berbeda.

Skenario 2: menyelesaikan CAPTCHA di situs eksternal

Kebutuhan kedua muncul saat aplikasi Django Anda menjadi klien, bukan pemilik formulir: misalnya sebuah agen scraping atau tim monitoring harga yang harus melewati halaman terlindungi CAPTCHA milik sumber data resmi. CaptchaAI menerima detail tantangan lewat in.php, memprosesnya dengan thread, lalu mengembalikan token siap pakai yang Anda ambil melalui polling ke res.php.

Membungkus alur submit/poll ke dalam service class

Alih-alih menyebar panggilan HTTP di banyak view, satukan seluruh logika CaptchaAI dalam satu kelas. Pola submit → simpan task ID → polling → pakai token cukup ditulis sekali di sini, lalu setiap pemanggil hanya memanggil satu method.

Empat langkah yang dibungkus kelas ini selalu berurutan sama:

1. Kirim detail tantangan ke in.php.
2. Simpan task ID yang dikembalikan pada respons.
3. Polling res.php sampai status bernilai 1.
4. Pakai token hasil solve sebelum masa berlakunya habis.

# 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

Satu kelas ini sudah menangani reCAPTCHA v2, Cloudflare Turnstile, dan image/OCR — tiga tipe yang termasuk didukung penuh oleh CaptchaAI. Perlu diingat bahwa hCaptcha dan FunCaptcha belum didukung, jadi jangan tambahkan method untuk keduanya.

Menyimpan kredensial di Django settings

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

Memakai service dari berbagai entry point

Satu service class bisa dipanggil dari empat konteks eksekusi. Pilih berdasarkan siapa yang menunggu hasilnya:

• View sinkron untuk pemanggilan internal yang tidak ditunggu pengguna di browser.
• Management command untuk cron dan skrip terminal.
• Async view saat Anda ingin melepaskan event loop selama menunggu solve.
• Task Celery untuk permintaan yang menghadap pengguna.

View untuk pengumpulan data eksternal

Endpoint ini menyelesaikan CAPTCHA lalu memakai token untuk mengakses resource terlindungi. Karena solve bisa memakan puluhan detik, pola sinkron seperti ini cocok untuk pemanggilan internal, bukan untuk request yang ditunggu pengguna di browser.

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

Management command untuk skrip dan cron

Untuk pekerjaan batch atau uji coba dari terminal, service yang sama dibungkus sebagai management command sehingga bisa dipanggil langsung tanpa server HTTP.

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

Cara menjalankannya:

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

Async view di Django 4.1+

Sejak Django 4.1, view asinkron sudah stabil. Ini pas untuk beban solve yang banyak menunggu I/O: alih-alih memblokir worker selama polling, aiohttp melepaskan event loop sehingga request lain tetap terlayani.

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

Aturan pentingnya: jangan mencampur requests yang sinkron di dalam async view — satu panggilan blocking akan mengunci seluruh event loop.

Task Celery untuk penyelesaian latar belakang

Untuk aplikasi yang menghadap pengguna, cara paling nyaman adalah memindahkan solve ke antrean Celery. Request web langsung mengembalikan task_id, worker menyelesaikan CAPTCHA di latar belakang, dan frontend melakukan polling status kapan pun siap.

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

Karena CaptchaAI menagih per thread — bukan per solve — beban Celery yang naik-turun tidak menambah biaya per penyelesaian. Jumlah thread menentukan berapa banyak solve berjalan bersamaan:

• BASIC — $15/bulan, 5 thread; menampung lima solve sekaligus.
• STANDARD — $30/bulan, 15 thread; untuk worker yang mulai sering antre.
• ADVANCE — $90/bulan, 50 thread; paralelisme lebih besar tanpa biaya tambahan per CAPTCHA.

Contoh nyata: worker scraping di ap-southeast-3 (Jakarta)

Bayangkan tim data di Jakarta men-deploy worker Celery di region AWS ap-southeast-3. Latensi ke sumber data mungkin fluktuatif, jadi default_retry_delay=10 dan max_retries=2 memberi ruang percobaan ulang tanpa membuat task berputar tanpa henti. Satu catatan kepatuhan yang relevan di Indonesia: sesuai UU Pelindungan Data Pribadi (UU 27/2022), pastikan Anda hanya mengambil data yang memang berhak Anda proses dan hindari data pribadi tanpa dasar hukum — panduan ini bukan nasihat hukum, tetapi menaruh pengecekan itu di awal pipeline adalah kebiasaan yang sehat.

Pemecahan masalah

Pertanyaan yang sering diajukan

Tipe CAPTCHA apa saja yang bisa diselesaikan service class ini?

Method di service class mencakup reCAPTCHA v2, Cloudflare Turnstile, dan image/OCR — semuanya tipe yang didukung penuh CaptchaAI. hCaptcha dan FunCaptcha belum didukung, jadi jangan menambahkan method untuk keduanya.

Sebaiknya solve dijalankan sinkron atau lewat Celery?

Untuk view yang ditunggu pengguna, pakai Celery agar tidak ada yang menunggu belasan detik di depan layar. Solve sinkron cukup untuk management command, cron, dan skrip latar belakang yang tidak menghadap pengguna.

Apakah biaya bertambah kalau banyak worker Celery jalan bersamaan?

Tidak per solve. CaptchaAI menagih per thread konkuren, jadi biaya bulanan tetap sesuai paket; yang perlu Anda sesuaikan hanyalah jumlah thread. BASIC ($15/bulan) memberi 5 thread, ADVANCE ($90/bulan) memberi 50 thread.

Perlukah token hasil solve disimpan di cache?

Tidak praktis. Token reCAPTCHA kedaluwarsa dalam 120 detik dan token Cloudflare Turnstile dalam 300 detik, jadi selesaikan tepat sebelum dipakai daripada menyimpannya.

Bagaimana menangani ERROR_CAPTCHA_UNSOLVABLE di worker Celery?

Service class melempar CaptchaSolveError, lalu task Celery menangkapnya dan memanggil self.retry dengan default_retry_delay. Periksa dulu sitekey dan page_url sebelum menaikkan max_retries — error yang berulang biasanya menandakan input yang salah, bukan gangguan sesaat.

Ringkasan

Kunci integrasi CAPTCHA di Django adalah memisahkan dua kebutuhan: verifikasi Turnstile pada formulir sendiri cukup memanggil siteverify Cloudflare, sedangkan penyelesaian CAPTCHA situs eksternal dibungkus dalam satu service class CaptchaAI yang menjalankan alur submit/poll. Dari kelas itu Anda bisa memanggilnya secara sinkron di management command, asinkron di async view Django 4.1+, atau di latar belakang lewat task Celery — semuanya berbagi logika reCAPTCHA, Turnstile, dan image yang sama.

Artikel Terkait

• Mengatasi Cloudflare Turnstile 403 setelah token valid
• Memahami mode widget Cloudflare Turnstile
• Penanganan CAPTCHA pada aplikasi Flask