Erstellen einer ereignisgesteuerten CAPTCHA-Lösung mit AWS SNS und CaptchaAI

Das Abfragen von CAPTCHA-Ergebnissen bindet Threads und sorgt für eine enge Kopplung zwischen Ihrem Scraper und der Lösungspipeline. AWS SNS (Simple Notification Service) entkoppelt diese Bedenken – CaptchaAI sendet Ergebnisse an Ihren Rückruf, der sie an SNS veröffentlicht, und eine beliebige Anzahl nachgeschalteter Verbraucher reagiert unabhängig.

Architekturübersicht

[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 bietet Fan-Out: Ein CAPTCHA-Ergebnis löst mehrere Verbraucher aus, ohne dass der Callback-Handler davon erfährt.

Schritt 1: Erstellen Sie das SNS-Thema

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

Schritt 2: Erstellen Sie den Rückrufempfänger

Diese Lambda-Funktion empfängt CaptchaAI-Callback-Ergebnisse und veröffentlicht sie an SNS.

Python (Lambda-Handler)

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-Handler)

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

Schritt 3: Senden Sie CAPTCHAs mit der Rückruf-URL

Richten Sie pingback von CaptchaAI auf Ihren API-Gateway-Endpunkt:

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

Schritt 4: Verbraucher abonnieren

SQS-Warteschlange (Ergebnisspeicherung)

# 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 (Audit Logger)

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

E-Mail (Fehlerbenachrichtigungen)

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

Schritt 5: Ergebnisse von SQS nutzen

Ihr Scraper liest Lösungen von SQS, anstatt CaptchaAI abzufragen:

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-Nachrichtenfilterung

Leiten Sie unterschiedliche Ergebnisse an unterschiedliche Verbraucher weiter:

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

Fehlerbehebung

Verwandte Leitfäden

• CaptchaAI Schnellstart
• API-Antwortformate und Fehlercodes
• reCAPTCHA v2 per API lösen