بناء خدمة مصغّرة لحل CAPTCHA باستخدام FastAPI وCaptchaAI

خدمة FastAPI مصغّرة تحوّل حلّ CAPTCHA إلى نقطة نهاية REST واحدة تستدعيها بقية أنظمتك، بدل أن يعيد كل مشروع كتابة منطق التكامل مع CaptchaAI من الصفر. تخيّل شركة إقليمية لديها ثلاثة فرق تعمل بالتوازي: فريق يشغّل اختبارات الجودة على بوابة الدفع، وآخر يجمع بيانات المنافسين، وثالث يؤتمت نماذج التسجيل — وكلها تصطدم بـ reCAPTCHA وTurnstile. بدل تكرار كود الإرسال والاستطلاع الدوري في كل مشروع، تستدعي الفرق الثلاثة الخدمة نفسها وتحصل على الرمز المحلول من مصدر واحد.

يناسب FastAPI هذه المهمة تحديداً لأن حلّ CAPTCHA عملية تقضي معظم وقتها في انتظار استجابة CaptchaAI، والنموذج غير المتزامن يخدم عشرات الطلبات في آنٍ واحد دون حجز خيوط المعالجة. تعمل الخدمة التي نبنيها هنا وفق أربع خطوات ثابتة لكل نوع:

1. تستقبل نقطة REST بيانات الطلب (مفتاح الموقع وعنوان الصفحة).
2. ترسل المهمة إلى CaptchaAI وتحتفظ بمعرّف المهمة.
3. تستطلع النتيجة دورياً حتى يجهز الرمز.
4. تعيد الرمز المحلول في استجابة JSON منظمة.

المتطلبات الأساسية

• مفتاح CaptchaAI API من captchaai.com
• Python 3.9+ بيئة التشغيل
• FastAPI + httpx للتعامل مع طلبات HTTP غير المتزامنة

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

pip install fastapi uvicorn httpx

بنية المشروع

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

وحدة الحل عبر CaptchaAI

تعزل هذه الوحدة كل تعامل مع CaptchaAI في مكان واحد: دالة submit_task ترسل المهمة إلى نقطة in.php وتعيد معرّف المهمة، ودالة poll_result تتولّى الاستطلاع الدوري عبر res.php حتى يجهز الرمز. تُبنى بقية الدوال فوق هاتين الأساسيتين، دالة لكل نوع CAPTCHA مع مهلة انتظار أولية مناسبة له.

# 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

يكشف التطبيق نقطة نهاية REST لكل نوع، ويعتمد على نماذج Pydantic للتحقق من صحة بيانات الطلب. عند فشل الحل يعيد رمز الحالة 502 مع رسالة الخطأ في الحقل detail، بينما تتيح نقطة /health لأنظمة المراقبة التأكد من أن الخدمة قيد التشغيل.

# 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

أمثلة عملية على الاستخدام

بعد تشغيل الخدمة محلياً، أرسِل طلباً إلى نقطة النهاية المناسبة لنوع CAPTCHA واستقبِل الرمز المحلول في الاستجابة.

حل 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..."
}

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

متى تستحق الخدمة المصغّرة عناء بنائها؟

الخدمة المستقلة ليست الخيار الأمثل دائماً. احسم الأمر بناءً على حجم فريقك وعدد الأنظمة التي تشترك في الحاجة إلى الحل:

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

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

تغطي الأمثلة أعلاه reCAPTCHA v2 وv3 وCloudflare Turnstile وصور CAPTCHA، ويمكنك إضافة نقاط نهاية أخرى بالنمط نفسه. يدعم CaptchaAI أنواع reCAPTCHA بالكامل وCloudflare Turnstile وChallenge وGeeTest v3 وصور OCR وGrid وBLS بشكل عام، إضافة إلى CaptchaFox وFriendly Captcha وLemin في مرحلة تجريبية (beta). أما hCaptcha وFunCaptcha (Arkose Labs) وGeeTest v4 فغير مدعومة حالياً، لذا لا تُنشئ لها نقاط نهاية تتوقّع حلاً.

كيف أربط عدد الطلبات المتزامنة بخطة CaptchaAI؟

يعتمد تسعير CaptchaAI على عدد الـ Threads المتزامنة لا على عدد عمليات الحل، وكل خطة تمنح عدداً محدداً من الـ Threads مع عمليات حل غير محدودة خلال الشهر. تبدأ خطة BASIC من 15$ شهرياً بـ 5 Threads، وترتفع حتى VIP-3 بسعر 7,500$ شهرياً و5,000 Thread. اجعل الحد الأقصى للطلبات المتزامنة داخل خدمتك متوافقاً مع عدد الـ Threads في خطتك حتى لا تصطدم بحد التزامن.

هل أحتاج إلى Redis أو قائمة انتظار لتوسيع الخدمة؟

للأحمال المتوسطة يكفي النموذج غير المتزامن في FastAPI وحده. عند الحاجة إلى معالجة دُفعية كبيرة أو فصل الإرسال عن الاستطلاع، أضِف قائمة انتظار مثل Redis أو عامل خلفية يستقبل المهام ويعيد النتائج لاحقاً عبر Callback أو نقطة استعلام مستقلة.

كيف أؤمّن مفتاح الـ API وأقيّد الوصول إلى الخدمة؟

لا تضع المفتاح داخل الكود كما في المثال التوضيحي؛ اقرأه من متغيّر بيئة أو من مدير أسرار. وقيّد الوصول إلى الخدمة نفسها بمصادقة داخلية عبر حقن التبعيات في FastAPI أو OAuth2، حتى لا تستدعيها إلا الأنظمة المصرّح لها داخل شبكتك.

كيف أنشر الخدمة في حاوية Docker مع حدّ لمعدل الطلبات؟

أضِف Dockerfile يبدأ بـ FROM python:3.11-slim، ثبّت التبعيات، واكشف المنفذ 8000. ولتحديد معدل الطلبات لكل عميل استخدم مكتبة slowapi أو خادماً وسيطاً عكسياً مثل nginx أو Traefik أمام الخدمة.

بمركزة منطق الحل ومراقبته في خدمة واحدة تحصل على طبقة داخلية مستقرة تستدعيها بقية الأنظمة، بدل تكرار التكامل وصيانته في كل مشروع على حدة.

أدلة ذات صلة

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