تخزين سجل حل CAPTCHA وتحليلاته في MongoDB

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

لماذا تناسب MongoDB بيانات حل CAPTCHA

تختلف حقول سجلّ الحل باختلاف نوع الاختبار؛ فـ reCAPTCHA يحتاج إلى googlekey، ويحتاج Cloudflare Turnstile إلى sitekey، بينما تحتاج كابتشا الصور إلى الحقل body. تتعامل وثائق MongoDB غير المقيّدة بمخطّط ثابت مع هذا التباين طبيعياً، فتضيف حقلاً جديداً متى ظهر نوع اختبار جديد دون ترحيل للمخطط أو تعديل للجداول. يضاف إلى ذلك أن إطار التجميع يتيح حساب معدلات الحل ومتوسطات الزمن مباشرة داخل قاعدة البيانات، بينما تتكفّل مؤشرات TTL بحذف السجلات القديمة تلقائياً دون وظائف تنظيف خارجية.

تصميم مخطط الوثيقة

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

{
  "_id": "ObjectId",
  "captcha_id": "12345678",
  "type": "recaptcha_v2",
  "method": "userrecaptcha",
  "sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
  "pageurl": "https://example.com/form",
  "status": "solved",
  "solution": "03AGdBq26...",
  "error": null,
  "submitted_at": "2026-04-04T10:15:30.000Z",
  "solved_at": "2026-04-04T10:15:45.000Z",
  "elapsed_ms": 15000,
  "polls": 3,
  "proxy_used": true,
  "cost": 0.00299,
  "metadata": {
    "project": "price-monitor",
    "worker_id": "worker-3",
    "target_domain": "example.com"
  }
}

تحمل حقول submitted_at وsolved_at وelapsed_ms أساس كل تحليلات الأداء لاحقاً، بينما يمنحك حقل metadata مرونة لوسم كل عملية بالمشروع أو النطاق المستهدف أو معرّف العامل.

التنفيذ بلغة Python

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

ابدأ بتحميل بيانات الاعتماد من متغيّرات البيئة، ثم أنشئ اتصالاً بقاعدة البيانات وبمجموعة solves. إبقاء مفتاح الـ API خارج الشيفرة شرط أساسي لأي بيئة إنتاج:

import os
import time
from datetime import datetime, timezone
from pymongo import MongoClient, ASCENDING, DESCENDING
import requests

MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017")
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

client = MongoClient(MONGO_URI)
db = client["captcha_tracking"]
solves = db["solves"]

إنشاء الفهارس

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

def setup_indexes():
    solves.create_index([("submitted_at", DESCENDING)])
    solves.create_index([("type", ASCENDING), ("status", ASCENDING)])
    solves.create_index([("metadata.project", ASCENDING)])
    solves.create_index([("metadata.target_domain", ASCENDING)])
    solves.create_index(
        [("submitted_at", ASCENDING)],
        expireAfterSeconds=90 * 24 * 3600,  # Auto-delete after 90 days
        name="ttl_cleanup"
    )

setup_indexes()

الحل والتخزين

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

def solve_and_store(sitekey, pageurl, captcha_type="recaptcha_v2", metadata=None):
    record = {
        "type": captcha_type,
        "method": "userrecaptcha",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "status": "submitted",
        "submitted_at": datetime.now(timezone.utc),
        "metadata": metadata or {}
    }

    result = solves.insert_one(record)
    doc_id = result.inserted_id

    # 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:
        solves.update_one(
            {"_id": doc_id},
            {"$set": {"status": "error", "error": data.get("request")}}
        )
        return None

    captcha_id = data["request"]
    solves.update_one(
        {"_id": doc_id},
        {"$set": {"captcha_id": captcha_id, "status": "polling"}}
    )

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

        if poll_resp.get("status") == 1:
            solved_at = datetime.now(timezone.utc)
            elapsed_ms = int(
                (solved_at - record["submitted_at"]).total_seconds() * 1000
            )
            solves.update_one({"_id": doc_id}, {"$set": {
                "status": "solved",
                "solution": poll_resp["request"],
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls
            }})
            return poll_resp["request"]

        if poll_resp.get("request") != "CAPCHA_NOT_READY":
            solves.update_one({"_id": doc_id}, {"$set": {
                "status": "error",
                "error": poll_resp.get("request"),
                "polls": polls
            }})
            return None

    solves.update_one({"_id": doc_id}, {"$set": {
        "status": "timeout", "polls": polls
    }})
    return None

