معالجة نتائج CAPTCHA بنمط حدثي باستخدام AWS SNS وCaptchaAI

بدل أن ينتظر السكربت لديك اكتمال كل اختبار CAPTCHA عبر الاستطلاع الدوري، اجعل CaptchaAI يدفع النتيجة إلى AWS SNS لحظة جهوزها، فتلتقطها قوائم SQS ووظائف Lambda وتنبيهات البريد كلٌّ على حِدة. النتيجة نظام مفكوك الارتباط: من يرسل الطلب لا يحتاج أن يعرف من يستهلك النتيجة، وكل طرف يتوسّع بمعزل عن الآخر.

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

متى يتفوّق النمط الحدثي على الاستطلاع الدوري

يعتمد الاستطلاع الدوري على مناداة نقطة res.php بشكل متكرر حتى تجهز النتيجة. يعمل هذا جيدًا مع طلب واحد، لكن تظهر ثغراته عند التوسّع:

• كل عملية استطلاع تُبقي خيط معالجة مشغولًا في الانتظار بدل تنفيذ عمل مفيد.
• يرتبط منطق الاستهلاك مباشرة بمنطق الإرسال، فيصعب إضافة مستهلك جديد دون تعديل الكود.
• تتضخّم كلفة زمن الانتظار حين تتفاوت أوقات الحل بين نوع CAPTCHA وآخر.

مع النمط الحدثي تختفي هذه القيود: يرسل CaptchaAI النتيجة مرة واحدة إلى نقطة رد النداء لديك، وتتكفّل SNS بتوزيعها على كل من يعنيه الأمر.

نظرة عامة على البنية

تمرّ كل نتيجة عبر أربع محطات: الإرسال إلى CaptchaAI، ثم رد النداء إلى بوابة API مع Lambda، ثم النشر إلى موضوع SNS، وأخيرًا التوزيع على المستهلكين.

[Scraper] → Submit CAPTCHA → [CaptchaAI API]
                                    ↓
                            Solve completes
                                    ↓
                            Callback → [API Gateway + Lambda]
                                    ↓
                            Publish → [SNS Topic]
                                    ↓
                    ┌───────────────┼───────────────┐
                    ↓               ↓               ↓
            [SQS Queue]      [Lambda Logger]   [Email Alert]
            (result store)   (audit trail)     (on failure)

توفّر SNS توزيعًا واسع النطاق: نتيجة CAPTCHA واحدة تُشغّل عدة مستهلكين دون أن يعلم معالج رد النداء بوجودهم. المكوّنات الأساسية في هذا التدفّق:

• موضوع SNS — نقطة النشر المركزية التي تستقبل كل نتيجة حل.
• قائمة SQS — تخزّن النتائج ليقرأها السكربت لاحقًا بوتيرته الخاصة.
• وظيفة Lambda للتدقيق — تسجّل مسار التحقق للمراجعة والمساءلة.
• تنبيه بريد إلكتروني — يُطلق عند الفشل فقط عبر سياسة تصفية.

الخطوة 1: أنشئ موضوع SNS

يبدأ كل شيء من موضوع SNS واحد يجمع نتائج الحل. أمامك طريقتان لإنشائه بحسب سياق عملك:

1. سطر أوامر AWS CLI للإعداد السريع أو ضمن النصوص التشغيلية.
2. مكتبة boto3 حين تريد إنشاء الموضوع من داخل كود التطبيق نفسه.

AWS CLI

aws sns create-topic --name captcha-results --output text
# Returns: arn:aws:sns:us-east-1:123456789:captcha-results

Python (boto3)

import boto3

sns = boto3.client("sns", region_name="us-east-1")

response = sns.create_topic(Name="captcha-results")
topic_arn = response["TopicArn"]
print(f"Topic ARN: {topic_arn}")

الخطوة 2: ابنِ مستقبِل رد النداء

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

ملاحظة: لا تضع منطق معالجة النتائج داخل هذه الوظيفة. مهمتها الوحيدة استقبال الحل ونشره على SNS، بينما يتولّى المستهلكون بقية العمل.

Python (معالج Lambda)

import json
import os
import boto3

sns = boto3.client("sns")
TOPIC_ARN = os.environ["SNS_TOPIC_ARN"]


def lambda_handler(event, context):
    """Receive CaptchaAI callback and publish to SNS."""
    # Parse query parameters from API Gateway
    params = event.get("queryStringParameters", {}) or {}
    task_id = params.get("id", "")
    solution = params.get("code", "")

    if not task_id or not solution:
        return {"statusCode": 400, "body": "Missing id or code"}

    # Publish to SNS
    message = {
        "task_id": task_id,
        "solution": solution,
        "status": "solved"
    }

    sns.publish(
        TopicArn=TOPIC_ARN,
        Message=json.dumps(message),
        Subject="captcha-solved",
        MessageAttributes={
            "task_id": {
                "DataType": "String",
                "StringValue": task_id
            }
        }
    )

    return {"statusCode": 200, "body": "OK"}

JavaScript (معالج Lambda)

const { SNSClient, PublishCommand } = require("@aws-sdk/client-sns");

