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

Самый частый вопрос при интеграции решения CAPTCHA во Flask — как не заблокировать обработчик запроса на 15–120 секунд, пока CaptchaAI решает reCAPTCHA или Turnstile. Ответ прост: вынести вызов API в отдельный сервисный класс, а тяжёлое ожидание — в фоновый поток. Flask при этом остаётся синхронным, а решение CAPTCHA не держит воркер простаивающим. Ниже — рабочий сервис на requests, маршруты для reCAPTCHA v2 и Cloudflare Turnstile, защита формы через Turnstile и фоновая обработка с опросом статуса задачи.

Подготовка проекта

Понадобятся Flask и requests — сервис общается с API CaptchaAI по HTTP, без дополнительных SDK:

pip install flask requests

Структура приложения

myapp/
├── app.py
├── config.py
├── services/
│   └── captcha_solver.py
└── templates/
    └── form.html

Отдельный модуль services/captcha_solver.py держит логику submit/poll изолированной от маршрутов Flask — это упрощает тестирование и повторное использование в фоновых задачах.

Сервис для работы с CaptchaAI

Класс CaptchaSolver инкапсулирует три шага работы с API CaptchaAI: отправку задачи в in.php, опрос res.php до готовности токена и проверку баланса. Тайм-аут и интервал опроса в примере захардкожены — в проде их стоит вынести в конфиг:

# services/captcha_solver.py
import time
import requests


class CaptchaSolver:
    """CaptchaAI solver service for Flask applications."""

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

    def __init__(self, api_key):
        self.api_key = api_key

    def solve_recaptcha_v2(self, sitekey, page_url):
        """Solve reCAPTCHA v2."""
        return self._submit_and_poll({
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
        })

    def solve_turnstile(self, sitekey, page_url):
        """Solve Cloudflare Turnstile."""
        return self._submit_and_poll({
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": page_url,
        })

    def solve_image(self, image_base64):
        """Solve image CAPTCHA."""
        return self._submit_and_poll({
            "method": "base64",
            "body": image_base64,
        })

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

    def _submit_and_poll(self, params, timeout=120):
        """Submit and poll for result."""
        submit_data = {"key": self.api_key, "json": 1, **params}

        resp = requests.post(f"{self.API_BASE}/in.php", data=submit_data, timeout=30)
        resp.raise_for_status()
        data = resp.json()

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

        task_id = data["request"]

        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

Обратите внимание на time.sleep(5) внутри цикла опроса — это блокирующий вызов. Внутри отдельного потока Python (см. раздел про фоновое решение ниже) это не проблема, но вызов _submit_and_poll напрямую из обработчика Flask означает, что HTTP-воркер занят всё время решения — от нескольких секунд до двух минут в зависимости от типа CAPTCHA.

Не путайте поток Python (threading.Thread) с «потоком» в тарифах CaptchaAI — это единица параллельного биллинга API, а не поток операционной системы. План BASIC ($15/мес, 5 потоков) даёт 5 одновременно решаемых задач на стороне CaptchaAI; план ADVANCE ($90/мес, 50 потоков) — 50. Сколько потоков Python вы при этом запускаете в своём Flask-приложении — вопрос архитектуры сервиса, а не тарифа.

Базовый Flask-сервис с решением CAPTCHA

Минимальный пример — два маршрута для решения reCAPTCHA v2 и Turnstile плюс проверка баланса:

# app.py
from flask import Flask, request, jsonify
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

app = Flask(__name__)
app.config["CAPTCHAAI_API_KEY"] = "YOUR_API_KEY"

solver = CaptchaSolver(app.config["CAPTCHAAI_API_KEY"])


@app.route("/solve/recaptcha", methods=["POST"])
def solve_recaptcha():
    """Solve reCAPTCHA v2 via API."""
    data = request.get_json()
    sitekey = data.get("sitekey")
    page_url = data.get("url")

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

    try:
        token = solver.solve_recaptcha_v2(sitekey, page_url)
        return jsonify({"token": token})
    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/solve/turnstile", methods=["POST"])
def solve_turnstile():
    """Solve Cloudflare Turnstile via API."""
    data = request.get_json()
    sitekey = data.get("sitekey")
    page_url = data.get("url")

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

    try:
        token = solver.solve_turnstile(sitekey, page_url)
        return jsonify({"token": token})
    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/balance", methods=["GET"])
def check_balance():
    """Check CaptchaAI balance."""
    balance = solver.get_balance()
    return jsonify({"balance": balance})


if __name__ == "__main__":
    app.run(debug=True, port=5000)

Проверка запросами

Проверьте маршруты через curl:

# Solve reCAPTCHA
curl -X POST http://localhost:5000/solve/recaptcha \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkSAAAA...", "url": "https://staging.example.com/qa-login"}'

