منع تكرار طلبات حل CAPTCHA باستخدام أقفال قاعدة البيانات

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

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

يقدّم هذا الدليل مسارين عمليين للتنسيق، تختار بينهما حسب بنيتك التحتية:

• قفل عبر Redis: الأسرع والأنسب للأنظمة الموزّعة التي تملك طبقة تخزين في الذاكرة.
• قفل استشاري في PostgreSQL: بلا خدمة إضافية، يعتمد على قاعدة بياناتك الحالية.

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

لماذا تتكرر طلبات حل CAPTCHA؟

قبل بناء الحل، من المفيد تحديد المصادر الأربعة التي تولّد معظم الطلبات المكررة في أنظمة الإنتاج:

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

تصميم مفتاح إلغاء التكرار

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

import hashlib


def dedup_key(method, sitekey, pageurl):
    """Generate a deduplication key for a CAPTCHA solve request."""
    raw = f"{method}:{sitekey}:{pageurl}"
    return f"captcha:dedup:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"

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

لاحظ أن reCAPTCHA v3 يضيف حقل action؛ إهماله يعني أن طلبين بأفعال مختلفة على نفس الصفحة سيتشاركان نتيجة واحدة خاطئة. أما كابتشا الصورة فتحتاج إلى تجزئة محتوى الصورة نفسها، لأن الصور المتطابقة تُنتج النص ذاته.

إلغاء التكرار عبر Redis

يوفّر Redis أسرع طريقة لبناء هذه الطبقة، لأنه يخزّن حالة كل مفتاح في الذاكرة ويتيح ضبط مهلة صلاحية (TTL) لكل حالة. الفكرة: عند وصول طلب، نفحص المفتاح أولاً. إذا كان "قيد الحل" ننتظر العامل الآخر؛ وإذا كان "محلولاً" نعيد الرمز مباشرة؛ وإلا نبدأ الحل ونعلّم المفتاح.

تطبيق Python

import os
import time
import json
import hashlib
import redis
import requests

r = redis.Redis(
    host=os.environ.get("REDIS_HOST", "localhost"),
    port=int(os.environ.get("REDIS_PORT", 6379)),
    decode_responses=True
)

API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# Dedup window: how long to consider a request "in progress"
DEDUP_TTL = 180  # seconds


def dedup_key(method, sitekey, pageurl, extra=""):
    raw = f"{method}:{sitekey}:{pageurl}:{extra}"
    return f"captcha:dedup:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"


def solve_with_dedup(sitekey, pageurl, method="userrecaptcha"):
    key = dedup_key(method, sitekey, pageurl)

    # Check if this request is already being solved
    existing = r.get(key)
    if existing:
        state = json.loads(existing)
        if state["status"] == "solving":
            # Wait for the result
            return wait_for_result(key)
        elif state["status"] == "solved":
            return {"solution": state["solution"], "source": "dedup_cache"}
        elif state["status"] == "error":
            pass  # Allow retry on error

    # Mark as solving
    r.set(key, json.dumps({"status": "solving", "started": time.time()}), ex=DEDUP_TTL)

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

    if data.get("status") != 1:
        r.set(key, json.dumps({"status": "error", "error": data.get("request")}), ex=30)
        return {"error": data.get("request")}

    captcha_id = data["request"]

    # Poll for result
    for _ in range(60):
        time.sleep(5)
        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:
            solution = result["request"]
            # Cache the result for other workers (short TTL since tokens expire)
            r.set(key, json.dumps({
                "status": "solved",
                "solution": solution,
                "solved_at": time.time()
            }), ex=60)  # Cache result for 60 seconds
            return {"solution": solution, "source": "api"}

        if result.get("request") != "CAPCHA_NOT_READY":
            r.set(key, json.dumps({
                "status": "error", "error": result.get("request")
            }), ex=30)
            return {"error": result.get("request")}

    r.set(key, json.dumps({"status": "error", "error": "TIMEOUT"}), ex=30)
    return {"error": "TIMEOUT"}


def wait_for_result(key, timeout=120):
    """Wait for another worker to finish solving."""
    start = time.time()
    while time.time() - start < timeout:
        data = r.get(key)
        if data:
            state = json.loads(data)
            if state["status"] == "solved":
                return {"solution": state["solution"], "source": "dedup_wait"}
            if state["status"] == "error":
                return {"error": state.get("error", "UNKNOWN")}
        time.sleep(2)
    return {"error": "DEDUP_WAIT_TIMEOUT"}

