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

Birden fazla hizmetin veya ekip üyesinin CAPTCHA çözümüne ihtiyacı olduğunda, bunu bir mikro hizmette merkezileştirmek, projeler arasında mantığın kopyalanmasını önler. FastAPI'nin eşzamansız desteği onu güçlü bir uyum haline getiriyor - CAPTCHA çözümü, iş parçacıklarını engellemeden eşzamansız olarak verimli bir şekilde işlenen harici API yanıtlarının beklenmesini içeriyor.

Bu kılavuz, REST üzerinden CAPTCHA çözme isteklerini kabul eden ve çözülmüş belirteçleri CaptchaAI aracılığıyla döndüren bir FastAPI mikro hizmeti oluşturur.

İhtiyacınız olan şey

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

pip install fastapi uvicorn httpx

Proje yapısı

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

CaptchaAI çözücü modülü

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

FastAPI uygulaması

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

Hizmeti çalıştırın

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

Kullanım örnekleri

reCAPTCHA v2'yi çözü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"}'

Cloudflare Turnstile'yi çözü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..."
}

Dağıtım güçlendirme notları

• API anahtarını ortamdan okuyun ve sır eksikse başlatma sırasında hızlı bir şekilde başarısız olun.
• İstek doğrulamasını çözümleyicinin yürütülmesinden ayırın, böylece geçersiz veriler hiçbir zaman giden çağrı yoluna ulaşmaz.
• Doğrulama hatalarını, çözümleyici hatalarını ve yukarı akış reddini ayırt eden yapılandırılmış hata yanıtlarını döndürün.

Sorun giderme

SSS

CAPTCHA çözme hizmeti için neden FastAPI kullanılmalı?

FastAPI, zaman uyumsuz I/O'yi yerel olarak işler; bu, CaptchaAI'nin yanıtını beklerken çoğu zamanın harcandığı CAPTCHA çözümü için idealdir. Birden fazla istek iş parçacığı olmadan aynı anda işlenebilir.

Uç noktalara kimlik doğrulama ekleyebilir miyim?

Evet. Erişimi kısıtlamak için FastAPI'nin bağımlılık eklemesini API anahtar başlığı doğrulaması veya OAuth2 ile ekleyin.

Bu, kaç eşzamanlı isteği karşılayabilir?

CaptchaAI planınızın eşzamanlı görev sınırıyla sınırlıdır. FastAPI'nin kendisi binlerce eşzamanlı bağlantıyı yönetebilir.

Bunu dockerize etmeli miyim?

Evet. FROM python:3.11-slim ile bir Dockerfile ekleyin, bağımlılıkları yükleyin ve 8000 numaralı bağlantı noktasını kullanıma açın.

Hız sınırlaması ekleyebilir miyim?

Evet. İstemci başına istekleri sınırlamak için slowapi veya ters proxy (nginx, Traefik) kullanın.

CAPTCHA çözme mikro hizmetinizi oluşturun

API anahtarınızı şu adresten alın:captchaai.com. FastAPI mikro hizmetiyle CAPTCHA çözümünü merkezileştirin.

İlgili kılavuzlar

• API Kullanarak reCAPTCHA v2 Nasıl Çözülür?
• API Kullanarak Cloudflare Turnstile Nasıl Çözülür?
• aiohttp ile eşzamansız CaptchaAI
• Paralel CAPTCHA Çözümü