Penanganan CAPTCHA pada Aplikasi Flask dengan CaptchaAI

Endpoint Flask yang memanggil solver CAPTCHA secara sinkron memblokir worker WSGI-nya sampai proses solve selesai — untuk reCAPTCHA v2 atau Turnstile itu bisa berarti request menggantung 10 sampai 60 detik, kadang lebih lama kalau CAPTCHA-nya berat. Panduan ini menunjukkan cara mengintegrasikan CaptchaAI ke aplikasi Flask dengan pola yang benar sejak awal, mulai dari service class sampai aplikasi berskala produksi.

Lima building block yang akan dibangun berurutan:

• Service class CaptchaAI — satu class yang menangani submit dan polling.
• Endpoint solving sinkron — solve reCAPTCHA dan Turnstile langsung dari route.
• Proteksi form Turnstile — verifikasi token di sisi server tanpa memanggil CaptchaAI.
• Background thread — solve tanpa memblokir worker Flask.
• Flask Blueprint — kelompokkan route CAPTCHA untuk aplikasi yang lebih besar.

Konteks lokal: pola ini umum dipakai tim otomatisasi dan agensi price-monitoring di Indonesia yang membangun layanan solving sebagai microservice terpisah dari aplikasi utama. API Flask yang ringan biasanya dideploy di region seperti Singapura (ap-southeast-1) atau Jakarta (asia-southeast2) supaya latency ke API CaptchaAI dan ke situs target tetap rendah.

Menyiapkan struktur proyek Flask

Mulai dari struktur minimal berikut. Modul captcha_solver.py sengaja dipisah ke folder services/ supaya service class-nya bisa dipakai ulang oleh endpoint sinkron, background thread, maupun Blueprint di bagian selanjutnya.

pip install flask requests

Struktur aplikasi

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

Service class CaptchaAI: submit dan polling

CaptchaSolver adalah satu lapisan tunggal yang bicara langsung ke API CaptchaAI. solve_recaptcha_v2() dan solve_turnstile() cuma membungkus parameter method berbeda, lalu keduanya lewat _submit_and_poll() yang sama: kirim task ke in.php, simpan task_id, lalu polling res.php tiap 5 detik sampai status bernilai 1 atau muncul ERROR_CAPTCHA_UNSOLVABLE. Timeout default 120 detik sudah longgar untuk reCAPTCHA v2 maupun Turnstile; naikkan hanya untuk tipe CAPTCHA yang solve time-nya lebih panjang. get_balance() memanggil getbalance — cocok dipanggil dari health-check supaya saldo menipis ketahuan sebelum task mulai gagal.

# 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

Endpoint Flask untuk solve reCAPTCHA dan Turnstile

Dengan service class siap, endpoint Flask-nya tinggal validasi input lalu tangani CaptchaSolveError. Dua endpoint di bawah — /solve/recaptcha dan /solve/turnstile — sama-sama mengembalikan token siap pakai lewat field token, dan /balance mengekspos saldo API supaya bisa dipantau dari luar aplikasi.

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

Contoh permintaan

# 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

Proteksi form Flask dengan Cloudflare Turnstile

Untuk form biasa (kontak, registrasi), verifikasi Turnstile dilakukan langsung di sisi server saat form disubmit — tidak perlu memanggil CaptchaAI sama sekali, karena yang diverifikasi adalah widget Turnstile milik situs Anda sendiri, bukan menyelesaikan CAPTCHA di situs pihak lain. verify_turnstile() mengirim token dari field cf-turnstile-response ke endpoint siteverify Cloudflare beserta remoteip pemohon; route /contact menolak submit kalau token kosong atau verifikasi gagal:

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

Background solving dengan threading agar Flask tidak terblokir

Flask itu sinkron — satu request memakai satu worker sampai tuntas. Kalau solve CAPTCHA butuh 30 detik, worker itu terkunci selama itu juga dan request lain harus antre. Tanda-tanda saatnya pindah dari endpoint sinkron ke pola ini:

• Solve time rata-rata sudah mendekati batas timeout WSGI server Anda.
• Worker Gunicorn/uWSGI kehabisan slot saat trafik naik, padahal CPU masih longgar.
• Klien Anda sanggup polling status, bukan cuma menunggu satu response panjang.