الفكرة الجوهرية هنا هي أن العامل الأول فقط يرسل الطلب إلى CaptchaAI، بينما تنتظر بقية العمال النتيجة عبر wait_for_result. ولاحظ أن مهلة النتيجة المخزّنة قصيرة (60 ثانية) لأن الرموز الناتجة عن reCAPTCHA تنتهي صلاحيتها بسرعة، فلا فائدة من الاحتفاظ برمز منتهٍ.

تطبيق JavaScript

نفس المنطق ينطبق في بيئة Node.js باستخدام ioredis. تبقى الخطوات متطابقة: فحص المفتاح، ثم الانتظار أو الإرسال، ثم الاستطلاع الدوري للنتيجة.

const Redis = require("ioredis");
const axios = require("axios");
const crypto = require("crypto");

const redis = new Redis(process.env.REDIS_URL || "redis://localhost:6379");
const API_KEY = process.env.CAPTCHAAI_API_KEY;
const DEDUP_TTL = 180;

function dedupKey(method, sitekey, pageurl) {
  const raw = `${method}:${sitekey}:${pageurl}`;
  const hash = crypto.createHash("sha256").update(raw).digest("hex").slice(0, 16);
  return `captcha:dedup:${hash}`;
}

async function solveWithDedup(sitekey, pageurl, method = "userrecaptcha") {
  const key = dedupKey(method, sitekey, pageurl);

  // Check existing
  const existing = await redis.get(key);
  if (existing) {
    const state = JSON.parse(existing);
    if (state.status === "solving") return await waitForResult(key);
    if (state.status === "solved") return { solution: state.solution, source: "dedup_cache" };
  }

  // Mark as solving
  await redis.set(key, JSON.stringify({ status: "solving", started: Date.now() }), "EX", DEDUP_TTL);

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

  if (submit.data.status !== 1) {
    await redis.set(key, JSON.stringify({ status: "error", error: submit.data.request }), "EX", 30);
    return { error: submit.data.request };
  }

  const captchaId = submit.data.request;

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    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) {
      await redis.set(key, JSON.stringify({ status: "solved", solution: poll.data.request }), "EX", 60);
      return { solution: poll.data.request, source: "api" };
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      await redis.set(key, JSON.stringify({ status: "error", error: poll.data.request }), "EX", 30);
      return { error: poll.data.request };
    }
  }

  await redis.set(key, JSON.stringify({ status: "error", error: "TIMEOUT" }), "EX", 30);
  return { error: "TIMEOUT" };
}

async function waitForResult(key, timeout = 120000) {
  const start = Date.now();
  while (Date.now() - start < timeout) {
    const data = await redis.get(key);
    if (data) {
      const state = JSON.parse(data);
      if (state.status === "solved") return { solution: state.solution, source: "dedup_wait" };
      if (state.status === "error") return { error: state.error };
    }
    await new Promise((r) => setTimeout(r, 2000));
  }
  return { error: "DEDUP_WAIT_TIMEOUT" };
}

البديل: أقفال PostgreSQL الاستشارية

لا يملك كل فريق طبقة Redis جاهزة. إذا كانت قاعدة بياناتك الأساسية هي PostgreSQL، يمكنك تحقيق نفس التنسيق عبر الأقفال الاستشارية (advisory locks) من دون خدمة إضافية. الفكرة: يحوّل كل طلب مفتاحه إلى رقم قفل، ويحاول العامل الأول الاستحواذ عليه بلا انتظار؛ فإذا فشل، فهذا يعني أن عاملاً آخر يحل الطلب، فينتظر ثم يقرأ النتيجة من الجدول المخزّن.

import psycopg2


def solve_with_pg_dedup(conn, sitekey, pageurl):
    """Use PostgreSQL advisory locks for deduplication."""
    # Generate a numeric lock key from the dedup key
    lock_id = hash(f"{sitekey}:{pageurl}") & 0x7FFFFFFF

    cursor = conn.cursor()

    # Try to acquire advisory lock (non-blocking)
    cursor.execute("SELECT pg_try_advisory_lock(%s)", (lock_id,))
    acquired = cursor.fetchone()[0]

    if not acquired:
        # Another worker is solving — wait for result
        cursor.execute("SELECT pg_advisory_lock(%s)", (lock_id,))
        # Lock acquired means other worker finished — check cache
        cursor.execute(
            "SELECT solution FROM captcha_cache "
            "WHERE sitekey = %s AND pageurl = %s "
            "AND created_at > NOW() - INTERVAL '60 seconds'",
            (sitekey, pageurl)
        )
        row = cursor.fetchone()
        cursor.execute("SELECT pg_advisory_unlock(%s)", (lock_id,))
        if row:
            return {"solution": row[0], "source": "pg_cache"}
        return {"error": "NO_CACHED_RESULT"}

    try:
        # Solve the CAPTCHA
        solution = solve_via_api(sitekey, pageurl)
        if solution:
            cursor.execute(
                "INSERT INTO captcha_cache (sitekey, pageurl, solution) "
                "VALUES (%s, %s, %s)",
                (sitekey, pageurl, solution)
            )
            conn.commit()
        return {"solution": solution} if solution else {"error": "SOLVE_FAILED"}
    finally:
        cursor.execute("SELECT pg_advisory_unlock(%s)", (lock_id,))