const sns = new SNSClient({ region: "us-east-1" });
const TOPIC_ARN = process.env.SNS_TOPIC_ARN;

exports.handler = async (event) => {
  const params = event.queryStringParameters || {};
  const taskId = params.id;
  const solution = params.code;

  if (!taskId || !solution) {
    return { statusCode: 400, body: "Missing id or code" };
  }

  const message = {
    task_id: taskId,
    solution: solution,
    status: "solved",
  };

  await sns.send(
    new PublishCommand({
      TopicArn: TOPIC_ARN,
      Message: JSON.stringify(message),
      Subject: "captcha-solved",
      MessageAttributes: {
        task_id: { DataType: "String", StringValue: taskId },
      },
    })
  );

  return { statusCode: 200, body: "OK" };
};

الخطوة 3: أرسل اختبارات CAPTCHA مع عنوان رد النداء

وجّه معامل pingback في طلب CaptchaAI إلى نقطة نهاية بوابة API لديك. بذلك يُدفَع الحل تلقائيًا إلى Lambda فور جهوزه بدل أن تستطلعه بنفسك:

Python

import os
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]
CALLBACK_URL = os.environ["CALLBACK_GATEWAY_URL"]  # API Gateway URL


def submit_captcha(sitekey, pageurl):
    """Submit CAPTCHA with SNS-backed callback."""
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "pingback": CALLBACK_URL,
        "json": 1
    })
    data = resp.json()

    if data.get("status") == 1:
        return data["request"]  # task_id
    raise RuntimeError(f"Submit failed: {data.get('request')}")

الخطوة 4: اشترك المستهلكين في الموضوع

بعد جهوز الموضوع، اربط به المستهلكين. كل اشتراك يستقبل النتيجة نفسها بشكل مستقل عن البقية. سنربط ثلاثة أنواع:

• قائمة SQS لتخزين كل النتائج وقراءتها لاحقًا.
• وظيفة Lambda لتسجيل مسار التدقيق.
• بريد إلكتروني للتنبيه عند الفشل فقط.

قائمة SQS (تخزين النتائج)

# Subscribe an SQS queue to receive all results
sqs_arn = "arn:aws:sqs:us-east-1:123456789:captcha-results-queue"

sns.subscribe(
    TopicArn=topic_arn,
    Protocol="sqs",
    Endpoint=sqs_arn
)

Lambda (سجل التدقيق)

# Subscribe a Lambda for audit logging
lambda_arn = "arn:aws:lambda:us-east-1:123456789:function:captcha-audit-logger"

sns.subscribe(
    TopicArn=topic_arn,
    Protocol="lambda",
    Endpoint=lambda_arn
)

البريد الإلكتروني (تنبيهات الفشل)

# Subscribe email for error notifications with filter
sns.subscribe(
    TopicArn=topic_arn,
    Protocol="email",
    Endpoint="ops@example.com"
)

الخطوة 5: استهلك النتائج من SQS

يقرأ السكربت الحلول من SQS بالاستطلاع الطويل بدل مناداة CaptchaAI مباشرة. لاحظ أن SNS تغلّف الرسالة داخل مظروف، لذا نفكّها على مستويين قبل استخدام النتيجة:

Python

import json
import boto3

sqs = boto3.client("sqs", region_name="us-east-1")
QUEUE_URL = os.environ["SQS_QUEUE_URL"]


def get_solved_captcha(timeout=30):
    """Wait for a CAPTCHA solution from the SQS queue."""
    response = sqs.receive_message(
        QueueUrl=QUEUE_URL,
        MaxNumberOfMessages=1,
        WaitTimeSeconds=min(timeout, 20)  # Long polling (max 20s)
    )

    messages = response.get("Messages", [])
    if not messages:
        return None

    msg = messages[0]
    # SNS wraps the message — unwrap it
    sns_envelope = json.loads(msg["Body"])
    result = json.loads(sns_envelope["Message"])

    # Delete message after processing
    sqs.delete_message(
        QueueUrl=QUEUE_URL,
        ReceiptHandle=msg["ReceiptHandle"]
    )

    return result

JavaScript

const {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
} = require("@aws-sdk/client-sqs");

const sqs = new SQSClient({ region: "us-east-1" });
const QUEUE_URL = process.env.SQS_QUEUE_URL;

async function getSolvedCaptcha(timeout = 30) {
  const response = await sqs.send(
    new ReceiveMessageCommand({
      QueueUrl: QUEUE_URL,
      MaxNumberOfMessages: 1,
      WaitTimeSeconds: Math.min(timeout, 20),
    })
  );

  const messages = response.Messages || [];
  if (messages.length === 0) return null;

  const msg = messages[0];
  const snsEnvelope = JSON.parse(msg.Body);
  const result = JSON.parse(snsEnvelope.Message);

  await sqs.send(
    new DeleteMessageCommand({
      QueueUrl: QUEUE_URL,
      ReceiptHandle: msg.ReceiptHandle,
    })
  );

  return result;
}

توجيه النتائج عبر تصفية رسائل SNS

