حل CAPTCHA بدون خوادم باستخدام AWS Lambda وCaptchaAI

لتشغيل حل اختبار CAPTCHA دون إدارة أي خادم تحتاج إلى دالة واحدة على AWS Lambda: ترسل التحدي إلى CaptchaAI، وتستطلع النتيجة حتى يجهز الرمز، ثم تعيده إلى تطبيقك — وتدفع مقابل مدة التنفيذ فقط. لا خادم يعمل في وقت الخمول، ولا سعة ثابتة تدفع ثمنها بلا استخدام، وتوسّع تلقائي يصل إلى آلاف عمليات الحل المتزامنة عند ارتفاع الحِمل. في هذا الدليل نبني هذا المسار خطوة بخطوة: المعالج، وتأمين المفتاح، وقالب البنية التحتية، والنشر، ثم المعالجة الدُفعية عبر SQS.

متى يكون التشغيل بدون خادم هو الخيار الأنسب؟

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

العوامل التي ترجّح هذا النهج:

• أحمال متقطعة أو غير متوقّعة لا تبرّر تشغيل خادم دائم.
• حاجة إلى توسّع سريع من صفر إلى آلاف الطلبات المتزامنة.
• رغبة في عزل منطق الحل داخل وحدة صغيرة يسهل ربطها ببوابة API أو SQS أو Step Functions.
• تفضيل تكلفة تتبع الاستخدام الفعلي بدلًا من سعة ثابتة مدفوعة مسبقًا.

في المقابل، إن كان لديك تدفّق ثابت وكثيف على مدار اليوم، فقد يكون تشغيل خدمة دائمة أوفر؛ أمّا الأحمال المتغيّرة فهي المجال الطبيعي لـ Lambda.

دالة Lambda: المعالج الأساسي

يقرأ المعالج مفتاح الـ API من متغيّر بيئة، ثم يحلّل جسم الطلب لاستخراج نوع التحدي (method) ومعاملاته (params). وتتولّى الدالة المساعدة solve_captcha منطق الحل: ترسل المهمة إلى نقطة النهاية in.php، وتلتقط معرّف المهمة، ثم تستطلع res.php كل خمس ثوانٍ حتى تجهز الاستجابة أو تنقضي المهلة. القيمة الافتراضية لـ method هي userrecaptcha الخاصة بـ reCAPTCHA، ويمكنك تمرير أي نوع مدعوم آخر — مثل turnstile لـ Cloudflare Turnstile — عبر جسم الطلب.

# lambda_function.py
import json
import os
import time
import urllib.request
import urllib.parse


def lambda_handler(event, context):
    """AWS Lambda handler for CaptchaAI solving."""
    api_key = os.environ["CAPTCHAAI_KEY"]

    # Parse input
    body = json.loads(event.get("body", "{}")) if isinstance(event.get("body"), str) else event

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

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


def solve_captcha(api_key, method, params, timeout=90):
    """Solve CAPTCHA using CaptchaAI API."""
    # Submit task
    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 for result
    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 المدمجة في Python بدلًا من مكتبة خارجية، ما يوفّر عليك إدارة طبقات Lambda ويُبقي حجم الحزمة صغيرًا وزمن البدء البارد قصيرًا.

تأمين مفتاح الـ API عبر Secrets Manager

لا تضع مفتاح الـ API نصًّا صريحًا داخل الشيفرة أو في مستودع Git. خزّنه في AWS Secrets Manager واقرأه عند التشغيل، لتفصل بيانات الاعتماد عن الكود وتتحكّم في صلاحيات الوصول عبر سياسات IAM.

import json
import boto3


def get_api_key():
    """Retrieve CaptchaAI key from AWS Secrets Manager."""
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="captchaai/api-key")
    secret = json.loads(response["SecretString"])
    return secret["api_key"]

لتخزين السر أول مرة:

aws secretsmanager create-secret \
  --name captchaai/api-key \
  --secret-string '{"api_key":"YOUR_API_KEY"}'

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

البنية التحتية كرمز: قالب SAM

عرّف كل شيء بصيغة تصريحية بدلًا من الضبط اليدوي في لوحة التحكم. يضبط قالب SAM التالي المهلة عند 120 ثانية، والذاكرة عند 256 ميجابايت، وبيئة تشغيل python3.11، ويربط الدالة بحدث بوابة API على المسار /solve، ويمنحها سياسة قراءة السر من Secrets Manager.

# template.yaml
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31

Globals:
  Function:
    Timeout: 120
    MemorySize: 256
    Runtime: python3.11

Resources:
  CaptchaSolverFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambda_function.lambda_handler
      Environment:
        Variables:
          CAPTCHAAI_KEY: !Sub "{{resolve:secretsmanager:captchaai/api-key:SecretString:api_key}}"
      Events:
        SolveApi:
          Type: Api
          Properties:
            Path: /solve
            Method: post
      Policies:

        - AWSSecretsManagerGetSecretValuePolicy:
            SecretArn: !Sub "arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:captchaai/api-key-*"

Outputs:
  SolveApiUrl:
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/solve"

