Flask 验证码识别教程：CaptchaAI API 实战

给 Flask 应用接入验证码识别，核心只有一件事：把 CaptchaAI 的提交/轮询逻辑封装成一个服务类，路由、后台任务、Blueprint 都只是调用它。国内团队常用 Flask 搭验证码识别网关——比如给采集脚本提供 reCAPTCHA v2 接口，或给后台登录表单接一层 Turnstile 防护。

本文按落地方式拆成三段：

同步端点：收到请求就调用服务类，等结果返回再响应，实现最简单
后台线程轮询：先返回 task_id，识别放独立线程跑，避免阻塞主线程
Blueprint 路由：验证码接口独立成模块，路由多了也好维护

采集脚本调用识别接口前，先确认目标只是你自己有权访问的数据或站点的公开表单——涉及登录态数据采集时，遵循《网络安全法》《数据安全法》等合规要求，别把这套流程接到未授权的第三方系统上。

项目准备

开始前确认环境：

Python 3.9 及以上
Flask 3.x、requests
一个可用的 CaptchaAI API Key（登录控制台的账号设置页即可生成）

pip install flask requests

国内网络访问 pip 官方源较慢时，换清华 TUNA 镜像会快很多：pip install -i https://pypi.tuna.tsinghua.edu.cn/simple flask requests

目录结构

把验证码逻辑单独放进 services 模块，方便路由、后台线程、Blueprint 共用：

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

封装 CaptchaAI 服务类

提交、轮询、异常处理都收敛在这一个类里，上层路由不用关心 HTTP 细节：

# 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

_submit_and_poll 先调用 in.php 拿 task_id，再每 5 秒轮询 res.php，直到拿到 token 或超时。get_balance 单独查余额，方便做限流判断。

基础 Flask 应用：同步接入识别接口

最简单的用法：收到 sitekey 和页面 URL，调用服务类，等结果返回再响应。适合调用量不大、能接受几十秒等待的场景：

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

调用示例

# 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

Turnstile 表单防护实战

只想给表单挡机器人，不用走 CaptchaAI 识别流程——直接用 Cloudflare 官方 Turnstile 校验即可。下面是联系表单示例：服务端拿到 cf-turnstile-response 后向 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>

用线程实现后台识别，避免阻塞主线程

Flask 默认同步，识别逻辑跑完才返回容易触发网关超时。更稳妥的做法是丢进后台线程，先返回 task_id，客户端自己轮询：

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)

调用示例

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

用 Blueprint 组织验证码路由

路由一多，把验证码相关接口独立成一个 Flask Blueprint，比全部塞进 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)

常见故障排查

症状	原因	处理方式
请求挂起 2 分钟以上	同步阻塞主线程	改用后台线程
`ConnectionError`	连不上 CaptchaAI API	检查网络/防火墙
token 返回为空	JSON 解析出错	检查返回格式
Turnstile 验证失败	`TURNSTILE_SECRET_KEY` 配错	核对 Cloudflare 密钥
后台任务内存增长	`tasks` 从未清理	加 TTL 定期清理

常见问题

Flask 项目里 CaptchaAI 能识别哪些验证码类型？

正式支持 reCAPTCHA v2/v3、Turnstile、GeeTest v3、图片/OCR 等 12 种；CaptchaFox、Friendly Captcha、Lemin 为测试版（beta）。hCaptcha、FunCaptcha 不支持，GeeTest v4 尚未上线。

轮询结果会不会卡住 Flask 主线程？

会，同步轮询占满 worker。想不阻塞就用后台线程方案，轮询放独立线程跑，主线程只返回 task_id。

生产环境该选哪个 CaptchaAI 套餐？

按线程数（并发数）计费，不按次数：BASIC $15/月、5 线程，STANDARD $30/月、15 线程，往上还有更高线程数的套餐。按高峰并发估算所需线程数即可。

Gunicorn 部署超时时间设多少合适？

识别通常需要 15–120 秒，比 Gunicorn 默认 30 秒长，启动参数加 --timeout 180，避免请求被网关提前掐断。

识别失败了要怎么重试？

ERROR_CAPTCHA_UNSOLVABLE 说明任务无法识别，重新提交即可；超时或网络错误建议加指数退避重试。

总结

Flask 集成 CaptchaAI 的核心，是把 submit/poll 流程封装进一个服务类：调用量不大用同步端点，要不阻塞就上后台线程，路由多了拆成 Blueprint。同一套服务类可同时处理 reCAPTCHA、Turnstile 和图片验证码。更多细节见 CaptchaAI 官方文档。

使用 CaptchaAI 在 Flask 应用程序中处理验证码

项目准备

目录结构

封装 CaptchaAI 服务类

基础 Flask 应用：同步接入识别接口

调用示例

Turnstile 表单防护实战

用线程实现后台识别，避免阻塞主线程

调用示例

用 Blueprint 组织验证码路由

常见故障排查

常见问题

Flask 项目里 CaptchaAI 能识别哪些验证码类型？

轮询结果会不会卡住 Flask 主线程？

生产环境该选哪个 CaptchaAI 套餐？

Gunicorn 部署超时时间设多少合适？

识别失败了要怎么重试？

总结

相关文章

使用 CaptchaAI 在 Django 应用程序中处理验证码

使用 CaptchaAI 构建客户端验证码管道

弹出模式中的验证码：检测和token 提交

高需求 e-commerce checkout 的 CAPTCHA 测试

使用 CaptchaAI 进行 QA 浏览器配置文件隔离

Bash 脚本 + cURL + CaptchaAI：Shell CAPTCHA 自动化