استعلامات التحليلات

بعد تراكم السجلات، تحوّل خطوط التجميع البيانات الخام إلى مؤشرات مفيدة: معدل الحل خلال ساعات محددة، ومتوسط زمن الحل لكل نوع، وحجم العمليات بالساعة لرسم المخططات، وتوزيع رموز الخطأ الأكثر تكراراً:

def get_success_rate(hours=24):
    """Success rate for the last N hours."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}}},
        {"$group": {
            "_id": "$status",
            "count": {"$sum": 1}
        }}
    ]
    results = {r["_id"]: r["count"] for r in solves.aggregate(pipeline)}
    total = sum(results.values())
    solved = results.get("solved", 0)
    return (solved / total * 100) if total else 0


def get_avg_solve_time_by_type():
    """Average solve time grouped by CAPTCHA type."""
    pipeline = [
        {"$match": {"status": "solved"}},
        {"$group": {
            "_id": "$type",
            "avg_time_ms": {"$avg": "$elapsed_ms"},
            "min_time_ms": {"$min": "$elapsed_ms"},
            "max_time_ms": {"$max": "$elapsed_ms"},
            "count": {"$sum": 1}
        }},
        {"$sort": {"count": -1}}
    ]
    return list(solves.aggregate(pipeline))


def get_hourly_solve_volume(days=7):
    """Hourly solve volume for charting."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}}},
        {"$group": {
            "_id": {
                "date": {"$dateToString": {"format": "%Y-%m-%d", "date": "$submitted_at"}},
                "hour": {"$hour": "$submitted_at"}
            },
            "total": {"$sum": 1},
            "solved": {"$sum": {"$cond": [{"$eq": ["$status", "solved"]}, 1, 0]}}
        }},
        {"$sort": {"_id.date": 1, "_id.hour": 1}}
    ]
    return list(solves.aggregate(pipeline))


