تتبّع حلول CAPTCHA في DynamoDB لتطبيقات بدون خادم

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

لماذا يناسب DynamoDB بنية بدون خادم

المشكلة الأولى التي تصطدم بها أي قاعدة بيانات علائقية داخل Lambda هي إدارة الاتصالات. كل استدعاء بارد يفتح اتصالاً جديداً، وسرعان ما تستنفد حدود الاتصال أو تضطر إلى إضافة طبقة تجميع مثل RDS Proxy بتكلفة وتعقيد إضافيين. أما DynamoDB فيتعامل عبر HTTP مع كل طلب على حدة، فلا يوجد اتصال دائم يجب الحفاظ عليه، وهو ما ينسجم تماماً مع طبيعة الدوال قصيرة العمر.

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

تخيّل فريق أتمتة في الرياض يشغّل عملياته على منطقة AWS الشرق الأوسط (me-central-1) ليقلّل زمن الوصول. مع DynamoDB يبقى الجدول قريباً من دوال Lambda نفسها، وترتفع سعة الكتابة والقراءة تلقائياً في ساعات الذروة دون تدخّل يدوي أو إعادة توفير. وبما أن CaptchaAI يحاسب على أساس عدد الـ Threads المتزامنة لا على كل عملية حل، فإن معدل الكتابة في DynamoDB يعكس مباشرةً عدد المهام التي يعالجها فريقك في اللحظة نفسها.

تصميم جدول DynamoDB

نمط الجدول الموحّد

بدلاً من توزيع البيانات على عدة جداول، يكفي جدول DynamoDB واحد لتخزين سجل الحلول، والمهام النشطة، والإحصائيات المجمّعة. يعتمد التصميم على مفتاح تقسيم (PK) ومفتاح فرز (SK) يميّزان نوع العنصر:

هذا النمط يتيح لك جلب تاريخ موقع محدّد، أو إحصائيات يوم معيّن، أو قائمة المهام النشطة، كلٌّ منها باستعلام واحد فعّال دون مسح كامل للجدول.

تعريف الجدول

يعرّف المخطّط التالي المفاتيح الأساسية، وفهرساً ثانوياً عاماً (GSI1) للاستعلام حسب الحالة، ونمط الفوترة عند الطلب، وتفعيل حقل TTL:

{
  "TableName": "CaptchaSolves",
  "KeySchema": [
    { "AttributeName": "PK", "KeyType": "HASH" },
    { "AttributeName": "SK", "KeyType": "RANGE" }
  ],
  "AttributeDefinitions": [
    { "AttributeName": "PK", "KeyType": "S" },
    { "AttributeName": "SK", "KeyType": "S" },
    { "AttributeName": "GSI1PK", "KeyType": "S" },
    { "AttributeName": "GSI1SK", "KeyType": "S" }
  ],
  "GlobalSecondaryIndexes": [
    {
      "IndexName": "GSI1",
      "KeySchema": [
        { "AttributeName": "GSI1PK", "KeyType": "HASH" },
        { "AttributeName": "GSI1SK", "KeyType": "RANGE" }
      ],
      "Projection": { "ProjectionType": "ALL" }
    }
  ],
  "BillingMode": "PAY_PER_REQUEST",
  "TimeToLiveSpecification": {
    "AttributeName": "ttl",
    "Enabled": true
  }
}

تنفيذ التتبّع في Python

الإعداد والاتصال

ابدأ بتهيئة مورد DynamoDB وقراءة مفتاح الـ API من متغيّرات البيئة، دون تضمينه في الشيفرة مطلقاً:

import os
import time
from datetime import datetime, timezone
import boto3
import requests

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ.get("DYNAMODB_TABLE", "CaptchaSolves"))
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

دالة الحل والتتبّع

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

