تخزين نتائج حل CAPTCHA وتتبعها محلياً باستخدام SQLite

كل عملية حل CAPTCHA تشغل خيط معالجة — thread — من حسابك لبضع ثوانٍ، وإذا أرسل السكربت الطلب نفسه مرتين فقد استهلكت خيطين مقابل نتيجة واحدة. العلاج لا يبدأ بكاش خارجي ولا بقاعدة مركزية، بل بملف SQLite واحد بجوار السكربت: جدول يسجّل كل محاولة، وجدول ثانٍ يحفظ الرمز الناتج حتى انتهاء صلاحيته. نبني الاثنين هنا بـ Python ثم بـ Node.js، مع الاستعلامات التي تكشف معدل الحل ووقته الفعلي.

لماذا يستحق كاش الحل المحلي في SQLite عناء الإعداد

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

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

وملف SQLite يعالج الثلاثة دفعة واحدة: لا خادم يُدار ولا منفذ يُفتح، والوحدة sqlite3 جزء من مكتبة Python القياسية.

متى يكون SQLite خياراً صحيحاً ومتى لا يكون

SQLite مناسب ما دامت العمليات كلها ترى القرص نفسه، ويصبح عائقاً حين يتوزّع العمل على خوادم متعددة.

الخطوة 1: صمّم جدول التتبع وجدول الكاش

CREATE TABLE IF NOT EXISTS captcha_solves (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    captcha_id TEXT,
    type TEXT NOT NULL,
    sitekey TEXT,
    pageurl TEXT,
    status TEXT NOT NULL DEFAULT 'submitted',
    solution TEXT,
    error TEXT,
    submitted_at TEXT NOT NULL DEFAULT (datetime('now')),
    solved_at TEXT,
    elapsed_ms INTEGER,
    polls INTEGER DEFAULT 0,
    project TEXT
);

CREATE INDEX IF NOT EXISTS idx_submitted_at ON captcha_solves(submitted_at);
CREATE INDEX IF NOT EXISTS idx_type_status ON captcha_solves(type, status);
CREATE INDEX IF NOT EXISTS idx_sitekey ON captcha_solves(sitekey);

-- Token cache for reuse within TTL
CREATE TABLE IF NOT EXISTS token_cache (
    sitekey TEXT NOT NULL,
    pageurl TEXT NOT NULL,
    token TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    expires_at TEXT NOT NULL,
    used INTEGER DEFAULT 0,
    PRIMARY KEY (sitekey, pageurl, token)
);

CREATE INDEX IF NOT EXISTS idx_cache_lookup
ON token_cache(sitekey, pageurl, used, expires_at);

الجدول الأول captcha_solves سجل تاريخي: صف لكل محاولة بحالة تتدرّج من submitted إلى polling ثم solved أو error أو timeout. العمودان elapsed_ms وpolls هما مصدر أرقام الأداء لاحقاً، وproject يفصل بين بيئاتك حين تشترك عدة سكربتات في الملف نفسه.

الجدول الثاني token_cache قصير العمر بطبيعته، ومفتاحه الأساسي مركّب. عمود expires_at يمنع تسليم رمز ميت، وused يضمن ألا يُسحب الرمز نفسه مرتين، والفهرس idx_cache_lookup يجعل البحث عن رمز صالح يقرأ من الفهرس بدل مسح الجدول.

الخطوة 2: هيّئ القاعدة وفعّل وضع WAL

import os
import time
import sqlite3
from datetime import datetime, timedelta, timezone
import requests

DB_PATH = os.environ.get("CAPTCHA_DB", "captcha_solves.db")
API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")  # Better concurrent read performance
    conn.execute("PRAGMA busy_timeout=5000")
    return conn


