Tutorials

Flask Uygulamalarında CaptchaAI ile CAPTCHA Çözme

Flask ile temiz bir CAPTCHA çözüm katmanı kurmanın yolu üç desenden geçer: tüm çözüm mantığını tek bir servis sınıfında toplayın, 15–120 saniye süren çağrıları bir thread'e taşıyın ve rotaları Flask Blueprint'leriyle düzenleyin. Bu üç adım, birkaç satırlık bir demo'yu üretime dayanıklı bir servise dönüştürür.

Flask'ın minimal yapısı bu iş için biçilmiş kaftan: hafif, açık ve zorunlu bir iş kuyruğu ya da async framework olmadan CaptchaAI API'sini sarmalamanıza izin verir. Aşağıda servis sınıfından senkron uç noktalara, arka plan thread'lerinden Blueprint'lere kadar tam bir entegrasyonu adım adım kuruyoruz.

Hangi deseni seçeceğinize hızlıca karar verin: tek seferlik bir betik ya da düşük trafikli bir dahili araç yazıyorsanız senkron uç nokta yeterlidir. İstemcinin beklememesi gereken bir kullanıcı akışınız varsa arka plan thread'leri devreye girer. Birden çok CAPTCHA türünü, farklı ekiplerin baktığı büyük bir kod tabanında yönetiyorsanız Blueprint deseni işleri temiz tutar. Bu rehberdeki servis sınıfı üçünde de aynı kalır; değişen yalnızca onu nasıl çağırdığınızdır.


Kurulum ve proje yapısı

pip install flask requests

Dizin yapısı

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

CaptchaAI servis sınıfı

Tüm submit/poll mantığını tek bir sınıfta toplayın; rotalarınız yalnızca bu sınıfı çağırsın. Böylece hata yönetimi, zaman aşımı ve yeniden deneme tek yerde kalır.

# 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 görevi in.php uç noktasına gönderir, ardından res.php üzerinden sonucu her 5 saniyede bir sorgular ve token'ı döndürür. Aynı akış tüm CAPTCHA türleri için kullanılır; tek değişen method parametresidir.


Temel Flask uygulaması: çözüm uç noktaları

Servis sınıfını ince bir Flask katmanının arkasına koyun. Her rota isteği doğrular, çözücüyü çağırır ve token'ı JSON olarak döner.

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

Örnek istekler

# 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

Flask formlarını Turnstile ile koruma

Türkiye'deki e-ticaret ve fintech ekipleri Turnstile'ı çoğunlukla iletişim, kayıt ve ödeme adımı formlarında kullanır. Formu Flask tarafında Turnstile ile koruyup token'ı Cloudflare'de doğrulayarak bu akışların QA otomasyonunu güvenle test edebilirsiniz:

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

Buradaki cf-turnstile-response alanı, widget'ın forma yerleştirdiği token'ı taşır; sunucu tarafı doğrulama bu token'ı Cloudflare'in siteverify uç noktasına gönderir.


Thread ile arka planda çözüm (engellemeyen)

Flask varsayılan olarak senkrondur; bir çözüm çağrısı 15–120 saniye sürebildiği için isteği doğrudan bloke eder. Görevi bir thread'e taşıyın, istemciye hemen bir task_id dönün ve sonucu ayrı bir uç noktadan sorgulatın:

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)

Async istek örnekleri

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

Bellek içi tasks sözlüğü yalnızca demo içindir; birden fazla worker ile çalışan üretimde durumu Redis gibi paylaşımlı bir depoda tutun.


Flask Blueprint ile rota düzenleme

Büyüyen uygulamalarda CAPTCHA rotalarını tek dosyada tutmak yerine bir Flask Blueprint altında toplayın; böylece uygulamanız modüler kalır:

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

Sorun giderme

