الدروس التطبيقية

التعامل مع CAPTCHA في تطبيقات Flask باستخدام CaptchaAI

الأتمتة Cloudflare Turnstile
بواسطة فريق CaptchaAI العربي
تاريخ النشر May 03, 2026
التعامل مع CAPTCHA في تطبيقات Flask باستخدام CaptchaAI
الصورة الرئيسية غير متاحة

يقوم دمج CaptchaAI في Flask على فكرة واحدة: فئة خدمة صغيرة تُرسل اختبار CAPTCHA إلى الـ API ثم تستطلع النتيجة حتى يعود الرمز الجاهز، بينما تبقى مسارات Flask نظيفة لا تعرف شيئاً عن تفاصيل الحل. من هذا الفصل تحصل على ثلاثة أنماط عملية تغطي معظم الحالات: نقطة نهاية متزامنة بسيطة، وحل في الخلفية عبر خيوط المعالجة لا يوقف الطلبات، وتنظيم المسارات داخل Flask Blueprint للتطبيقات الأكبر.

خفّة Flask هي ما يجعله الخيار المفضّل لبناء خدمة حل CAPTCHA مستقلة أو بوابة أتمتة داخلية: لا طبقات زائدة، ونقطة النهاية جاهزة في أسطر قليلة. نبني الخدمة هنا خطوة بخطوة، ثم نحمي نماذج Flask بـ Cloudflare Turnstile، ونعالج أوقات الحل الطويلة دون تعطيل الخادم.

أين يظهر هذا في مشروع فعلي؟

تخيّل فريقاً في القاهرة أو الرياض يشغّل منصة حجوزات أو متجراً إلكترونياً، ويحتاج إلى التحقق من أن نماذج التسجيل والدفع المحمية بـ Cloudflare Turnstile ما زالت تعمل بعد كل عملية نشر. بدل تعطيل اختبارات الجودة الآلية عند أول اختبار CAPTCHA، تستدعي مسارات Flask خدمة CaptchaAI فتحصل على رمز صالح وتُكمل السيناريو. تتكرر الحاجة نفسها في حالات كثيرة:

  • أتمتة اختبارات الجودة (QA) لنماذج تحتوي reCAPTCHA v2 أو Turnstile قبل كل إصدار.
  • خدمات استخراج بيانات متاحة للعموم تحتاج إلى حل اختبار CAPTCHA واحد للوصول إلى المحتوى المسموح.
  • بوابة داخلية تجمع طلبات الحل من عدة خدمات صغيرة وتوجّهها إلى مزوّد واحد.
  • معالجة دُفعات من صور CAPTCHA عبر التعرّف الضوئي على الحروف (OCR) قادمة من نظام قديم.

إعداد مشروع Flask

تحتاج الخدمة إلى حزمتين فقط: إطار Flask نفسه، ومكتبة requests التي تتولّى استدعاء الـ API. ثبّتهما داخل بيئة افتراضية نظيفة حتى تبقى تبعيات المشروع معزولة:

pip install flask requests

هيكل مشروع Flask

نفصل منطق الحل في وحدة services/captcha_solver.py ونترك app.py للمسارات فقط؛ هذا الفصل يبقي الشيفرة قابلة للاختبار والصيانة مع نمو المشروع:

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

بناء فئة خدمة CaptchaAI

قلب التكامل هو فئة CaptchaSolver. تُرسل المهمة إلى نقطة النهاية in.php، ثم تستطلع res.php كل خمس ثوانٍ حتى تعود الحالة 1 مع الرمز أو حتى تنتهي المهلة. النمط نفسه — أرسِل ثم استطلِع النتيجة — يعمل مع كل الأنواع المدعومة: reCAPTCHA v2 وv3، وCloudflare Turnstile وChallenge، وGeeTest v3، وصور OCR والشبكات (grid). أما hCaptcha وFunCaptcha فغير مدعومَين حالياً، وGeeTest v4 «قريباً» وليس متاحاً بعد، لذا لا تبنِ مساراً يعتمد عليها:

# 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

المهلة الافتراضية البالغة 120 ثانية تكفي لأغلب الأنواع؛ فبحسب النوع والحِمل قد يكتمل الحل خلال ثوانٍ أو يقترب من الحدّ الأعلى. وعندما تعود ERROR_CAPTCHA_UNSOLVABLE نرفع استثناءً واضحاً بدل إعادة المحاولة إلى ما لا نهاية، ما يمنع تعليق العامل ويجعل الخطأ ظاهراً في السجلّات.

نقاط نهاية Flask لحل reCAPTCHA وTurnstile

بعد جاهزية الخدمة، نكشفها عبر مسارات Flask بسيطة: مسار لكل نوع يستقبل sitekey وurl، ويعيد الرمز في استجابة JSON أو رمز خطأ واضحاً. هذا هو النمط المتزامن المناسب للطلبات المنخفضة التكرار:

# 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: أرسِل sitekey وعنوان الصفحة، وستستقبل الرمز المحلول جاهزاً للحقن في النموذج المستهدف.

# Solve reCAPTCHA
curl -X POST http://localhost:5000/solve/recaptcha \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkSAAAA...", "url": "https://example.com/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

حماية نماذج Flask بـ Cloudflare Turnstile

الوجه الآخر للتكامل هو حماية نماذجك أنت. بدل حلّ اختبار خارجي، تضيف هنا أداة Cloudflare Turnstile إلى نموذج Flask ثم تتحقق من الرمز على الخادم قبل قبول أي إرسال. ينقسم العمل إلى جزأين: دالة تحقق تستدعي واجهة siteverify، وقالب يعرض الأداة ويرسل حقل cf-turnstile-response مع النموذج:

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