لا حاجة لأن يستقبل كل مستهلك كل نتيجة. تتيح سياسة التصفية (FilterPolicy) توجيه حالات محددة فقط — مثل الأخطاء — إلى قائمة مخصّصة للعمليات:

# Only send failures to the ops queue
sns.subscribe(
    TopicArn=topic_arn,
    Protocol="sqs",
    Endpoint=failure_queue_arn,
    Attributes={
        "FilterPolicy": json.dumps({
            "status": ["failed", "error"]
        })
    }
)

سيناريو عملي: متجر إلكتروني في موسم التخفيضات

لنفترض متجرًا إلكترونيًا في منطقة الخليج يعالج آلاف عمليات التحقق يوميًا أثناء موجات الطلب في مواسم التخفيضات. مع الاستطلاع الدوري، كان كل ارتفاع مفاجئ في الطلب يستهلك خيوط المعالجة في الانتظار ويؤخّر إتمام الشراء. بعد الانتقال إلى النمط الحدثي، صار CaptchaAI يدفع كل نتيجة إلى SNS، فتُخزَّن في SQS ويقرأها فريق الواجهة الخلفية بوتيرته دون أن يتأثر مسار الشراء بذروة الطلب.

يتكامل هذا مع نموذج التسعير القائم على thread في CaptchaAI: تدفع مقابل عدد الخيوط المتزامنة لا مقابل كل عملية حل، ما يجعل الكلفة قابلة للتنبؤ حتى مع تذبذب الأحجام. على سبيل المثال، تتيح خطة ADVANCE بسعر 90 دولارًا شهريًا تشغيل 50 thread متزامنًا مع عدد غير محدود من عمليات الحل — وهو ما يكفي لاستيعاب ذروة الطلب دون رسوم على كل عملية.

ما الذي يكسبه هذا المتجر عمليًا من التحوّل إلى النمط الحدثي؟

• زمن استجابة أقصر عند إتمام الشراء لأن السكربت لم يعد ينتظر النتيجة.
• إضافة مستهلكين جدد للتحليلات أو التنبيهات دون لمس كود الإرسال.
• كلفة شهرية ثابتة يسهل التخطيط لها مهما ارتفعت أحجام الذروة.

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

أكثر المشكلات شيوعًا في هذا المسار تتعلق بالصلاحيات وتغليف الرسائل والتكرار. الجدول التالي يلخّص أبرزها مع الحل المناسب:

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

هل يستحق هذا النمط العناء مع الأحجام الصغيرة؟

مع بضع عمليات تحقق في الدقيقة، يظل الاستطلاع الدوري البسيط كافيًا وأقل تعقيدًا. تبرز قيمة SNS عندما ترتفع الأحجام أو تتعدّد الجهات التي تحتاج النتيجة نفسها؛ عندها يصنع فكّ الارتباط وتوسّع المستهلكين فرقًا حقيقيًا في الاعتمادية.

كيف أتعامل مع رد نداء لم يصل أو تأخّر؟

احتفظ بمسار احتياطي يستطلع نقطة res.php بعد مهلة محددة لكل مهمة لم تصلك نتيجتها عبر SNS. بهذا تجمع بين سرعة النمط الحدثي وأمان الاستطلاع كخط دفاع أخير، دون إغراق الخدمة بطلبات متكررة بلا داعٍ.

هل أحتاج بوابة API أم يمكن استدعاء Lambda مباشرة؟

يحتاج CaptchaAI إلى عنوان HTTP عام ليصله رد النداء، وبوابة API هي الطريقة المعتادة لكشف Lambda عبر HTTPS. يمكن بدلًا منها استخدام Lambda Function URL إذا أردت تبسيط الإعداد وتقليل عدد المكوّنات في البنية.

ما الفرق بين تصفية رسائل SNS واستخدام قوائم انتظار منفصلة؟

تُبقي تصفية SNS موضوعًا واحدًا وتوجّه كل مستهلك إلى ما يعنيه عبر FilterPolicy، فتقلّ الرسائل غير الضرورية. أما تخصيص قائمة لكل نوع فيمنحك عزلًا أوضح على حساب إدارة اشتراكات أكثر. الخلاصة العملية:

• تصفية SNS: أنسب حين تتشارك النتائج بنية واحدة وتختلف فقط في الحالة.
• قوائم منفصلة: أنسب حين تحتاج كل جهة إلى صلاحيات أو سياسات احتفاظ مختلفة.

اختر بحسب عدد الأنواع وحاجتك إلى العزل.

الخطوات التالية

ابنِ على هذا الأساس عبر أدلة التكامل العملية التالية:

• ابدأ سريعًا وحُلّ أول اختبار CAPTCHA في خمس دقائق
• حلّ reCAPTCHA v2 عبر الـ API خطوة بخطوة
• حلّ Cloudflare Turnstile عبر الـ API
• حلّ GeeTest v3 عبر الـ API

أدلة ذات صلة

• حل CAPTCHA بلا خوادم باستخدام AWS Lambda مع CaptchaAI
• التعامل مع CAPTCHA داخل روبوتات الأتمتة