يقرأ التعبير resolve:secretsmanager قيمة المفتاح عند النشر ويحقنها في متغيّر البيئة CAPTCHAAI_KEY، فتبقى بيانات الاعتماد مركزية وقابلة للتدوير دون تعديل الشيفرة.

النشر والاختبار

بعد تجهيز القالب، ابنِ الحزمة وانشرها بأمر واحد، ثم اختبر نقطة النهاية بطلب POST مباشر:

# Build and deploy
sam build
sam deploy --guided

# Test
curl -X POST https://YOUR_API_ID.execute-api.us-east-1.amazonaws.com/Prod/solve \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

يسألك الأمر sam deploy --guided عن اسم المكدّس والمنطقة والأذونات في المرة الأولى فقط، ثم يحفظ الإعدادات لعمليات النشر اللاحقة. استبدل SITE_KEY وpageurl بقيم الصفحة المستهدفة الفعلية عند الاختبار.

المعالجة الدُفعية عبر SQS

عند الحجم الكبير، افصل الإنتاج عن الاستهلاك عبر قائمة انتظار SQS: يدفع المنتِجون المهام إلى القائمة، وتسحبها Lambda على دفعات وتحلّها بالتوازي. هنا يظهر التفاعل المهم بين تزامن Lambda وخطط CaptchaAI المبنية على الـ Threads؛ إذ يحدّد عدد الـ Threads في خطتك كم عملية حل يمكن أن تجري في آنٍ واحد. فخطة ADVANCE بسعر 90 دولارًا شهريًا تمنحك 50 Thread — أي ما يوازي 50 عملية حل متزامنة — بينما ترفعها خطة PREMIUM بسعر 170 دولارًا إلى 100 Thread. اضبط حدّ التزامن في Lambda بما يتناسب مع عدد الـ Threads المتاحة حتى لا تصطدم الطلبات الزائدة بسقف خطتك.

import json
import os
import time
import urllib.request
import urllib.parse


def sqs_handler(event, context):
    """Process CAPTCHA tasks from SQS queue."""
    api_key = os.environ["CAPTCHAAI_KEY"]
    results = []

    for record in event["Records"]:
        task = json.loads(record["body"])
        try:
            token = solve_captcha(
                api_key,
                task["method"],
                task["params"],
            )
            results.append({
                "task_id": task.get("id"),
                "status": "success",
                "token": token[:50],
            })
        except Exception as e:
            results.append({
                "task_id": task.get("id"),
                "status": "error",
                "error": str(e),
            })

    return {"results": results}

يعالج هذا المعالج كل سجلّ في الدفعة على حدة، ويعيد نتيجة مستقلة لكل مهمة، فلا يُسقط فشلُ عنصر واحد بقية الدفعة.

ضبط إعدادات Lambda لأحمال CAPTCHA

اضبط الإعدادات التالية بما يتناسب مع طبيعة أحمال الحل لديك:

معالجة المشكلات الشائعة

أكثر الأعطال شيوعًا وحلولها المباشرة:

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

كيف يتفاعل تزامن Lambda مع خطط CaptchaAI المبنية على الـ Threads؟

عدد الـ Threads في خطتك هو سقف عمليات الحل المتزامنة، وتزامن Lambda هو سقف الاستدعاءات المتزامنة. اجعلهما متقاربين: إن أتاحت خطتك 50 Thread فلا فائدة من السماح بمئات الاستدعاءات المتوازية، لأنها ستنتظر دورها على الخدمة. وازِن بين حدّ التزامن في Lambda وعدد الـ Threads لتجنّب الاختناق.

هل يرفع الانتظار أثناء الاستطلاع الدوري فاتورة Lambda؟

نعم، لأنك تُحاسَب على مدة التنفيذ كاملةً بما فيها فترات time.sleep بين محاولات الاستطلاع. لتخفيض ذلك عند الحجم الكبير، انقل الحل إلى مسار SQS غير متزامن، أو زِد الفاصل بين الاستطلاعات قليلًا، بدل إبقاء الدالة معلّقة تنتظر رمزًا لم يجهز بعد.

أستخدم بوابة API أم SQS لتشغيل الدالة؟

اختر بوابة API عندما تحتاج استجابة فورية متزامنة ضمن طلب المستخدم. واختر SQS للأحمال الدُفعية أو الخلفية التي تتحمّل بعض التأخير وتستفيد من التوسّع التلقائي مع عمق القائمة والمعالجة المرنة عند الفشل.

ما المنطقة (Region) الأنسب لنشر الدالة؟

انشر الدالة في منطقة قريبة من الصفحات المستهدفة ومن CaptchaAI لتقليل زمن الرحلة الشبكية. إن كان جمهورك أو مواقعك في أوروبا فمنطقة أوروبية أنسب من us-east-1، إذ يقلّص القرب بضع عشرات من المللي ثانية لكل طلب.

أدلة ذات صلة

• التعامل مع CAPTCHA داخل روبوتات الأتمتة
• أتمتة تسجيل الدخول مع حل CAPTCHA

انطلق بلا خوادم — احصل على مفتاح CaptchaAI اليوم.