# Solve Turnstile
curl -X POST http://localhost:5000/solve/turnstile \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "0x4AAAAAAAC3DHQ...", "url": "https://example.com/signup"}'

# Check balance
curl http://localhost:5000/balance

Первый запрос вернёт токен reCAPTCHA в поле token — дальше он передаётся в форму или API целевого сайта тем же способом, что и обычный g-recaptcha-response.

Защита формы Flask через Cloudflare Turnstile

Turnstile закрывает форму от автоматических отправок без визуальных загадок для реального посетителя. Маршрут /contact ниже проверяет токен на сервере через siteverify, прежде чем обрабатывать данные формы:

# app.py
from flask import Flask, request, render_template, redirect, url_for, flash
import requests as http_requests

app = Flask(__name__)
app.secret_key = "your-secret-key"
app.config["TURNSTILE_SITE_KEY"] = "0x4AAAAAAAC3DHQhMMQ_Rxrg"
app.config["TURNSTILE_SECRET_KEY"] = "0x4AAAAAAAC3DHQhYYY_secret"


def verify_turnstile(token, remote_ip=None):
    """Verify Turnstile token with Cloudflare."""
    data = {
        "secret": app.config["TURNSTILE_SECRET_KEY"],
        "response": token,
    }
    if remote_ip:
        data["remoteip"] = remote_ip

    resp = http_requests.post(
        "https://challenges.cloudflare.com/turnstile/v0/siteverify",
        data=data,
        timeout=10,
    )
    return resp.json().get("success", False)


@app.route("/contact", methods=["GET", "POST"])
def contact():
    if request.method == "POST":
        turnstile_token = request.form.get("cf-turnstile-response")

        if not turnstile_token:
            flash("CAPTCHA required")
            return redirect(url_for("contact"))

        if not verify_turnstile(turnstile_token, request.remote_addr):
            flash("CAPTCHA verification failed")
            return redirect(url_for("contact"))

        # Process the form
        name = request.form.get("name")
        email = request.form.get("email")
        # ... save or email the data
        flash("Message sent successfully")
        return redirect(url_for("contact"))

    return render_template("form.html",
                           turnstile_sitekey=app.config["TURNSTILE_SITE_KEY"])

<!-- templates/form.html -->
<!DOCTYPE html>
<html>
<body>
    <form method="post">
        <input name="name" placeholder="Name" required>
        <input name="email" type="email" placeholder="Email" required>
        <textarea name="message" placeholder="Message" required></textarea>
        <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>
</body>
</html>

Если форма собирает персональные данные посетителя — имя, email, сообщение, — стоит заранее продумать, где и как долго они хранятся. Для аудитории из РФ это требование 152-ФЗ «О персональных данных», для трансграничной — GDPR-подобная гигиена данных. Собирайте только те поля, которые реально нужны обработчику формы, и не логируйте токен Turnstile целиком в открытом виде.

Фоновое решение через потоки Python

Flask по умолчанию синхронен: пока один запрос ждёт ответа от CaptchaAI, WSGI-воркер занят и не обслуживает других клиентов. Для маршрута, который должен ответить сразу, вынесите решение в отдельный поток Python и опрашивайте статус отдельным эндпоинтом:

import uuid
import threading
from flask import Flask, request, jsonify
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

app = Flask(__name__)
solver = CaptchaSolver("YOUR_API_KEY")

# In-memory task storage (use Redis in production)
tasks = {}


def solve_in_background(task_id, captcha_type, sitekey, page_url):
    """Background CAPTCHA solver."""
    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}")

        tasks[task_id] = {"status": "solved", "token": token}

    except CaptchaSolveError as e:
        tasks[task_id] = {"status": "failed", "error": str(e)}


@app.route("/solve/async", methods=["POST"])
def solve_async():
    """Submit CAPTCHA for background solving."""
    data = request.get_json()
    captcha_type = data.get("type", "recaptcha_v2")
    sitekey = data.get("sitekey")
    page_url = data.get("url")

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

    task_id = str(uuid.uuid4())
    tasks[task_id] = {"status": "pending"}

    thread = threading.Thread(
        target=solve_in_background,
        args=(task_id, captcha_type, sitekey, page_url),
    )
    thread.start()

    return jsonify({"task_id": task_id}), 202


@app.route("/solve/status/<task_id>")
def solve_status(task_id):
    """Check solving status."""
    task = tasks.get(task_id)
    if not task:
        return jsonify({"error": "Task not found"}), 404
    return jsonify(task)

Проверка асинхронного маршрута

Клиент получает task_id сразу, не дожидаясь решения синхронно:

# Submit async solve
curl -X POST http://localhost:5000/solve/async \
  -H "Content-Type: application/json" \
  -d '{"type": "turnstile", "sitekey": "0x4AAA...", "url": "https://example.com"}'
# Returns: {"task_id": "abc-123-..."}

