Xử lý CAPTCHA trong ứng dụng Flask với CaptchaAI

Muốn cắm CaptchaAI vào một ứng dụng Flask có sẵn mà không đổi kiến trúc? Chỉ cần một service class gọi hai endpoint in.php (gửi task) và res.php (lấy kết quả) là đủ. Bài này dành cho dev đã có app Flask và cần một tích hợp chạy được ngay, không phải bài tổng quan lý thuyết.

Bốn mẫu triển khai, dùng lại cùng một service class:

• Route đồng bộ gọi CaptchaAI, trả token trực tiếp.
• Bảo vệ form của chính bạn bằng widget Turnstile.
• Xử lý nền bằng threading khi giải mất 15-120 giây.
• Tổ chức route bằng Flask Blueprint khi app lớn dần.

Chuẩn bị dự án

Chỉ cần Flask và requests - CaptchaAI dùng REST thuần, không cần thư viện client riêng. Yêu cầu tối thiểu:

• Python 3.9 trở lên.
• API key CaptchaAI (lấy trong dashboard sau khi đăng ký).
• venv để tách dependency (khuyến nghị, không bắt buộc).

pip install flask requests

Cấu trúc thư mục gợi ý

Tách phần gọi CaptchaAI ra khỏi route giúp code dễ test và dễ tái dùng lại giữa các Blueprint sau này:

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

Viết service gọi CaptchaAI

CaptchaSolver gói gọn luồng gửi/polling thành ba việc:

1. Gửi task tới in.php với method tương ứng loại CAPTCHA.
2. Polling res.php mỗi 5 giây đến khi status trả về 1 hoặc gặp ERROR_CAPTCHA_UNSOLVABLE.
3. Trả token về, hoặc raise CaptchaSolveError nếu hết timeout.

get_balance() kiểm tra số dư trước khi chạy batch lớn:

# 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

Route Flask gọi CaptchaAI đồng bộ

Cách nhanh nhất để có một endpoint giải CAPTCHA chạy được: route nhận sitekey và url, gọi thẳng CaptchaSolver, trả token khi xong. Phù hợp lưu lượng thấp - request sẽ chờ đến khi CaptchaAI trả kết quả:

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

Kiểm tra nhanh bằng 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

Bảo vệ form liên hệ bằng Cloudflare Turnstile

Đây là chiều ngược lại: thay vì gọi CaptchaAI để giải, bạn tự đặt widget Turnstile lên form của mình để chặn bot. Luồng xác minh gồm ba bước:

1. Nhúng script turnstile/v0/api.js và div.cf-turnstile vào template.
2. Form submit kèm field cf-turnstile-response chứa kết quả widget.
3. Server gửi field đó cùng TURNSTILE_SECRET_KEY sang siteverify của Cloudflare, chỉ xử lý form khi success là true.

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

Giải CAPTCHA nền bằng threading, tránh chặn request

Flask xử lý đồng bộ theo mặc định, nên một lệnh gọi CaptchaAI mất 15-120 giây sẽ giữ request client chờ suốt thời gian đó. Cách xử lý:

• Trả task_id ngay lập tức (HTTP 202) thay vì chờ giải xong.
• Chạy việc giải trong threading.Thread riêng, không chặn worker Flask chính.
• Cho client polling endpoint trạng thái để lấy token khi xong.

Dict tasks chỉ để minh họa - production nên dùng Redis hoặc hàng đợi có TTL:

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)

Gọi thử luồng bất đồng bộ

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

Tổ chức route bằng Flask Blueprint khi app lớn dần

Khi số route giải CAPTCHA tăng lên, gom chúng vào một Blueprint riêng thay vì rải khắp app.py. current_app.config giúp Blueprint lấy CAPTCHAAI_API_KEY mà không cần import trực tiếp app instance - hữu ích khi có nhiều môi trường (dev/staging/production) với key khác nhau.

Đây cũng là mẫu nhiều đội automation tại các công ty outsourcing ở TP.HCM và Hà Nội hay dùng: gom toàn bộ route giải CAPTCHA của nhiều dự án khách hàng vào một Blueprint dùng chung trong service nội bộ, rồi mount vào từng app Flask theo nhu cầu - dễ audit, dễ đổi API key khi luân phiên giữa các gói.

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

Khắc phục sự cố thường gặp

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

Nên dùng Flask hay Django để tích hợp CaptchaAI?

Flask phù hợp hơn cho API và microservice nhẹ - đúng như các ví dụ trong bài. Django hợp hơn khi cần bảng quản trị và nhiều module có sẵn. Cách gọi CaptchaAI giống hệt nhau ở cả hai framework, chỉ khác cách bạn gắn nó vào route.

CaptchaAI có hỗ trợ hCaptcha trong ứng dụng Flask không?

Không. CaptchaAI không hỗ trợ hCaptcha và không hỗ trợ FunCaptcha (Arkose Labs) - đừng thêm method cho hai loại này vào CaptchaSolver. Các loại đang hỗ trợ chính thức gồm reCAPTCHA v2/v3, Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3, CAPTCHA hình ảnh/OCR và BLS; CaptchaFox, Friendly Captcha, Lemin đang ở giai đoạn beta.

Nên polling res.php hay chạy threading nền trong Flask?

Polling đồng bộ (route /solve/recaptcha) đơn giản nhất, đủ dùng khi lưu lượng thấp. Khi endpoint phục vụ nhiều client cùng lúc, threading nền (/solve/async) tốt hơn vì trả task_id ngay, không giữ worker Flask chờ.

Cấu hình timeout cho Flask/Gunicorn khi giải CAPTCHA thế nào?

Đặt timeout WSGI server dài hơn thời gian giải thực tế, ví dụ Gunicorn: --timeout 180. Giải một CAPTCHA qua CaptchaAI thường mất 15-120 giây; dùng mẫu đồng bộ mà không nới timeout, Gunicorn sẽ tự kill worker giữa chừng.

Chi phí giải CAPTCHA trong ứng dụng Flask tính theo gì?

CaptchaAI tính phí theo thread (số CAPTCHA giải đồng thời), không theo số lượt giải - gói BASIC ($15/tháng, 5 thread) cho phép giải không giới hạn số lượt trong mỗi thread. Một app Flask chạy threading nền như trên thường cần vài thread song song là đủ; tăng lên ADVANCE hay PREMIUM khi cần chạy nhiều worker cùng lúc.

Tóm tắt

Flask tích hợp với CaptchaAI qua một service class xử lý luồng gửi/polling. Dùng route đồng bộ cho use case đơn giản, threading nền khi cần trả response ngay, và Blueprint khi app lớn dần và có nhiều route giải CAPTCHA. Cùng một service class xử lý được reCAPTCHA, Turnstile và CAPTCHA hình ảnh.

Bài viết liên quan

• So sánh GeeTest và Cloudflare Turnstile
• Khắc phục lỗi Cloudflare Turnstile 403 sau khi sửa token
• Các chế độ widget Cloudflare Turnstile