وظائف جوجل السحابية + تكامل CaptchaAI

توفر Google Cloud Functions حل اختبار CAPTCHA بدون خادم من خلال التوسع التلقائي، وفواتير الدفع لكل استخدام، والتكامل المحكم للنظام البيئي لبرنامج Google Cloud Platform.

وظيفة تشغيل HTTP

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

المتطلبات

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

نشر

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

معالجة الدُفعات Pub/Sub-Triggered

معالجة مهام اختبار CAPTCHA من موضوع Pub/Sub:

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

النشر في Pub/Sub:

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

إرسال المهام إلى 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}")

مقارنة التكلفة

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

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

وظائف السحابة Gen1 أو Gen2؟

استخدم Gen2. وهو يدعم مهلات أطول (تصل إلى 60 دقيقة)، والمزيد من الذاكرة، والتزامن - وكلها مفيدة لحل اختبار CAPTCHA.

كيف أتعامل مع المصادقة؟

للاستخدام الداخلي، يتطلب المصادقة مع --no-allow-unauthenticated. بالنسبة لواجهات برمجة التطبيقات الخارجية، استخدم بوابة API مع مفاتيح API أمام الوظيفة.

هل يمكنني إبقاء الوظيفة دافئة؟

استخدم Cloud Scholer لإجراء اختبار اتصال الوظيفة كل 5 دقائق. أو قم بتعيين --min-instances=1 للحفاظ على دفء مثيل واحد (التكلفة ~ 7/month).

أدلة ذات صلة

• حل CAPTCHA بدون خوادم باستخدام AWS Lambda وCaptchaAI
• Azure Functions + CaptchaAI: تكامل سحابي بدون خوادم

بدون خادم على Google Cloud Platform — احصل على مفتاح CaptchaAI الخاص بك اليوم.