def init_db():
    conn = get_db()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS captcha_solves (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            captcha_id TEXT,
            type TEXT NOT NULL,
            sitekey TEXT,
            pageurl TEXT,
            status TEXT NOT NULL DEFAULT 'submitted',
            solution TEXT,
            error TEXT,
            submitted_at TEXT NOT NULL DEFAULT (datetime('now')),
            solved_at TEXT,
            elapsed_ms INTEGER,
            polls INTEGER DEFAULT 0,
            project TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_submitted_at ON captcha_solves(submitted_at);
        CREATE INDEX IF NOT EXISTS idx_type_status ON captcha_solves(type, status);

        CREATE TABLE IF NOT EXISTS token_cache (
            sitekey TEXT NOT NULL,
            pageurl TEXT NOT NULL,
            token TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            expires_at TEXT NOT NULL,
            used INTEGER DEFAULT 0,
            PRIMARY KEY (sitekey, pageurl, token)
        );
        CREATE INDEX IF NOT EXISTS idx_cache_lookup
        ON token_cache(sitekey, pageurl, used, expires_at);
    """)
    conn.close()

init_db()

سطران في get_db() يفرّقان بين قاعدة تعمل وأخرى تتعطل عند أول ضغط: journal_mode=WAL يسمح للقراءات بالاستمرار أثناء الكتابة، وbusy_timeout=5000 يجعل العملية تنتظر خمس ثوانٍ قبل أن ترفع خطأ القفل. أما row_factory فيتيح قراءة الأعمدة بالاسم، وعليه يعتمد كود الكاش لاحقاً.

واستدعاء init_db() عند الإقلاع آمن للتكرار لأن كل عبارة إنشاء مكتوبة بصيغة IF NOT EXISTS، فلا حاجة إلى ترحيل يدوي على جهاز جديد.

الخطوة 3: أرسل الطلب وسجّل كل محاولة

def solve_recaptcha(sitekey, pageurl, project=None):
    conn = get_db()

    # Check cache first
    cached = get_cached_token(conn, sitekey, pageurl)
    if cached:
        conn.close()
        return cached

    # Insert tracking record
    now = datetime.now(timezone.utc).isoformat()
    cursor = conn.execute(
        "INSERT INTO captcha_solves (type, sitekey, pageurl, submitted_at, project) "
        "VALUES (?, ?, ?, ?, ?)",
        ("recaptcha_v2", sitekey, pageurl, now, project)
    )
    row_id = cursor.lastrowid
    conn.commit()

    # 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:
        conn.execute(
            "UPDATE captcha_solves SET status=?, error=? WHERE id=?",
            ("error", data.get("request"), row_id)
        )
        conn.commit()
        conn.close()
        return None

    captcha_id = data["request"]
    conn.execute(
        "UPDATE captcha_solves SET captcha_id=?, status=? WHERE id=?",
        (captcha_id, "polling", row_id)
    )
    conn.commit()

    # Poll
    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()
            submitted = datetime.fromisoformat(now)
            elapsed = int((datetime.now(timezone.utc) - submitted).total_seconds() * 1000)

            conn.execute(
                "UPDATE captcha_solves SET status=?, solution=?, solved_at=?, "
                "elapsed_ms=?, polls=? WHERE id=?",
                ("solved", result["request"], solved_at, elapsed, polls, row_id)
            )
            # Cache the token
            cache_token(conn, sitekey, pageurl, result["request"])
            conn.commit()
            conn.close()
            return result["request"]

        if result.get("request") != "CAPCHA_NOT_READY":
            conn.execute(
                "UPDATE captcha_solves SET status=?, error=?, polls=? WHERE id=?",
                ("error", result.get("request"), polls, row_id)
            )
            conn.commit()
            conn.close()
            return None

    conn.execute(
        "UPDATE captcha_solves SET status=?, polls=? WHERE id=?",
        ("timeout", polls, row_id)
    )
    conn.commit()
    conn.close()
    return None

ترتيب العمليات مقصود: الكاش أولاً، ثم إدراج صف التتبع، ثم الإرسال إلى in.php، ثم الاستطلاع الدوري على res.php. إذا وُجد رمز صالح في الكاش تنتهي الدالة قبل أن تلمس الـ API أصلاً.

كل انتقال في الحالة يُحفظ فوراً بـ commit(). لو توقّف السكربت أثناء الاستطلاع يبقى الصف على polling ومعه captcha_id، فتستطيع استئناف الاستفسار لاحقاً. والفصل بين timeout وerror مفيد للتشخيص: الأول انقضاء مهلة، والثاني رمز خطأ صريح يستحق قراءة نصه.

الخطوة 4: اضبط عمر الرمز داخل الكاش

def cache_token(conn, sitekey, pageurl, token, ttl_seconds=90):
    expires_at = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat()
    conn.execute(
        "INSERT OR REPLACE INTO token_cache (sitekey, pageurl, token, expires_at) "
        "VALUES (?, ?, ?, ?)",
        (sitekey, pageurl, token, expires_at)
    )


def get_cached_token(conn, sitekey, pageurl):
    now = datetime.now(timezone.utc).isoformat()
    row = conn.execute(
        "SELECT token FROM token_cache "
        "WHERE sitekey=? AND pageurl=? AND used=0 AND expires_at > ? "
        "ORDER BY expires_at ASC LIMIT 1",
        (sitekey, pageurl, now)
    ).fetchone()

    if row:
        conn.execute(
            "UPDATE token_cache SET used=1 WHERE token=?",
            (row["token"],)
        )
        conn.commit()
        return row["token"]
    return None

قيمة ttl_seconds=90 ليست اعتباطية: رموز reCAPTCHA v2 تبقى صالحة في حدود 90 إلى 120 ثانية، والهامش المحافظ أسلم من تسليم رمز يرفضه الموقع المستهدف. مع نوع تحقق آخر، اضبط القيمة على أقصر عمر معروف لديك ولا ترفعها لمجرد زيادة نسبة الإصابة.

الدالة get_cached_token تفعل شيئين في النداء الواحد: تختار أقرب رمز صالح إلى انتهاء صلاحيته، ثم تعلّمه used=1 مباشرة. هذا يمنع سحب الرمز نفسه من عمليتين متوازيتين، ويُبقي قاعدة «استخدام واحد لكل رمز» داخل قاعدة البيانات لا في انضباط المستدعي.

الخطوة 5: قِس الأداء ونظّف السجلات القديمة

def get_stats(hours=24):
    conn = get_db()
    cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()

    total = conn.execute(
        "SELECT COUNT(*) FROM captcha_solves WHERE submitted_at >= ?", (cutoff,)
    ).fetchone()[0]

    solved = conn.execute(
        "SELECT COUNT(*) FROM captcha_solves WHERE submitted_at >= ? AND status='solved'",
        (cutoff,)
    ).fetchone()[0]

    avg_time = conn.execute(
        "SELECT AVG(elapsed_ms) FROM captcha_solves "
        "WHERE submitted_at >= ? AND status='solved'",
        (cutoff,)
    ).fetchone()[0]

    conn.close()
    return {
        "total": total,
        "solved": solved,
        "success_rate": (solved / total * 100) if total else 0,
        "avg_time_ms": round(avg_time) if avg_time else 0
    }


def cleanup_old_records(days=30):
    conn = get_db()
    cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    conn.execute("DELETE FROM captcha_solves WHERE submitted_at < ?", (cutoff,))
    conn.execute("DELETE FROM token_cache WHERE expires_at < ?",
                 (datetime.now(timezone.utc).isoformat(),))
    conn.execute("VACUUM")
    conn.commit()
    conn.close()

get_stats() تعطيك ثلاثة أرقام تكفي لمتابعة يومية: عدد المحاولات، عدد الناجح منها، ومتوسط الزمن بالمللي ثانية. اربطها بمهمة مجدولة ترسل الناتج إلى قناة الفريق، وستلاحظ تدهور الأداء مبكراً.

أما cleanup_old_records() فهي ما يمنع الملف من التضخّم: احذف سجلات التتبع الأقدم من ثلاثين يوماً وصفوف الكاش المنتهية، ثم شغّل VACUUM لاستعادة المساحة فعلياً — حذف الصفوف وحده لا يقلّص حجم الملف. شغّلها خارج ساعات الذروة لأنها تعيد كتابة القاعدة بالكامل.

نفس المنطق في Node.js

const Database = require("better-sqlite3");
const axios = require("axios");

const db = new Database(process.env.CAPTCHA_DB || "captcha_solves.db");
const API_KEY = process.env.CAPTCHAAI_API_KEY;

db.pragma("journal_mode = WAL");
db.exec(`
  CREATE TABLE IF NOT EXISTS captcha_solves (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    captcha_id TEXT, type TEXT NOT NULL, sitekey TEXT, pageurl TEXT,
    status TEXT DEFAULT 'submitted', solution TEXT, error TEXT,
    submitted_at TEXT DEFAULT (datetime('now')),
    solved_at TEXT, elapsed_ms INTEGER, polls INTEGER DEFAULT 0
  );
  CREATE INDEX IF NOT EXISTS idx_submitted ON captcha_solves(submitted_at);
`);

async function solveAndStore(sitekey, pageurl) {
  const submittedAt = new Date().toISOString();
  const insert = db.prepare(
    "INSERT INTO captcha_solves (type, sitekey, pageurl, submitted_at) VALUES (?, ?, ?, ?)"
  );
  const { lastInsertRowid } = insert.run("recaptcha_v2", sitekey, pageurl, submittedAt);

  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) {
    db.prepare("UPDATE captcha_solves SET status=?, error=? WHERE id=?")
      .run("error", submit.data.request, lastInsertRowid);
    return null;
  }

  const captchaId = submit.data.request;
  db.prepare("UPDATE captcha_solves SET captcha_id=?, status=? WHERE id=?")
    .run(captchaId, "polling", lastInsertRowid);

  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() - new Date(submittedAt).getTime();
      db.prepare(
        "UPDATE captcha_solves SET status=?, solution=?, solved_at=?, elapsed_ms=?, polls=? WHERE id=?"
      ).run("solved", poll.data.request, new Date().toISOString(), elapsed, polls, lastInsertRowid);
      return poll.data.request;
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      db.prepare("UPDATE captcha_solves SET status=?, error=?, polls=? WHERE id=?")
        .run("error", poll.data.request, polls, lastInsertRowid);
      return null;
    }
  }

  db.prepare("UPDATE captcha_solves SET status=?, polls=? WHERE id=?")
    .run("timeout", polls, lastInsertRowid);
  return null;
}

حزمة better-sqlite3 متزامنة عمداً، وهذا مناسب هنا: الكتابة قصيرة، والانتظار الحقيقي يقع على الشبكة لا على القرص. ويُضبط journal_mode = WAL مرة واحدة عند فتح القاعدة كما في نسخة Python، بينما يُعاد استخدام العبارات المُحضَّرة بـ prepare() بدل تحليل نص SQL في كل نداء.

سيناريو تشغيلي: فريق اختبار في القاهرة

تخيّل فريق QA يشغّل كل ليلة مجموعة اختبارات على بوابة حجوزات، وتمرّ 40 حالة اختبار بصفحة التحقق نفسها. بدون كاش يرسل السكربت 40 طلب حل لنفس sitekey خلال دقيقتين. ومع جدول token_cache وعمر صلاحية 90 ثانية، تُخدم الحالات الواقعة داخل النافذة من رمز واحد.

الأثر المباشر ليس على الفاتورة الشهرية، فتسعير CaptchaAI قائم على عدد الخيوط المتزامنة لا على عدد العمليات، وسقف باقة STANDARD ثابت عند 15 thread بسعر ‏$30 شهرياً. الأثر الحقيقي يقع على الزمن: كل طلب موفَّر هو ثوانٍ محذوفة من التشغيل الليلي، وخيط متاح لحالة اختبار أخرى. وإن وزّع الفريق الاختبارات على أجهزة عدة، فسيحمل كل جهاز كاشه المنفصل، وعندها يكون وقت الانتقال إلى قاعدة مشتركة قد حان.

أعطال شائعة وكيف تعالجها

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

هل يقلّل الكاش المحلي تكلفة CaptchaAI؟

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

هل يجوز إعادة استخدام الرمز نفسه لأكثر من طلب؟

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

أين أضع ملف القاعدة عند التشغيل داخل Docker؟

في مجلد مربوط بـ volume خارج طبقة الحاوية، مع ضبط CAPTCHA_DB على مساره؛ وإلا فقدت السجل والكاش مع كل إعادة نشر. وتجنّب المشاركات الشبكية مثل NFS، فقفل الملفات هناك غير موثوق مع SQLite.

متى أنتقل من SQLite إلى قاعدة مركزية؟

عند أول خادم ثانٍ. تشغيل عدة عمليات على الجهاز نفسه وضع سليم؛ أما توزيع العمل على أجهزة مختلفة فيجعل لكل جهاز كاشاً منفصلاً، فيعود التكرار الذي حاولت التخلّص منه. والترحيل بسيط: sqlite3 captcha_solves.db ".dump" ثم الاستيراد إلى PostgreSQL أو MongoDB.

الخطوة التالية بعد الكاش

الكاش يفترض أن مسار الحل نفسه يعمل. إن كنت لا تزال تبنيه، ابدأ من دليل البدء السريع مع CaptchaAI للحصول على مفتاح الـ API، ثم راجع شرح حل reCAPTCHA v2 عبر الـ API لأن الدالة أعلاه مبنية على المعاملات نفسها.

وبعدها يمكنك تطبيق الجدولين على أنواع تحقق أخرى دون تغيير المخطط:

• حل Cloudflare Turnstile عبر الـ API
• حل GeeTest v3 عبر الـ API