Обработка CAPTCHA в приложениях Django с помощью CaptchaAI

У Django-разработчика CAPTCHA встречается с двух сторон: нужно проверить токен на своей форме (Turnstile или reCAPTCHA уже стоят и отключать их нельзя) — и отдельно решить чужую CAPTCHA, когда бэкенд сам обращается к внешнему сайту за данными.

Коротко, где что применяется:

• своя форма — только проверка токена на сервере, без обращения к сторонним сервисам;
• чужой сайт — нужен внешний решатель вроде CaptchaAI: submit задачи → опрос результата → токен.

Первое закрывается парой строк на сервере, второе требует внешнего сервиса. Ниже — рабочий код для обоих случаев, от sync-view до Celery.

Проверка CAPTCHA на собственных формах Django

Если на форме стоит Turnstile или reCAPTCHA, виджет отдаёт токен, а проверять его подлинность нужно строго на сервере — доверять клиенту здесь нельзя.

Что понадобится до того, как писать код:

• TURNSTILE_SITE_KEY и TURNSTILE_SECRET_KEY из панели Cloudflare;
• домен формы, добавленный в настройки виджета Turnstile;
• HTTPS на проде — без него виджет и серверная проверка теряют смысл.

Как подключить Cloudflare 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>

В шаблоне важны две вещи: data-sitekey берётся из settings.py, а не хардкодится в HTML, и скрипт api.js подключается один раз на странице.

Решение CAPTCHA на внешних сайтах через CaptchaAI

Другая задача — когда парсер на Django должен пройти CAPTCHA на чужом сайте. Здесь подключается CaptchaAI: приложение отправляет параметры задачи на in.php, опрашивает res.php и получает токен.

Цикл одинаковый для любого типа CAPTCHA:

1. отправить параметры задачи на in.php;
2. опрашивать res.php, пока статус не станет 1;
3. получить токен и использовать его в запросе к целевому сайту.

Сервис-класс CaptchaAI для Django

# 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

Настройки Django (settings.py)

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

Ключи не должны попадать в систему контроля версий — держите их в переменных окружения и подтягивайте через django-environ или аналог, а не хардкодьте прямо в settings.py, как в примере выше для краткости.

Как использовать сервис в представлениях (views)

Сервис-класс не завязан на конкретный view — его можно вызвать и из обычной синхронной вьюхи, и из management-команды, и из Celery-задачи.

View для сбора данных с внешнего сайта

Прежде чем собирать данные со стороннего ресурса, стоит решить, какие поля вы вправе забирать: для читателей из РФ — короткая сверка с 152-ФЗ «О персональных данных», для трансграничных проектов — то же в духе GDPR. Это не юридическая консультация, а инженерная гигиена: собирайте только то, что разрешено, и не храните лишнее в базе дольше, чем нужно для задачи.

# 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-команда 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}")

Запуск:

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

Такую команду удобно гонять в CI перед деплоем — быстрая проверка, что sitekey ещё валиден и API-ключ не истёк.

Асинхронные представления Django и CaptchaAI

Начиная с Django 4.1 доступны асинхронные вьюхи, и для CaptchaAI это удобно: пока идёт опрос res.php, event loop не простаивает.

Если бэкенд стоит в европейском регионе, а часть трафика приходит с нестабильных мобильных сетей ближе к Центральной Азии, таймаут 120 секунд в цикле опроса — консервативная стартовая точка. По логам быстро станет видно, нужно ли его увеличивать под конкретный поток трафика.

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

Фоновое решение CAPTCHA через Celery

Опрос res.php может занять до пары минут — держать HTTP-запрос открытым всё это время неудобно, поэтому решение CAPTCHA логично выносить в фоновую задачу 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)

Сколько задач solve_captcha_task идёт параллельно, зависит не от Celery, а от тарифа CaptchaAI: тарификация — по числу одновременных потоков, а не по числу решений.

Если воркер регулярно упирается в лимит потоков раньше, чем в лимит CPU/памяти, это сигнал перейти на тариф выше, а не наращивать concurrency Celery.

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

Типичные проблемы и их решение

Большинство проблем интеграции сводятся к пяти сценариям — вот они с быстрым диагнозом:

Перед деплоем стоит быстро пройтись по списку:

• ключи (CAPTCHAAI_API_KEY, TURNSTILE_SECRET_KEY) заданы через переменные окружения, а не захардкожены;
• max_retries в Celery-задаче не бесконечен;
• таймаут _submit_and_poll покрывает реальное время решения нужного типа CAPTCHA, а не только happy path.

Частые вопросы

Как решить CAPTCHA прямо в Django-миддлваре, а не в отдельной view?

Технически можно — из process_view, но это блокирует весь запрос до ответа CaptchaAI.

Практичнее оставить решение в самой view или в Celery, а миддлварь использовать только для лёгких проверок вроде наличия токена в запросе.

Стоит ли решать CAPTCHA синхронно во view или выносить в Celery?

Для веб-запросов — Celery: пользователь не должен ждать 15+ секунд ответа сервера.

Синхронный вызов оправдан только в management-командах и фоновых скриптах, где никто не ждёт HTTP-ответа — там дополнительная очередь Celery только усложнит отладку.

Можно ли закэшировать уже решённый токен и использовать его повторно?

Нет. Токен reCAPTCHA живёт 120 секунд, токен Cloudflare Turnstile — 300 секунд, и оба привязаны к конкретному запросу.

Решайте CAPTCHA непосредственно перед использованием — кэш здесь не сработает, а попытка переиспользовать токен вернёт ошибку проверки на стороне защищённого ресурса.

Что делать, если задача Celery с решением CAPTCHA постоянно уходит в retry?

Чаще всего — устаревший или неверный sitekey, либо неразрешимая CAPTCHA.

Ограничьте max_retries (в примере — 2), логируйте request из ответа res.php и проверьте sitekey ручным запросом до того, как включать задачу в продовый пайплайн.

Нужен ли отдельный Celery-worker и очередь только под CAPTCHA-задачи?

Для небольшой нагрузки — нет, при заметном объёме — да.

Решение растягивается на десятки секунд и не должно занимать воркер, обслуживающий быстрые задачи вроде отправки email. Отдельная очередь (queue="captcha") держит долгие задачи изолированно от остального пайплайна, а быстрые задачи не выстраиваются за ними в ожидании.

Итог

Django-приложения подключают CaptchaAI через один сервис-класс, оборачивающий цикл submit/poll к in.php и res.php. Тот же класс обслуживает reCAPTCHA, Turnstile и image CAPTCHA без изменений — меняется только вызывающий код:

• management-команда — для разового запуска и отладки;
• обычная синхронная view — если решение вписывается в бюджет ответа;
• async-view на Django 4.1+ — когда event loop не должен простаивать при опросе res.php;
• Celery-задача — для веб-запросов, где решение не должно блокировать HTTP-ответ, с потоками под тариф CaptchaAI.

Похожие статьи

Дальше по теме — материалы, которые дополняют это руководство:

• GeeTest и Cloudflare Turnstile: что выбрать
• Turnstile возвращает 403 после исправления токена
• Режимы виджета Cloudflare Turnstile