FastAPI ve CaptchaAI ile CAPTCHA Çözme Mikro Hizmeti oluşturun

CAPTCHA çözme mantığını her otomasyon projesine ayrı ayrı gömmek yerine tek bir serviste toplarsanız, hem bakım kolaylaşır hem de ekipteki herkes aynı çözüm yolunu kullanır. Bu rehberin sonunda elinizde çalışan bir FastAPI mikro hizmeti olacak: REST üzerinden istek alan, CaptchaAI ile çözülmüş token'ı döndüren ve reCAPTCHA v2, reCAPTCHA v3, Cloudflare Turnstile ile görüntü CAPTCHA'larını tek bir servisten karşılayan bir yapı.

FastAPI'yi seçmemizin nedeni basit: CAPTCHA çözümünde zamanın büyük kısmı CaptchaAI'nin yanıtını beklemekle geçer. FastAPI'nin yerel async desteği bu bekleme süresini thread'leri bloke etmeden yönetir, böylece tek bir işlem onlarca çözümü aynı anda yürütebilir.

Neden ayrı bir mikro hizmet?

Türkiye'deki e-ticaret ve fintech ekiplerinin çoğu, giriş ve ödeme adımı akışlarını otomatik QA testleriyle sürekli doğruluyor. Bu testlerin her biri CAPTCHA çözümüne aynı anda ihtiyaç duyduğunda, çözme mantığını tek bir uç noktanın arkasına almak birkaç somut avantaj getirir:

• API anahtarını tek yerde saklarsınız; onu birden fazla repoya kopyalamazsınız.
• Eşzamanlılık ve zaman aşımı politikalarını merkezî olarak ayarlarsınız.
• CaptchaAI planınızdaki thread sayısı, tüm ekibin paylaştığı gerçek eşzamanlılık tavanınız olur — dağınık script'lerde bunu takip etmek zordur.

CaptchaAI'nin thread tabanlı fiyatlandırması bu modele iyi oturur: BASIC planı ($15/ay, 5 thread) ile başlayıp yük arttıkça ADVANCE'a ($90/ay, 50 thread) geçebilirsiniz. Fiyatlar USD üzerinden sabit kaldığı için TL dalgalanmasından bağımsız, öngörülebilir bir maliyet elde edersiniz.

Ön koşullar

Bağımlılıkları yükleyin:

pip install fastapi uvicorn httpx

requests yerine httpx kullanmamız kasıtlı: senkron bir istemci her çağrıda event loop'u bloke eder ve FastAPI'nin async avantajını sıfırlar.

Proje yapısı

Servisi iki dosyada tutuyoruz — çözüm mantığı ile HTTP katmanı ayrı kalsın:

captcha-service/
├── main.py          # FastAPI app with endpoints
├── solver.py        # CaptchaAI solving logic
└── requirements.txt

Çözüm modülü: solver.py

Bu modül CaptchaAI ile konuşan tek yerdir. submit_task görevi in.php'ye gönderir ve görev kimliğini döndürür; poll_result sonucu res.php üzerinden periyodik olarak sorgular. Her CAPTCHA türü için ayrı bir yardımcı fonksiyon var, ancak hepsi aynı gönderim ve sorgulama akışını paylaşır.

# solver.py
import httpx
import asyncio

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


async def submit_task(params: dict) -> str:
    """Submit a CAPTCHA task and return the task ID."""
    params["key"] = API_KEY
    params["json"] = 1

    async with httpx.AsyncClient() as client:
        response = await client.post(f"{BASE_URL}/in.php", data=params)
        data = response.json()

    if data.get("status") != 1:
        raise ValueError(f"Submit error: {data.get('request')}")
    return data["request"]


async def poll_result(task_id: str, initial_wait: int = 15, max_attempts: int = 30) -> dict:
    """Poll for the CAPTCHA result."""
    await asyncio.sleep(initial_wait)

    async with httpx.AsyncClient() as client:
        for _ in range(max_attempts):
            response = await client.get(f"{BASE_URL}/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id, "json": 1
            })
            data = response.json()

            if data.get("status") == 1:
                return {
                    "token": data["request"],
                    "user_agent": data.get("user_agent", "")
                }
            if data.get("request") != "CAPCHA_NOT_READY":
                raise ValueError(f"Solve error: {data['request']}")

            await asyncio.sleep(5)

    raise TimeoutError("Solve timed out")