def solve_and_track(sitekey, pageurl, captcha_type="recaptcha_v2", project=None):
    now = datetime.now(timezone.utc)
    timestamp = now.isoformat()
    ttl_90_days = int(now.timestamp()) + (90 * 24 * 3600)

    # Submit to CaptchaAI
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        # Store error record
        table.put_item(Item={
            "PK": f"SITE#{sitekey}",
            "SK": f"SOLVE#{timestamp}",
            "captcha_type": captcha_type,
            "pageurl": pageurl,
            "status": "error",
            "error": data.get("request"),
            "submitted_at": timestamp,
            "project": project or "default",
            "ttl": ttl_90_days,
            "GSI1PK": f"STATUS#error",
            "GSI1SK": timestamp
        })
        return {"error": data.get("request")}

    captcha_id = data["request"]

    # Track active task
    table.put_item(Item={
        "PK": f"ACTIVE#{captcha_id}",
        "SK": "TASK",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "captcha_type": captcha_type,
        "submitted_at": timestamp,
        "ttl": int(now.timestamp()) + 600  # Auto-clean in 10 min
    })

    # Poll for result
    polls = 0
    for _ in range(60):
        time.sleep(5)
        polls += 1
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get",
            "id": captcha_id, "json": 1
        }).json()

        if result.get("status") == 1:
            solved_at = datetime.now(timezone.utc).isoformat()
            elapsed_ms = int(
                (datetime.now(timezone.utc) - now).total_seconds() * 1000
            )

            # Store success record
            table.put_item(Item={
                "PK": f"SOLVE#{captcha_id}",
                "SK": "META",
                "captcha_type": captcha_type,
                "sitekey": sitekey,
                "pageurl": pageurl,
                "status": "solved",
                "submitted_at": timestamp,
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls,
                "project": project or "default",
                "ttl": ttl_90_days,
                "GSI1PK": f"STATUS#solved",
                "GSI1SK": timestamp
            })

            # Also store in site history
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "solved",
                "elapsed_ms": elapsed_ms,
                "ttl": ttl_90_days
            })

            # Remove active task
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })

            # Update daily stats
            update_daily_stats(captcha_type, True, elapsed_ms)

            return {"solution": result["request"]}

        if result.get("request") != "CAPCHA_NOT_READY":
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "error",
                "error": result.get("request"),
                "ttl": ttl_90_days
            })
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })
            update_daily_stats(captcha_type, False, 0)
            return {"error": result.get("request")}

    table.delete_item(Key={"PK": f"ACTIVE#{captcha_id}", "SK": "TASK"})
    update_daily_stats(captcha_type, False, 0)
    return {"error": "TIMEOUT"}


def update_daily_stats(captcha_type, success, elapsed_ms):
    date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    update_expr = "SET total_solves = if_not_exists(total_solves, :zero) + :one"
    expr_values = {":zero": 0, ":one": 1}

    if success:
        update_expr += ", successful = if_not_exists(successful, :zero) + :one"
        update_expr += ", total_elapsed = if_not_exists(total_elapsed, :zero) + :elapsed"
        expr_values[":elapsed"] = elapsed_ms
    else:
        update_expr += ", failed = if_not_exists(failed, :zero) + :one"

    table.update_item(
        Key={"PK": f"STATS#{date_str}", "SK": f"TYPE#{captcha_type}"},
        UpdateExpression=update_expr,
        ExpressionAttributeValues=expr_values
    )

أنماط الاستعلام

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

def get_site_history(sitekey, limit=50):
    """Get recent solves for a specific site key."""
    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"SITE#{sitekey}"},
        ScanIndexForward=False,
        Limit=limit
    )
    return response["Items"]


def get_daily_stats(date_str=None):
    """Get stats for a specific date (default: today)."""
    if not date_str:
        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"STATS#{date_str}"}
    )
    return response["Items"]


def get_active_tasks():
    """List all currently active CAPTCHA tasks."""
    response = table.query(
        IndexName="GSI1",
        KeyConditionExpression="GSI1PK = :pk",
        ExpressionAttributeValues={":pk": "STATUS#polling"}
    )
    return response["Items"]

تنفيذ التتبّع في JavaScript

إذا كانت دوالك مكتوبة بـ Node.js، فالمنطق نفسه ينطبق باستخدام حزمة AWS SDK v3 وعميل المستندات. المثال التالي يرسل المهمة، ويكتب سجل الخطأ عند فشل الإرسال، ثم يستطلع النتيجة ويخزّن سجل النجاح مع زمن الاستجابة:

const { DynamoDBClient } = require("@aws-sdk/client-dynamodb");
const { DynamoDBDocumentClient, PutCommand, QueryCommand, UpdateCommand } = require("@aws-sdk/lib-dynamodb");
const axios = require("axios");

const client = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const TABLE = process.env.DYNAMODB_TABLE || "CaptchaSolves";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

