تشغيل CaptchaAI على Google Cloud Functions بدون خادم

يختصر تشغيل CaptchaAI على Google Cloud Functions البنية التحتية كلها في دالة واحدة: تقرأ مفتاح الـ API من Secret Manager، وتُرسل المهمة إلى CaptchaAI عبر in.php، وتستطلع res.php حتى يجهز التوكن، ثم تعيده إلى المُستدعي. المحاسبة على زمن التنفيذ وحده، والتوسّع متروك للمنصّة تتولّاه تلقائياً حين يقفز الحِمل، فلا سعة ثابتة تدفع ثمنها في ساعات الخمول. يبني هذا الدليل المسار خطوة بخطوة: دالة HTTP أولاً، ثم معالجة دُفعية عبر Pub/Sub، مع ضبط التكلفة والمراقبة.

لماذا تناسب الدوال السحابية أحمال الكابتشا؟

حل الكابتشا حِمل متقطّع بطبيعته: قد ترسل مئات المهام في نافذة قصيرة أثناء جولة استخراج بيانات، ثم لا شيء لساعات. الخادم الدائم يظل يستهلك تكلفته في الحالتين، بينما تتقلّص الدالة السحابية إلى صفر عند غياب الطلبات وتتمدّد فوراً عند وصولها. تناسبك هذه البنية تحديداً حين:

• يكون الطلب متذبذباً ولا يبرّر تشغيل خادم على مدار الساعة.
• تريد فوترة مرتبطة بالاستخدام الفعلي لا بزمن التشغيل الكامل.
• تفضّل ترك التوسّع للمنصّة بدل إدارته يدوياً عند ذروة الحِمل.

نقطة يغفل عنها كثيرون: توسّع الدوال لا يعني توسّعاً بلا حدود في الحل. فوترة CaptchaAI تقوم على الـ Threads (المهام المتزامنة) لا على عدد عمليات الحل، وكل خطة تتيح عدداً محدداً من الـ Threads مع عمليات حل غير محدودة داخل الشهر. فإن رفعت --max-instances إلى 100 بينما تشغّل خطة BASIC ($15 شهرياً، 5 Threads)، فلن تتجاوز خمس مهام متزامنة فعلياً؛ والفائض ينتظر دوره. لرفع سقف التزامن انتقل إلى خطة أعلى مثل ADVANCE ($90 شهرياً، 50 Thread). لذا اضبط --max-instances بما يوازي عدد الـ Threads المتاح لك، لا أكثر.

الدالة المُشغَّلة عبر HTTP

نقطة الدخول دالة واحدة تقرأ جسم JSON، وتستخرج method وparams، وتجلب مفتاح الـ API من Secret Manager، ثم تُسند الحل إلى دالة مساعدة تُرسل المهمة عبر in.php وتستطلع res.php حتى تجهز الاستجابة. المفتاح لا يظهر في الشيفرة ولا في متغيرات البيئة، بل يُقرأ من مخزن الأسرار عند كل استدعاء:

# main.py
import json
import time
import urllib.request
import urllib.parse
import functions_framework


@functions_framework.http
def solve_captcha(request):
    """HTTP Cloud Function for CAPTCHA solving."""
    # Parse request
    request_json = request.get_json(silent=True)
    if not request_json:
        return json.dumps({"error": "JSON body required"}), 400

    method = request_json.get("method", "userrecaptcha")
    params = request_json.get("params", {})

    # Get API key from Secret Manager
    api_key = _get_secret("captchaai-key")

    try:
        token = _solve(api_key, method, params)
        return json.dumps({"token": token})
    except Exception as e:
        return json.dumps({"error": str(e)}), 500


def _get_secret(secret_id):
    """Get secret from GCP Secret Manager."""
    from google.cloud import secretmanager
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{_get_project_id()}/secrets/{secret_id}/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return response.payload.data.decode("UTF-8")


def _get_project_id():
    """Get current GCP project ID."""
    import urllib.request
    req = urllib.request.Request(
        "http://metadata.google.internal/computeMetadata/v1/project/project-id",
        headers={"Metadata-Flavor": "Google"},
    )
    with urllib.request.urlopen(req) as resp:
        return resp.read().decode()


