أنشئ خدمة صغيرة لحل اختبار CAPTCHA باستخدام FastAPI وCaptchaAI

عندما تحتاج خدمات أو أعضاء فريق متعددون إلى حل اختبار CAPTCHA، فإن مركزيته في خدمة صغيرة يتجنب تكرار المنطق عبر المشاريع. إن الدعم غير المتزامن لـ FastAPI يجعله مناسبًا تمامًا - يتضمن حل اختبار CAPTCHA انتظار استجابات واجهة برمجة التطبيقات الخارجية، والتي تتعامل بشكل غير متزامن بكفاءة دون حظر سلاسل الرسائل.

ينشئ هذا الدليل خدمة FastAPI الصغيرة التي تقبل طلبات حل اختبار CAPTCHA عبر REST وتعيد الرموز المميزة التي تم حلها عبر CaptchaAI.

متى تبني خدمة صغيرة، ومتى يكفي دمج مباشر؟

ما تحتاجه

تثبيت التبعيات:

pip install fastapi uvicorn httpx

هيكل المشروع

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

وحدة حل CaptchaAI

# 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

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

قم بتشغيل الخدمة

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

أمثلة الاستخدام

حل reCAPTCHA v2

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

حل Cloudflare Turnstile

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

الرد:

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

استكشاف الأخطاء وإصلاحها

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

لماذا نستخدم FastAPI لخدمة حل اختبار CAPTCHA؟

يتعامل FastAPI مع I/O غير المتزامن محليًا، وهو مثالي لحل اختبار CAPTCHA حيث يتم قضاء معظم الوقت في انتظار استجابة CaptchaAI. يمكن معالجة طلبات متعددة بشكل متزامن دون الحاجة إلى ترابط.

هل يمكنني إضافة المصادقة إلى نقاط النهاية؟

نعم. أضف حقنة تبعية FastAPI مع التحقق من صحة رأس مفتاح API أو OAuth2 لتقييد الوصول.

كم عدد الطلبات المتزامنة التي يمكن لهذا التعامل معها؟

محدودة بحد المهام المتزامنة لخطة CaptchaAI. يمكن لـ FastAPI نفسه التعامل مع آلاف الاتصالات المتزامنة.

هل يجب علي إرساء هذا؟

نعم. أضف Dockerfile مع FROM python:3.11-slim، وقم بتثبيت التبعيات، واكشف عن المنفذ 8000.

هل يمكنني إضافة حد للسعر؟

نعم. استخدم slowapi أو الوكيل العكسي (nginx، Traefik) للحد من الطلبات لكل عميل.

إذا كنت تريد توحيد منطق الحل ومراقبته في مكان واحد، فخدمة FastAPI الصغيرة تمنحك طبقة داخلية مستقرة يمكن لبقية الأنظمة استدعاؤها بدل تكرار التكامل في كل مشروع.

أدلة ذات صلة

• البدء السريع مع CaptchaAI: حلّ أول كابتشا في 5 دقائق
• استخراج معلمات reCAPTCHA من مصدر الصفحة