async def solve_recaptcha_v2(sitekey: str, pageurl: str, enterprise: bool = False) -> dict:
    params = {"method": "userrecaptcha", "googlekey": sitekey, "pageurl": pageurl}
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)


async def solve_recaptcha_v3(sitekey: str, pageurl: str, action: str, enterprise: bool = False) -> dict:
    params = {
        "method": "userrecaptcha", "version": "v3",
        "googlekey": sitekey, "pageurl": pageurl, "action": action
    }
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)


async def solve_turnstile(sitekey: str, pageurl: str) -> dict:
    task_id = await submit_task({"method": "turnstile", "sitekey": sitekey, "pageurl": pageurl})
    return await poll_result(task_id, initial_wait=10)


async def solve_image(image_base64: str) -> dict:
    task_id = await submit_task({"method": "base64", "body": image_base64})
    return await poll_result(task_id, initial_wait=5, max_attempts=15)

Görev tipine göre initial_wait değerinin farklı olduğuna dikkat edin: Turnstile genelde daha hızlı döner, görüntü CAPTCHA'ları ise en kısa beklemeyle başlar. Bu değerler ilk sorgulamayı boşa harcamamak içindir.

FastAPI uygulaması: main.py

HTTP katmanı ince kalır: her uç nokta gelen isteği bir Pydantic modeliyle doğrular, solver'ı çağırır ve token'ı döndürür. Çözücü bir ValueError veya TimeoutError fırlatırsa, bunu istemciye 502 olarak iletiriz — böylece giden çağrıdaki bir sorun ile geçersiz istek gövdesi birbirine karışmaz.

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import solver

app = FastAPI(title="CaptchaAI Solver Service")


class RecaptchaV2Request(BaseModel):
    sitekey: str
    pageurl: str
    enterprise: bool = False


class RecaptchaV3Request(BaseModel):
    sitekey: str
    pageurl: str
    action: str
    enterprise: bool = False


class TurnstileRequest(BaseModel):
    sitekey: str
    pageurl: str


class ImageRequest(BaseModel):
    image_base64: str


class SolveResponse(BaseModel):
    token: str
    user_agent: Optional[str] = ""