def _solve(api_key, method, params, timeout=90):
    """Solve CAPTCHA via CaptchaAI API."""
    # Submit
    submit_data = urllib.parse.urlencode({
        "key": api_key,
        "method": method,
        "json": 1,
        **params,
    }).encode()

    req = urllib.request.Request(
        "https://ocr.captchaai.com/in.php",
        data=submit_data,
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read())

    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")

    task_id = result["request"]

    # Poll
    start = time.time()
    while time.time() - start < timeout:
        time.sleep(5)
        poll_url = (
            f"https://ocr.captchaai.com/res.php"
            f"?key={api_key}&action=get&id={task_id}&json=1"
        )
        with urllib.request.urlopen(poll_url, timeout=15) as resp:
            data = json.loads(resp.read())

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

    raise TimeoutError("Solve timeout")

ملف المتطلبات

نكتفي بمكتبتين. لاحظ اعتمادنا على urllib من المكتبة القياسية بدل requests؛ فكل تبعية إضافية تُثقل حزمة النشر وتُطيل زمن البدء البارد، وهو ما نحرص على تقليصه في البيئات بدون خادم:

# requirements.txt
functions-framework==3.*
google-cloud-secret-manager==2.*

نشر الدالة

قبل النشر خزّن المفتاح في Secret Manager، ثم انشر الدالة من نوع Gen2 مع مهلة كافية (120 ثانية) وحدٍّ أقصى للنسخ يوازي عدد الـ Threads لديك. العَلَم --allow-unauthenticated يفتح الدالة للعموم؛ فإن كان الاستدعاء داخلياً فاستبدله بـ --no-allow-unauthenticated كما نوضح في الأسئلة الشائعة:

# Create secret
echo -n "YOUR_API_KEY" | gcloud secrets create captchaai-key --data-file=-

# Deploy function
gcloud functions deploy solve-captcha \
  --gen2 \
  --runtime=python311 \
  --region=us-central1 \
  --source=. \
  --entry-point=solve_captcha \
  --trigger-http \
  --allow-unauthenticated \
  --timeout=120s \
  --memory=256MB \
  --max-instances=100

# Test
curl -X POST https://us-central1-PROJECT.cloudfunctions.net/solve-captcha \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

يمنح دور IAM المسمّى secretmanager.secretAccessor للدالة صلاحية قراءة السر — من دونه سيفشل الاستدعاء برفض الإذن.

المعالجة الدُفعية عبر Pub/Sub

يصلح المسار عبر HTTP للطلبات الفورية، لكن الأحمال الكبيرة — كطابور من آلاف الصفحات — تُدار بأمان أكبر عبر Pub/Sub. يسير التدفّق في ثلاث مراحل:

1. تنشر كل مهمة كرسالة في موضوع المهام.
2. تلتقطها دالة مُشغَّلة بالحدث فتحلّ الكابتشا عبر CaptchaAI.
3. تنشر الدالة التوكن الناتج في موضوع النتائج ليستهلكه نظامك.

هكذا تحصل على فصل واضح بين الإنتاج والاستهلاك وعلى إعادة محاولة تلقائية عند الفشل:

import base64
import json
import functions_framework
from google.cloud import pubsub_v1


@functions_framework.cloud_event
def process_captcha_task(cloud_event):
    """Process CAPTCHA task from Pub/Sub message."""
    data = base64.b64decode(cloud_event.data["message"]["data"])
    task = json.loads(data)

    api_key = _get_secret("captchaai-key")

    try:
        token = _solve(api_key, task["method"], task["params"])
        # Publish result
        publisher = pubsub_v1.PublisherClient()
        topic = f"projects/{_get_project_id()}/topics/captcha-results"
        publisher.publish(topic, json.dumps({
            "task_id": task["id"],
            "status": "success",
            "token": token,
        }).encode())

    except Exception as e:
        print(f"Task {task.get('id')} failed: {e}")

ثم انشر دالة المعالجة مربوطةً بالموضوع مباشرة:

gcloud functions deploy process-captcha-task \
  --gen2 \
  --runtime=python311 \
  --trigger-topic=captcha-tasks \
  --timeout=120s \
  --memory=256MB

إرسال المهام إلى قائمة Pub/Sub