Belirti Sebep Çözüm
İstek 2+ dakika askıda kalıyor Senkron çözüm Flask worker'ını bloke ediyor Thread veya async desenini kullanın
ConnectionError CaptchaAI API'sine ulaşılamıyor Ağı ve firewall ayarlarını kontrol edin
Token boş döndü JSON ayrıştırma hatası Yanıt formatını doğrulayın
Turnstile doğrulaması başarısız oluyor Yanlış secret key TURNSTILE_SECRET_KEY değerini kontrol edin
Arka plan görevlerinde bellek şişiyor tasks sözlüğü hiç temizlenmiyor TTL ekleyin ve tamamlanan görevleri temizleyin

Sık sorulan sorular

Flask senkronken CAPTCHA çözümünü nasıl engellemeyen hale getiririm?

Çözüm çağrısı 15–120 saniye sürebildiği için senkron bir uç nokta Flask worker'ını bloke eder. Görevi bir thread'e taşıyın, istemciye hemen bir task_id dönün ve sonucu ayrı bir durum uç noktasından sorgulatın.

CaptchaAI bu servis sınıfıyla hangi CAPTCHA türlerini çözer?

Örnek servis reCAPTCHA v2/v3, Cloudflare Turnstile ve resim/OCR CAPTCHA'larını çözer; aynı _submit_and_poll akışına yeni method değerleri ekleyerek genişletebilirsiniz. hCaptcha ve FunCaptcha (Arkose Labs) desteklenmez.

Flask projem için hangi CaptchaAI planı uygun?

Fiyatlandırma thread tabanlıdır: maliyeti belirleyen, aynı anda kaç CAPTCHA çözdüğünüzdür; çözüm başına ücret yoktur. Düşük hacimli bir Flask servisi için BASIC ($15/ay, 5 thread) yeterli olur ve sabit USD aylık ücret, TL kur dalgalanmasından etkilenmeyen öngörülebilir bir maliyet sağlar.

Çözüm uzun sürerse istek zaman aşımını nasıl ayarlarım?

WSGI sunucunuzda zaman aşımını yükseltin (Gunicorn için --timeout 180). _submit_and_poll içindeki 120 saniyelik iç zaman aşımını da iş yükünüze göre değiştirebilirsiniz.

Çözüm uç noktamı hız sınırlamaya almalı mıyım?

Evet. Uç noktanın kötüye kullanılmasını önlemek için flask-limiter ile IP başına istek sınırı koyun; halka açık bir çözüm servisi çalıştırıyorsanız bu adım şarttır.

Üretimde thread yerine Celery kullanmalı mıyım?

Küçük ve orta yüklerde threading ve bir Redis destekli görev deposu fazlasıyla yeterlidir. Ancak yeniden deneme, kalıcı kuyruk ve birden çok worker arasında yük dağıtımı gerektiğinde Celery ya da RQ gibi bir kuyruk sistemine geçmek daha sağlıklıdır; servis sınıfı aynı kalır, yalnızca çağrıldığı yer görev işleyicisine taşınır.

Bir freelance otomasyon projesinde kişisel veri işlerken nelere dikkat etmeliyim?

Bionluk veya Upwork üzerinden aldığınız QA ve veri toplama işlerinde kazınan kişisel veriler Türkiye'de KVKK kapsamına girer. CaptchaAI'yi yalnızca yetkilendirilmiş QA ve veri toplama akışlarında kullanın, üzerinde hakkınız olmayan oturum arkası verilere erişmeyin ve müşterinizin veri işleme iznini yazılı olarak netleştirin.


Özet

CaptchaAI, submit/poll akışını yöneten bir servis sınıfı üzerinden Flask'a entegre olur. Basit senaryolar için senkron uç noktaları, engellemeyen çözümler için arka plan thread'lerini ve büyük projelerde düzenli rotalar için Flask Blueprint'lerini kullanın. Aynı servis sınıfı reCAPTCHA, Turnstile ve resim CAPTCHA'larını tek noktadan yönetir.

İlgili makaleler

Bu makale için yorumlar devre dışı bırakılmıştır.