@app.post("/solve/recaptcha-v2", response_model=SolveResponse)
async def solve_recaptcha_v2(req: RecaptchaV2Request):
    try:
        result = await solver.solve_recaptcha_v2(req.sitekey, req.pageurl, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/recaptcha-v3", response_model=SolveResponse)
async def solve_recaptcha_v3(req: RecaptchaV3Request):
    try:
        result = await solver.solve_recaptcha_v3(req.sitekey, req.pageurl, req.action, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/turnstile", response_model=SolveResponse)
async def solve_turnstile(req: TurnstileRequest):
    try:
        result = await solver.solve_turnstile(req.sitekey, req.pageurl)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.post("/solve/image", response_model=SolveResponse)
async def solve_image(req: ImageRequest):
    try:
        result = await solver.solve_image(req.image_base64)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "ok"}

/health uç noktası, servisi Docker veya Kubernetes altında çalıştırdığınızda yük dengeleyicinin canlılık kontrolü için işinize yarar.

Servisi çalıştırın

uvicorn main:app --host 0.0.0.0 --port 8000

Uç noktaları test edin

Servis ayağa kalktıktan sonra her uç noktayı curl ile doğrulayabilirsiniz. Aşağıdaki örneklerde hedef URL olarak canlı bir site yerine bir QA staging adresi kullanılıyor — kendi test ortamınızın adresiyle değiştirin.

reCAPTCHA v2 uç noktasını çağırın

curl -X POST http://localhost:8000/solve/recaptcha-v2 \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkS...", "pageurl": "https://staging.example.com/qa-login"}'

Turnstile uç noktasını çağırın

curl -X POST http://localhost:8000/solve/turnstile \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "0x4AAAA...", "pageurl": "https://example.com/form"}'

Yanıt:

{
  "token": "03AGdBq24PBCqLmOx2V4...",
  "user_agent": "Mozilla/5.0..."
}

Dönen token'ı ve gerektiğinde user_agent'ı, hedef formu gönderirken aynı istek içinde kullanın.

Üretime almadan önce

Örnek kod çalışır durumda, ancak birkaç ekleme onu üretime hazır hale getirir:

• API anahtarını ortam değişkeninden okuyun. Anahtar eksikse servisi başlatma sırasında hemen hata vererek durdurun; sessizce çalışıp ilk istekte patlamasın.
• Doğrulamayı çözümden ayrı tutun. İstek doğrulaması Pydantic modeli katmanında biter; geçersiz veri hiçbir zaman giden CaptchaAI çağrısına ulaşmaz.
• Yapılandırılmış hata yanıtları döndürün. Doğrulama hatası, çözücü hatası ve yukarı akış (upstream) reddini birbirinden ayırt edin ki çağıran taraf doğru tepkiyi verebilsin.
• İstek sınırlaması ekleyin. İstemci başına istekleri sınırlamak için slowapi veya bir ters proxy (nginx, Traefik) kullanın.
• Docker ile paketleyin. FROM python:3.11-slim tabanlı bir Dockerfile yazın, bağımlılıkları yükleyin ve 8000 numaralı bağlantı noktasını açın.

Sorun giderme

Sık sorulan sorular

Bu mikro hizmet hangi CAPTCHA türlerini çözebilir?

Örnek kod reCAPTCHA v2, reCAPTCHA v3, Cloudflare Turnstile ve görüntü tabanlı CAPTCHA'ları kapsar. CaptchaAI ayrıca GeeTest v3, Cloudflare Challenge ve BLS gibi türleri de destekler; solver modülüne benzer bir yardımcı fonksiyon ekleyerek bunları da açabilirsiniz. hCaptcha ve FunCaptcha ise şu anda desteklenmiyor.

Aynı anda kaç eşzamanlı çözüm işleyebilir?

Sınır, CaptchaAI planınızın thread sayısıdır: BASIC 5, STANDARD 15, ADVANCE 50 eşzamanlı görev. FastAPI'nin kendisi binlerce bağlantıyı yönetir, ancak gerçek çözüm eşzamanlılığınızı plan belirler.

Üretimde requests kullanmam sorun olur mu?

Evet. requests senkrondur ve her çağrıda event loop'u bloke ederek async'in kazandırdığı eşzamanlılığı yok eder. Bu servis boyunca httpx.AsyncClient'te kalın.

Uç noktalara kimlik doğrulama nasıl eklenir?

FastAPI'nin bağımlılık ekleme (dependency injection) mekanizmasını kullanın: bir API anahtarı başlığı doğrulaması ya da OAuth2 ekleyerek servisi yalnızca yetkili istemcilere açabilirsiniz.

Bu servis için hangi CaptchaAI planı yeterli?

İhtiyacınız eşzamanlı görev sayısına bağlıdır. Küçük bir QA ekibi için BASIC ($15/ay, 5 thread) genellikle yeterlidir; birden çok test paketini paralel çalıştıran ekipler ADVANCE ($90/ay, 50 thread) veya üstünü tercih eder. Tüm planlarda thread başına çözüm sınırsızdır.

CAPTCHA çözümünü tek serviste toplayın

API anahtarınızı şu adresten alın: captchaai.com. Ardından çözme mantığını bu FastAPI mikro hizmetinde toplayarak tüm otomasyon projelerinizin aynı, bakımı kolay uçtan yararlanmasını sağlayın.

İlgili kılavuzlar

• reCAPTCHA v2'yi API ile çözme rehberi
• Cloudflare Turnstile'ı API ile çözme rehberi
• aiohttp ile eşzamansız CaptchaAI entegrasyonu
• Paralel CAPTCHA çözümü