من أي خدمة منتِجة، انشر المهام دفعةً واحدة في الموضوع؛ يوزّعها Pub/Sub على نسخ الدالة المتاحة ضمن حدّ التزامن الذي ضبطته:

from google.cloud import pubsub_v1
import json

publisher = pubsub_v1.PublisherClient()
topic = "projects/YOUR_PROJECT/topics/captcha-tasks"

# Submit batch
urls = ["https://site1.com", "https://site2.com", "https://site3.com"]
for i, url in enumerate(urls):
    task = {
        "id": f"task-{i}",
        "method": "userrecaptcha",
        "params": {"googlekey": "SITE_KEY", "pageurl": url},
    }
    publisher.publish(topic, json.dumps(task).encode())
    print(f"Published task-{i}")

مقارنة التكلفة: دالة سحابية مقابل خادم دائم

تُظهر الأرقام التقديرية أدناه أين تتفوّق الدالة السحابية: عند الأحجام المنخفضة والمتقطّعة تكاد تكلفتها تكون صفراً، ولا تقترب من تكلفة الخادم الدائم إلا حين يصبح الحِمل مستمراً على مدار اليوم. الأرقام تقريبية وتختلف بحسب المنطقة ونمط الاستخدام.

عند الحِمل الثابت المرتفع قد يصبح الخادم الدائم أوفر؛ ونقطة التعادل هنا قرب 10,000 عملية حل يومياً.

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

أغلب مشكلات هذا الإعداد تتكرّر، وهذه أسرعها حلاً:

أسئلة شائعة

كم عدد الـ Threads التي يحتاجها هذا التكامل؟

بقدر التزامن الذي تريده. بما أن CaptchaAI يفوتر على الـ Threads لا على عدد عمليات الحل، حدّد --max-instances بما يوازي حصّتك: خطة BASIC ($15 شهرياً) تمنح 5 Threads، وADVANCE ($90 شهرياً) تمنح 50، مع عمليات حل غير محدودة داخل كل Thread. وضبط حدّ النسخ أعلى من عدد الـ Threads يزيد الانتظار لا الإنتاجية.

هل يقتصر الإعداد على reCAPTCHA أم يدعم أنواعاً أخرى؟

يكفي تغيير قيمة method في الطلب. يدعم CaptchaAI reCAPTCHA v2/v3 عبر userrecaptcha، وCloudflare Turnstile عبر turnstile، وCloudflare Challenge عبر cloudflare_challenge، وGeeTest v3 عبر geetest، إضافةً إلى الصور/OCR والشبكات وBLS. أما hCaptcha وFunCaptcha (Arkose Labs) فغير مدعومة حالياً، وGeeTest v4 قيد الإعداد ولم تُتَح بعد.

هل أستخدم Gen1 أم Gen2؟

اختر Gen2؛ فهو يتيح مهلات أطول (حتى 60 دقيقة) وذاكرة أكبر وتزامناً داخل النسخة الواحدة — وكلها مفيدة لانتظار حلّ الكابتشا الذي قد يمتد عشرات الثواني.

كيف أمنع البدء البارد من إبطاء أول استدعاء؟

أبقِ نسخة واحدة دافئة عبر --min-instances=1 (بتكلفة ~7$ شهرياً)، أو استخدم Cloud Scheduler لإرسال نبضة إلى الدالة كل خمس دقائق. وتقليل التبعيات — كاستخدام urllib بدل requests — يخفض زمن البدء البارد من الأساس.

أستطلع النتيجة داخل الدالة أم أعتمد على رد النداء؟

داخل دالة قصيرة العمر، يبقى الاستطلاع الدوري لـ res.php النمط الأبسط والأكثر قابلية للتنبؤ، وهو ما تفعله الشيفرة أعلاه. أما إن أردت تحرير النسخة بدل إبقائها منتظرة، فافصل الإرسال عن الاستلام عبر Pub/Sub: دالة تُرسل المهمة، وأخرى تلتقط النتيجة لاحقاً.

أدلة ذات صلة

• حل الكابتشا بدون خادم عبر AWS Lambda
• تكامل CaptchaAI مع Azure Functions بدون خادم

بدون خادم على Google Cloud Platform — أنشئ مفتاح CaptchaAI وانشر أول دالة حل اليوم.