def get_error_breakdown(hours=24):
    """Error frequency by error code."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}, "status": "error"}},
        {"$group": {"_id": "$error", "count": {"$sum": 1}}},
        {"$sort": {"count": -1}}
    ]
    return list(solves.aggregate(pipeline))

التنفيذ بلغة JavaScript

يعيد المثال التالي بناء المسار نفسه — الاتصال، الفهرسة، الحل والتخزين، وحساب معدل الحل — لمن يبني خط الأتمتة على Node.js بدلاً من Python:

const { MongoClient } = require("mongodb");
const axios = require("axios");

const MONGO_URI = process.env.MONGO_URI || "mongodb://localhost:27017";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

let db, solves;

async function connect() {
  const client = await MongoClient.connect(MONGO_URI);
  db = client.db("captcha_tracking");
  solves = db.collection("solves");

  await solves.createIndex({ submitted_at: -1 });
  await solves.createIndex({ type: 1, status: 1 });
  await solves.createIndex({ "metadata.project": 1 });
  await solves.createIndex(
    { submitted_at: 1 },
    { expireAfterSeconds: 90 * 24 * 3600 }
  );
}

async function solveAndStore(sitekey, pageurl, type = "recaptcha_v2", metadata = {}) {
  const submittedAt = new Date();
  const { insertedId } = await solves.insertOne({
    type, method: "userrecaptcha", sitekey, pageurl,
    status: "submitted", submitted_at: submittedAt, metadata,
  });

  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 solves.updateOne({ _id: insertedId }, { $set: { status: "error", error: submit.data.request } });
    return null;
  }

  const captchaId = submit.data.request;
  await solves.updateOne({ _id: insertedId }, { $set: { captcha_id: captchaId, status: "polling" } });

  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 solvedAt = new Date();
      await solves.updateOne({ _id: insertedId }, { $set: {
        status: "solved", solution: poll.data.request,
        solved_at: solvedAt, elapsed_ms: solvedAt - submittedAt, polls,
      }});
      return poll.data.request;
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      await solves.updateOne({ _id: insertedId }, { $set: { status: "error", error: poll.data.request, polls } });
      return null;
    }
  }

  await solves.updateOne({ _id: insertedId }, { $set: { status: "timeout", polls } });
  return null;
}

async function getSuccessRate(hours = 24) {
  const cutoff = new Date(Date.now() - hours * 3600 * 1000);
  const pipeline = [
    { $match: { submitted_at: { $gte: cutoff } } },
    { $group: { _id: "$status", count: { $sum: 1 } } },
  ];
  const results = await solves.aggregate(pipeline).toArray();
  const total = results.reduce((s, r) => s + r.count, 0);
  const solved = results.find((r) => r._id === "solved")?.count || 0;
  return total ? ((solved / total) * 100).toFixed(1) : 0;
}

سيناريو عملي: مراقبة معدل الحل لفريق أتمتة

تخيّل فريق أتمتة يراقب أسعار متجر إلكتروني إقليمي ويشغّل عشرات عمليات الحل يومياً عبر reCAPTCHA v2 وCloudflare Turnstile. بتخزين كل عملية في MongoDB، يستطيع الفريق تشغيل get_avg_solve_time_by_type() كل صباح لرصد أي ارتفاع مفاجئ في زمن الحل قد يشير إلى تغيّر في الموقع المستهدف، واستخدام get_error_breakdown() لتحديد أكثر رموز الخطأ تكراراً خلال اليوم. وإذا كان الفريق يشغّل خطة ADVANCE ($90 شهرياً، 50 thread)، فإن لوحة تحليلات مبنية على هذه الاستعلامات تكشف ما إذا كان يستغل سعة الـ threads بالكامل أم أن هناك اختناقاً في مكان آخر من الخط. هذه الرؤية اليومية تحوّل التتبّع من سجلّ سلبي إلى أداة تشغيلية تسبق المشكلات قبل وقوعها.

سياسات الاحتفاظ بالبيانات

تعتمد مدة الاحتفاظ على الغرض: بيانات التطوير قصيرة العمر، بينما تتطلب تحليلات الإنتاج نافذة أطول، وقد يفرض الامتثال أرشفة دائمة. تضبط مؤشرات TTL هذا تلقائياً:

استكشاف أخطاء التخزين والتحليلات

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

ما الفهارس التي أحتاجها لتسريع استعلامات التحليلات؟

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

كيف أحسب معدّل الحل ومتوسط زمن الاستجابة؟

عبر إطار التجميع داخل MongoDB مباشرة. تجمّع الدالة get_success_rate السجلات حسب الحالة وتقسم الناجحة على الإجمالي، بينما تحسب get_avg_solve_time_by_type متوسط elapsed_ms وأدنى وأقصى قيمة لكل نوع اختبار. تشغيلها على نافذة زمنية محددة يمنحك لوحة أداء محدّثة باستمرار دون أدوات خارجية.

كيف أتحكّم في نمو قاعدة البيانات مع تراكم السجلات؟

اعتمد مؤشر TTL يحذف السجلات تلقائياً بعد 30 أو 90 يوماً حسب حاجتك، وخزّن البيانات الوصفية — النوع والزمن والحالة والخطأ — فقط دون التوكن الكامل الذي يصبح عديم الفائدة بعد انتهاء صلاحيته. هذا يبقي حجم التخزين تحت السيطرة مهما ارتفع عدد العمليات.

هل تفرض CaptchaAI رسوماً لكل عملية حل عند التسجيل المكثّف؟

لا؛ تعتمد CaptchaAI تسعيراً قائماً على الـ threads المتزامنة مع عدد غير محدود من عمليات الحل لكل thread خلال الشهر. تبدأ الخطط من BASIC ($15 شهرياً، 5 threads) وتتوسّع حتى ENTERPRISE ($300 شهرياً، 200 thread)، فتبقى تكلفة اشتراكك ثابتة مهما بلغ عدد السجلات التي تخزّنها وتحلّلها.

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

• ابدأ بسرعة مع CaptchaAI وحلّ أول كابتشا خلال دقائق
• حلّ reCAPTCHA v2 عبر الـ API خطوة بخطوة
• حلّ Cloudflare Turnstile باستخدام الـ API
• حلّ GeeTest v3 باستخدام الـ API

أدلة ذات صلة

• دليل البدء السريع مع CaptchaAI
• حلّ reCAPTCHA v2 عبر الـ API