Pola /solve/async di bawah memecah masalah ini: endpoint langsung mengembalikan task_id berstatus 202, sementara threading.Thread menjalankan solve di background dan menyimpan hasilnya ke dict tasks. Klien lalu polling /solve/status/<task_id> sampai statusnya berubah dari pending menjadi solved atau failed — persis pola submit/poll CaptchaAI sendiri, cuma dilapis sekali lagi di aplikasi Anda. Untuk produksi, ganti dict in-memory dengan Redis supaya status task tidak hilang saat Flask di-restart atau di-scale.

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)

Contoh permintaan

# 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": "..."}

Pola Flask Blueprint untuk aplikasi skala besar

Begitu aplikasi Anda punya lebih dari dua-tiga route CAPTCHA, pindahkan ke Blueprint. captcha_bp mengelompokkan semua route solving di bawah prefix /api/captcha, dan get_solver() membaca API key dari current_app.config alih-alih hardcode — Blueprint yang sama jadi bisa dipasang ke beberapa environment (staging, production) dengan API key berbeda tanpa mengubah kode:

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

Troubleshooting: masalah umum integrasi CaptchaAI di Flask

Gejala yang paling sering muncul saat memasang service class ini di Flask, dan cara mengatasinya:

Pertanyaan yang sering diajukan

Apakah CaptchaAI mendukung semua tipe CAPTCHA yang saya temui di endpoint Flask?

Sebagian besar iya — reCAPTCHA v2/v3, Turnstile dan Challenge, GeeTest v3, CAPTCHA gambar, dan BLS. hCaptcha, FunCaptcha (Arkose Labs) belum didukung; GeeTest v4 masih segera hadir. Kalau form target memakai salah satu dari tiga tipe itu, service class ini tidak bekerja untuknya.

Bagaimana cara menangani timeout request kalau solve CAPTCHA lama?

Set timeout WSGI server Anda lebih longgar daripada waktu solve terpanjang — untuk Gunicorn misalnya --timeout 180. Endpoint sinkron di panduan ini cocok untuk solve singkat; begitu solve time mendekati batas timeout server, pindah ke pola background thread supaya request tidak digantung sampai terputus.

Berapa thread CaptchaAI yang cukup untuk endpoint solving dengan trafik tinggi?

Sesuaikan jumlah thread dengan kapasitas endpoint Anda, bukan dengan jumlah request per hari:

• BASIC — $15/bulan, 5 thread, cukup untuk endpoint bertrafik rendah atau tahap uji coba.
• ADVANCE — $90/bulan, 50 thread, untuk endpoint yang mulai menangani puluhan solve serentak.
• PREMIUM — $170/bulan, 100 thread, untuk trafik solving yang konsisten tinggi.

Setiap thread menyertakan solve tak terbatas, jadi menambah trafik tidak menambah biaya per solve — hanya jumlah thread yang perlu naik.

Apakah solve di background thread bisa mengganggu request Flask lain?

Selama threading.Thread hanya menunggu I/O (request HTTP ke CaptchaAI), thread background tidak memblokir worker Flask secara berarti. Yang perlu diawasi justru jumlah thread yang menumpuk kalau trafik solve sangat tinggi; di titik itu task queue seperti Celery atau RQ lebih terkontrol daripada spawn thread per request.

Sebaiknya pakai Flask atau Django untuk integrasi CaptchaAI?

Flask lebih ringan untuk API atau microservice solving yang berdiri sendiri. Django lebih cocok kalau aplikasi Anda sudah punya admin panel dan model data lengkap. Pola integrasinya sama persis di keduanya — service class submit/poll ini bisa dipakai ulang tanpa perubahan.

Ringkasan

Flask cocok jadi lapisan tipis di atas CaptchaAI: service class yang sama menangani submit dan polling untuk reCAPTCHA, Turnstile, dan CAPTCHA gambar — dipanggil dari endpoint sinkron untuk kasus sederhana, dari background thread saat solve time berpotensi lama, atau dari Blueprint begitu route CAPTCHA-nya pantas dikelompokkan. Arsitekturnya tidak perlu ditulis ulang saat trafik naik, cuma dipanggil lewat pola berbeda. Pelajari lebih lanjut di CaptchaAI.

Artikel terkait

• Mengatasi error 403 Turnstile setelah token diverifikasi
• Memahami mode widget Cloudflare Turnstile
• Penanganan CAPTCHA pada aplikasi Django dengan CaptchaAI