هذا النهج يمنحك تنسيقاً قوياً بضمانات قاعدة البيانات نفسها، مع الانتباه إلى أن pg_advisory_lock هنا حاجب (blocking)، لذا اضبط مهلة معقولة على مستوى الاتصال حتى لا يعلق عامل إلى الأبد إذا تعطّل صاحب القفل.

قياس فعالية إلغاء التكرار

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

def track_dedup_stats(source):
    """Increment counters for dedup tracking."""
    today = time.strftime("%Y-%m-%d")
    r.hincrby(f"dedup:stats:{today}", source, 1)
    r.expire(f"dedup:stats:{today}", 7 * 86400)


def get_dedup_report():
    today = time.strftime("%Y-%m-%d")
    stats = r.hgetall(f"dedup:stats:{today}")
    total = sum(int(v) for v in stats.values())
    saved = int(stats.get("dedup_cache", 0)) + int(stats.get("dedup_wait", 0))
    return {
        "total_requests": total,
        "deduplicated": saved,
        "savings_pct": f"{saved / total * 100:.1f}%" if total else "0%",
        "breakdown": stats
    }

هذا التتبّع مهم لسبب اقتصادي مباشر: CaptchaAI يحاسب على أساس عدد الـ Threads المتزامنة، لا على أساس عدد الحلول. كل طلب مكرّر توقفه هو Thread يتحرّر ليخدم طلباً حقيقياً آخر. فريق يعمل على خطة ADVANCE ($90 شهرياً، 50 Thread) قد يجد أن إلغاء التكرار يمنحه سعة فعلية أكبر بكثير من دون ترقية الخطة، لأن الـ Threads لم تعد تُهدر على نسخ متطابقة.

معالجة المشكلات الشائعة

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

هل يقلّل إلغاء التكرار عدد الـ Threads التي أحتاجها فعلاً؟

نعم بشكل غير مباشر. بما أن CaptchaAI يحاسب على الـ Threads المتزامنة، فإن كل طلب مكرّر تمنعه يحرّر Thread لطلب مختلف. عند حجم كبير مع عمال متوازين على نفس الهدف، قد يعني ذلك البقاء على خطة أدنى بدل الترقية.

ما المدة المثالية لصلاحية مفتاح "قيد الحل" (TTL)؟

اجعلها أطول قليلاً من أقصى وقت حل متوقّع للنوع الذي تتعامل معه، وهو ما تعكسه القيمة الافتراضية 180 ثانية في الأمثلة. إن كانت قصيرة جداً فسيبدأ عامل ثانٍ حلاً موازياً قبل انتهاء الأول؛ وإن كانت طويلة جداً فسيعلق المنتظِرون بعد تعطّل صاحب القفل.

هل يعمل إلغاء التكرار مع reCAPTCHA v3 المعتمد على action؟

نعم، لكن يجب تضمين حقل action ضمن مكوّنات المفتاح. طلبان بنفس sitekey وpageurl لكن بفعلين مختلفين هما طلبان مختلفان فعلياً، وتجاهل action سيعيد لأحدهما رمزاً لا يخصّه.

ماذا يحدث إذا تعطّل العامل الذي يحمل القفل؟

في نهج Redis تنتهي صلاحية حالة solving تلقائياً بعد المهلة المحددة، فيتولّى عامل آخر الحل. أما في نهج PostgreSQL فيُحرَّر القفل الاستشاري عند إغلاق الجلسة، لذا اضبط مهلة على مستوى الاتصال حتى لا ينتظر المنتظِرون بلا نهاية.

خطوات تالية

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

أدلة ذات صلة

• دليل البدء السريع لأول رمز محلول
• تصحيح استدعاءات CAPTCHA API عبر Charles Proxy