# Check status
curl http://localhost:5000/solve/status/abc-123-...
# Returns: {"status": "pending"}  or  {"status": "solved", "token": "..."}

В продакшене замените словарь tasks на Redis или очередь задач (Celery, RQ) — процесс Flask может перезапуститься под управлением Gunicorn, и данные в оперативной памяти пропадут вместе с ним.

Маршруты CAPTCHA через Flask Blueprint

Когда маршрутов решения становится больше двух-трёх, вынесите их в отдельный Blueprint — стандартный способ Flask группировать связанную функциональность без разрастания app.py:

# blueprints/captcha.py
from flask import Blueprint, request, jsonify, current_app
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

captcha_bp = Blueprint("captcha", __name__, url_prefix="/api/captcha")


def get_solver():
    return CaptchaSolver(current_app.config["CAPTCHAAI_API_KEY"])


@captcha_bp.route("/solve", methods=["POST"])
def solve():
    data = request.get_json()
    captcha_type = data.get("type")
    sitekey = data.get("sitekey")
    url = data.get("url")

    solver = get_solver()

    try:
        if captcha_type == "recaptcha_v2":
            token = solver.solve_recaptcha_v2(sitekey, url)
        elif captcha_type == "turnstile":
            token = solver.solve_turnstile(sitekey, url)
        elif captcha_type == "image":
            image_b64 = data.get("image")
            token = solver.solve_image(image_b64)
        else:
            return jsonify({"error": f"Unknown type: {captcha_type}"}), 400

        return jsonify({"token": token})

    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@captcha_bp.route("/balance")
def balance():
    solver = get_solver()
    return jsonify({"balance": solver.get_balance()})

# app.py
from flask import Flask
from blueprints.captcha import captcha_bp

app = Flask(__name__)
app.config["CAPTCHAAI_API_KEY"] = "YOUR_API_KEY"
app.register_blueprint(captcha_bp)

Типичные проблемы и их устранение

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

Часто задаваемые вопросы

Сколько потоков CaptchaAI нужно для Flask-сервиса под нагрузкой?

Ориентируйтесь на пиковое число одновременных решений, а не на общий объём запросов в день: один поток CaptchaAI держит одну CAPTCHA в моменте решения. Если сервис в среднем обрабатывает 5 одновременных решений, плана BASIC ($15/мес, 5 потоков) достаточно; для 50 одновременных решений нужен ADVANCE ($90/мес, 50 потоков). Реальная параллельность зависит от того, сколько фоновых потоков Python вы запускаете одновременно, а не от общего трафика Flask.

Что делать, если res.php возвращает ERROR_CAPTCHA_UNSOLVABLE?

Это значит, что CaptchaAI не смог решить конкретную задачу — обычно из-за нечитаемого изображения, устаревшего sitekey или несовпадения pageurl с реальным доменом страницы. Повторите отправку с актуальными параметрами; если ошибка повторяется системно, проверьте, что тип CAPTCHA действительно поддерживается.

Стоит ли заменить threading.Thread на Celery или RQ?

Для одного-двух маршрутов решения threading.Thread из примера выше достаточно. Как только фоновых задач становится много или нужен перезапуск без потери очереди, переходите на Celery с Redis или RQ — они переживают рестарт процесса и дают повтор попыток из коробки, чего нет у словаря tasks в оперативной памяти.

Как настроить тайм-аут Gunicorn, если решение занимает больше 120 секунд?

Увеличьте --timeout в Gunicorn (например, --timeout 180) под самый долгий тип CAPTCHA в вашем сценарии — решение может занимать от нескольких секунд до двух минут в зависимости от типа. Тайм-аут WSGI-сервера должен быть больше, чем timeout внутри _submit_and_poll, иначе Gunicorn оборвёт соединение раньше, чем CaptchaAI успеет ответить.

Flask или FastAPI для сервиса решения CAPTCHA?

Flask проще для небольшого синхронного сервиса и хорошо ложится на пример из этого руководства. FastAPI выигрывает, если решений много одновременно: async/await ждёт ответ CaptchaAI, не блокируя обработчик, и не требует ручного управления потоками Python. Логика самого сервиса (CaptchaSolver) при этом почти не меняется.

Итоги

Flask интегрируется с CaptchaAI через один сервисный класс, который отвечает за отправку и опрос задачи. Простые синхронные маршруты подходят, пока решений немного; под нагрузкой — фоновые потоки Python (а на масштабе — Celery или RQ) и Blueprint, чтобы не раздувать app.py. Один и тот же класс CaptchaSolver решает reCAPTCHA v2, Cloudflare Turnstile и image CAPTCHA — тип задачи передаётся параметром, код сервиса не меняется.

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

• Сравнение Geetest и Cloudflare Turnstile
• Cloudflare Turnstile 403 после исправления токена
• Режимы виджета Cloudflare Turnstile: подробное объяснение