async function solveAndTrack(sitekey, pageurl, type = "recaptcha_v2") {
  const now = new Date();
  const timestamp = now.toISOString();
  const ttl = Math.floor(now.getTime() / 1000) + 90 * 24 * 3600;

  const submit = await axios.post("https://ocr.captchaai.com/in.php", null, {
    params: { key: API_KEY, method: "userrecaptcha", googlekey: sitekey, pageurl, json: 1 },
  });

  if (submit.data.status !== 1) {
    await client.send(new PutCommand({
      TableName: TABLE,
      Item: { PK: `SITE#${sitekey}`, SK: `SOLVE#${timestamp}`, status: "error", error: submit.data.request, ttl },
    }));
    return { error: submit.data.request };
  }

  const captchaId = submit.data.request;
  let polls = 0;

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    polls++;
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
    });

    if (poll.data.status === 1) {
      const elapsed = Date.now() - now.getTime();
      await client.send(new PutCommand({
        TableName: TABLE,
        Item: {
          PK: `SOLVE#${captchaId}`, SK: "META", captcha_type: type,
          sitekey, pageurl, status: "solved", submitted_at: timestamp,
          solved_at: new Date().toISOString(), elapsed_ms: elapsed, polls, ttl,
        },
      }));
      return { solution: poll.data.request };
    }

    if (poll.data.request !== "CAPCHA_NOT_READY") {
      return { error: poll.data.request };
    }
  }
  return { error: "TIMEOUT" };
}

async function getSiteHistory(sitekey, limit = 50) {
  const result = await client.send(new QueryCommand({
    TableName: TABLE,
    KeyConditionExpression: "PK = :pk",
    ExpressionAttributeValues: { ":pk": `SITE#${sitekey}` },
    ScanIndexForward: false,
    Limit: limit,
  }));
  return result.Items;
}

خفض التكلفة

DynamoDB اقتصادي لهذا النوع من الأحمال إذا ضبطت بضع خيارات منذ البداية. فوترة عند الطلب تعفيك من إعادة التوفير اليدوي، وحقل TTL يقلّص التخزين تلقائياً، وإسقاط السمات غير الضرورية في الاستعلامات يوفّر وحدات القراءة:

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

معظم المشكلات هنا تتعلّق بحدود الكتابة أو توزيع المفاتيح، لا بمنطق الحل نفسه:

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

كيف يعمل حقل TTL في DynamoDB لتنظيف السجلات؟

تحدّد لكل عنصر قيمة ttl بصيغة طابع زمني Unix، وبعد تجاوز ذلك الوقت يصبح العنصر مؤهّلاً للحذف. لكن الحذف نهائي التناسق وقد يتأخر حتى نحو 48 ساعة، لذا لا تعامله كتنظيف لحظي، بل صفِّ العناصر المنتهية داخل استعلاماتك عند الحاجة إلى دقة فورية.

هل يمكن تشغيل هذا التتبّع داخل دالة AWS Lambda مباشرة؟

نعم، وهذا هو الاستخدام الأمثل. لأن DynamoDB لا يعتمد على اتصال دائم، يستدعي كل تشغيل للدالة الجدول عبر HTTP دون طبقة تجميع. امنح دور تنفيذ الدالة صلاحيات PutItem وQuery وUpdateItem على الجدول فقط، والتزم بمبدأ أقل الامتيازات.

ما الفرق بين المهام النشطة وسجل الحلول في الجدول نفسه؟

عنصر ACTIVE#{captcha_id} يمثّل مهمة قيد التنفيذ بحقل TTL قصير (عشر دقائق) يمحوها تلقائياً إن تعطّل التنفيذ، بينما SOLVE#{captcha_id} هو السجل الدائم الذي يبقى 90 يوماً لأغراض التحليل. الفصل بينهما يمنع المهام المعلّقة من تلويث إحصائياتك.

كيف أتجنّب مشكلة التقسيم الساخن عند تجميع الإحصائيات؟

عندما يكتب كل العمّال إلى STATS#{date} نفسه ينشأ قسم ساخن يخنق الأداء. وزّع الحِمل بإضافة لاحقة عشوائية مثل STATS#{date}#shard{0-9}، ثم اجمع نتائج الأجزاء عند القراءة. بديل آخر هو تفريغ التجميع إلى DynamoDB Streams ودالة Lambda.

كيف أربط عدد الـ Threads في خطة CaptchaAI بمعدل الكتابة؟

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

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

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

أدلة ذات صلة

• دليل البدء السريع مع CaptchaAI
• اكتشاف اختبارات CAPTCHA المحمّلة ديناميكياً