لاحظ أن التحقق يجري دائماً على الخادم عبر verify_turnstile؛ فالتحقق من جهة العميل وحده لا يكفي. ومرّر request.remote_addr إلى Cloudflare لربط الرمز بعنوان IP الذي أنشأه.

الحل في الخلفية عبر خيوط المعالجة (Thread)

خادم Flask متزامن بطبيعته، ما يعني أن استدعاءً واحداً للحل قد يوقف الطلب من 15 إلى 120 ثانية. لتفادي ذلك ننقل الحل إلى خيط معالجة (Thread) في الخلفية، ونعيد للعميل معرّف مهمة فوراً ليستطلع الحالة لاحقاً:

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)

تجربة الحل غير المتزامن

أرسِل المهمة فيعود المعرّف والحالة 202 فوراً، ثم استطلِع نقطة /solve/status حتى تتحول الحالة من pending إلى solved:

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

تنظيم المسارات باستخدام Flask Blueprint

عندما يكبر التطبيق، لا يبقى تجميع كل المسارات في app.py عملياً. ننقل مسارات الحل إلى Flask Blueprint مستقل تحت البادئة /api/captcha، فيصبح المنطق قابلاً لإعادة الاستخدام والاختبار بمعزل عن باقي التطبيق. لاحظ أن الخدمة تُنشأ عبر current_app.config بدل متغيّر عام، ما يجعل الإعداد مرناً بين البيئات:

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

معالجة الأخطاء الشائعة

معظم المشكلات في هذا التكامل تعود إلى حجب الخادم المتزامن أو إلى إعداد خاطئ للمفاتيح. يلخّص الجدول الأعراض الأكثر تكراراً وحلّها المباشر:

العَرَض السبب الحل
يتوقف الطلب أكثر من دقيقتين الحل المتزامن يحجب خادم Flask انقل الحل إلى خيط معالجة أو نمط غير متزامن
ConnectionError تعذّر الوصول إلى CaptchaAI API تحقق من الشبكة وإعدادات الجدار الناري
يعود الرمز فارغاً خطأ في تحليل JSON تأكد من تمرير json: 1 وفحص تنسيق الاستجابة
فشل التحقق من Turnstile مفتاح سري خاطئ راجِع قيمة TURNSTILE_SECRET_KEY
نمو الذاكرة مع المهام الخلفية قاموس المهام لا يُنظَّف أبداً أضِف مهلة صلاحية (TTL) وتنظيفاً دورياً

الأسئلة الشائعة

كيف أخزّن مفتاح CaptchaAI API بأمان في تطبيق Flask؟

لا تضع المفتاح في الشيفرة مباشرة. اقرأه من متغيّر بيئة عبر os.environ واحفظه في app.config، واستعن بـ python-dotenv وملف .env أثناء التطوير مع استبعاده من نظام التحكم بالإصدارات. في الإنتاج مرّر المفتاح عبر أسرار المنصة (مثل متغيّرات البيئة في الحاوية) لا عبر ملف مرفوع.

متى أختار الحل في الخلفية بدل النقطة المتزامنة؟

استخدم النقطة المتزامنة حين يكون الطلب نادراً ويحتمل انتظاراً قصيراً. أما إذا كان الحل قد يمتد من 15 إلى 120 ثانية ويحجب عمّال الخادم، فانقله إلى خيط معالجة في الخلفية وأعِد معرّف مهمة للاستطلاع. هكذا يبقى الخادم مستجيباً تحت الحِمل.

ما أنواع CAPTCHA التي تحلّها هذه الخدمة في Flask؟

النمط ذاته يغطي reCAPTCHA v2/v3، وCloudflare Turnstile وChallenge، وGeeTest v3، وصور OCR والشبكات. يكفي إضافة دالة solve_* جديدة بقيمة method المناسبة لكل نوع. لكن hCaptcha وFunCaptcha غير مدعومَين، وGeeTest v4 «قريباً» وغير متاح بعد، فلا تعتمد عليها في مسارك.

كم تكلفة تشغيل خدمة الحل وكيف أوسّعها؟

يعتمد CaptchaAI على تسعير قائم على عدد الـ Threads المتزامنة مع حلول غير محدودة لكل Thread، لا على عدد العمليات. تبدأ الباقات من BASIC ($15 شهرياً، 5 Threads)، وكل اختبار قيد الحل يشغل Thread واحداً حتى ينتهي. للتوسّع أفقياً عبر عدة عمّال Flask، استبدل تخزين المهام في الذاكرة بـ Redis مشترك.

كيف أحمي نقطة نهاية الحل من الاستخدام الخاطئ؟

اجعلها خلف مصادقة داخلية، وحدِّد معدل الطلبات عبر flask-limiter، وقيّد الأصول المسموح بها إذا استُدعيت من المتصفح. هذا يمنع استنزاف رصيدك من طلبات غير مشروعة على نقطة النهاية.

الخلاصة

يتكامل Flask مع CaptchaAI عبر فئة خدمة واحدة تتولّى دورة الإرسال والاستطلاع، بينما تبقى المسارات نظيفة. اختر النقطة المتزامنة للحالات البسيطة، وخيوط المعالجة في الخلفية للحلول التي لا يجب أن توقف الطلبات، وFlask Blueprint لتنظيم التطبيقات الأكبر. الخدمة نفسها تخدم reCAPTCHA وCloudflare Turnstile وصور CAPTCHA دون تغيير في البنية.

مقالات ذات صلة

أدلة ذات صلة

التعليقات غير مفعّلة لهذا المقال.