Redis لإدارة TTL لرموز CAPTCHA والتخزين المؤقت

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

النافذة ضيقة بطبيعتها: يبقى رمز reCAPTCHA صالحاً بين 90 و120 ثانية، ورمز Cloudflare Turnstile نحو 300 ثانية. ما تخزّنه أصل سريع التلف، وأي تصميم لا يضع انتهاء الصلاحية في صميمه ينتج أخطاء متقطعة يصعب تتبّعها.

متى يستحق تخزين الرموز العناء ومتى لا

الذاكرة المؤقتة للرموز ليست تحسيناً لكل حالة. تكسب منها في نمط واحد: طلبات كثيرة تستهدف مفتاح الموقع نفسه والصفحة نفسها خلال دقائق قليلة.

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

ونقطة مهمة: يقوم تسعير CaptchaAI على عدد الـ threads المتزامنة لا على عدد عمليات الحل، فالذاكرة المؤقتة تشتري زمن استجابة أقصر لا خصماً على الفاتورة.

اختر TTL بهامش أمان لا بالعمر الكامل

العمر المعلن للرمز سقف، لا مدة تعتمد عليها. اطرح منه ما يغطي زمن الشبكة وطابور المهام وإعادة المحاولة، ثم خزّن الباقي:

الهامش هو الوقت بين سحب الرمز ووصوله إلى الموقع المستهدف؛ وسّعه إذا كان بينهما طابور مهام بطيء. ولاحظ أن قاموس TOKEN_TTLS أدناه يحمل مفاتيح متعددة للتوضيح فقط: لا يدعم CaptchaAI حالياً hCaptcha، فاكتفِ بما تستخدمه فعلاً — إصدارات reCAPTCHA وTurnstile وGeeTest v3 والصور.

دورة حياة الرمز داخل Redis من الطلب إلى الاستهلاك

يمر كل طلب بالمسار نفسه: فحص الذاكرة المؤقتة، ثم القفل، ثم استدعاء الـ API، ثم الكتابة:

Solve Request → Check Redis → Cache Hit?
                    │               │
                    │ No            │ Yes → Return cached token
                    ▼               
              CaptchaAI API
                    │
                    ▼
              Store in Redis (TTL = token_lifetime - safety_margin)
                    │
                    ▼
              Return token

تُكتب القيمة بعد نجاح الحل مباشرة، ويُحسب TTL من عمر الرمز مطروحاً منه الهامش — لا من لحظة إرسال الطلب.

التنفيذ في Python خطوة بخطوة

اضبط الاتصال ومهل TTL

اجعل الاتصال واحداً مشتركاً على مستوى العملية، واقرأ بياناته ومفتاح الـ API من متغيرات البيئة:

import os
import time
import json
import redis
import requests

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

API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# TTLs with safety margin (seconds before actual expiration)
TOKEN_TTLS = {
    "recaptcha_v2": 80,     # Actual: ~120s, cache for 80s
    "recaptcha_v3": 80,
    "hcaptcha": 80,
    "turnstile": 250,       # Actual: ~300s, cache for 250s
}

الخيار decode_responses=True يعيد الرموز نصوصاً جاهزة للحقن في النموذج بدل bytes.

اقرأ الرمز واكتبه بمفتاح يعكس الهدف

يجب أن يعكس المفتاح الهدف بدقة: مفتاح الموقع والصفحة معاً. رمز صفحة تسجيل الدخول لا يصلح لصفحة الدفع:

def cache_key(sitekey, pageurl):
    """Generate Redis key for a specific CAPTCHA target."""
    return f"captcha:token:{sitekey}:{pageurl}"


def get_cached_token(sitekey, pageurl):
    """Pop a cached token from the queue."""
    key = cache_key(sitekey, pageurl)
    token = r.lpop(key)
    if token:
        # Verify TTL still valid on the list
        ttl = r.ttl(key)
        if ttl > 10:  # At least 10 seconds remaining
            return token
    return None


def cache_token(sitekey, pageurl, token, captcha_type="recaptcha_v2"):
    """Push a solved token to the cache with appropriate TTL."""
    key = cache_key(sitekey, pageurl)
    ttl = TOKEN_TTLS.get(captcha_type, 80)
    r.rpush(key, token)
    r.expire(key, ttl)

تسحب الدالة الرمز بـ lpop ثم تتحقق من TTL المتبقي. عتبة العشر ثوانٍ ليست رقماً ثابتاً؛ ارفعها إذا كان بين السحب والاستهلاك طابور بطيء.

احلّ مع فحص الذاكرة المؤقتة وقفل يمنع التكرار

هذه هي الدالة التي يستدعيها كودك فعلياً، وترتيب خطواتها مقصود: الذاكرة المؤقتة، ثم القفل، ثم الـ API، ثم الاستطلاع الدوري للنتيجة:

def solve_recaptcha(sitekey, pageurl, captcha_type="recaptcha_v2"):
    """Solve reCAPTCHA with Redis cache check."""
    # 1. Check cache
    cached = get_cached_token(sitekey, pageurl)
    if cached:
        return {"solution": cached, "source": "cache"}

    # 2. Check if solve is already in progress (dedup)
    lock_key = f"captcha:lock:{sitekey}:{pageurl}"
    if not r.set(lock_key, "1", nx=True, ex=120):
        # Another worker is solving — wait for result
        for _ in range(60):
            time.sleep(2)
            cached = get_cached_token(sitekey, pageurl)
            if cached:
                return {"solution": cached, "source": "cache_wait"}
        return {"error": "TIMEOUT_WAITING_FOR_OTHER_WORKER"}

    try:
        # 3. Solve via 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:
            return {"error": data.get("request")}

        captcha_id = data["request"]

        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:
                token = result["request"]
                cache_token(sitekey, pageurl, token, captcha_type)
                return {"solution": token, "source": "api"}

            if result.get("request") != "CAPCHA_NOT_READY":
                return {"error": result.get("request")}

        return {"error": "TIMEOUT"}
    finally:
        r.delete(lock_key)

يُنشأ القفل بـ nx=True فينجح عامل واحد فقط، ومهلته 120 ثانية تنظّفه تلقائياً إذا توقف العامل قبل الإفراج عنه. أما من لم يحصل عليه فينتظر ظهور النتيجة في الذاكرة المؤقتة بدل إرسال طلب جديد — وهذا وحده يوفّر عشرات الطلبات المتطابقة عند الذروة.

ابنِ مجمّع رموز محلولة مسبقاً

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

import threading


class TokenPool:
    def __init__(self, sitekey, pageurl, pool_size=5, captcha_type="recaptcha_v2"):
        self.sitekey = sitekey
        self.pageurl = pageurl
        self.pool_size = pool_size
        self.captcha_type = captcha_type
        self.pool_key = f"captcha:pool:{sitekey}:{pageurl}"
        self._running = False

    def start(self):
        self._running = True
        thread = threading.Thread(target=self._refill_loop, daemon=True)
        thread.start()

    def stop(self):
        self._running = False

    def _refill_loop(self):
        while self._running:
            current = r.llen(self.pool_key)
            if current < self.pool_size:
                self._solve_and_add()
            time.sleep(2)

    def _solve_and_add(self):
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": self.sitekey,
            "pageurl": self.pageurl,
            "json": 1
        })
        data = resp.json()
        if data.get("status") != 1:
            return

        captcha_id = data["request"]
        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:
                ttl = TOKEN_TTLS.get(self.captcha_type, 80)
                r.rpush(self.pool_key, result["request"])
                r.expire(self.pool_key, ttl)
                return
            if result.get("request") != "CAPCHA_NOT_READY":
                return

    def get_token(self):
        return r.lpop(self.pool_key)


# Usage
pool = TokenPool("6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "https://example.com")
pool.start()

# When you need a token:
token = pool.get_token()

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

التنفيذ في JavaScript مع ioredis

المنطق نفسه في Node.js عبر ioredis وAxios، مع الانتباه إلى أن set تستقبل خياراتها كمعاملات نصية متتابعة:

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

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

const TOKEN_TTLS = { recaptcha_v2: 80, recaptcha_v3: 80, hcaptcha: 80, turnstile: 250 };

function cacheKey(sitekey, pageurl) {
  return `captcha:token:${sitekey}:${pageurl}`;
}

async function getCachedToken(sitekey, pageurl) {
  const key = cacheKey(sitekey, pageurl);
  const token = await redis.lpop(key);
  if (token) {
    const ttl = await redis.ttl(key);
    if (ttl > 10) return token;
  }
  return null;
}

async function solveWithCache(sitekey, pageurl, type = "recaptcha_v2") {
  // Check cache
  const cached = await getCachedToken(sitekey, pageurl);
  if (cached) return { solution: cached, source: "cache" };

  // Dedup lock
  const lockKey = `captcha:lock:${sitekey}:${pageurl}`;
  const locked = await redis.set(lockKey, "1", "NX", "EX", 120);
  if (!locked) {
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 2000));
      const waitCached = await getCachedToken(sitekey, pageurl);
      if (waitCached) return { solution: waitCached, source: "cache_wait" };
    }
    return { error: "TIMEOUT_WAITING" };
  }

  try {
    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) 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) {
        const key = cacheKey(sitekey, pageurl);
        const ttl = TOKEN_TTLS[type] || 80;
        await redis.rpush(key, poll.data.request);
        await redis.expire(key, ttl);
        return { solution: poll.data.request, source: "api" };
      }
      if (poll.data.request !== "CAPCHA_NOT_READY") return { error: poll.data.request };
    }
    return { error: "TIMEOUT" };
  } finally {
    await redis.del(lockKey);
  }
}

لاحظ كتلة finally التي تحذف القفل في كل المسارات، بما فيها مسار الخطأ؛ قفل معلق بلا مالك أسوأ من غياب القفل.

تصميم مفاتيح Redis واصطلاح التسمية

القاعدة العملية: إذا لم تستطع تسمية اللحظة التي يُحذف فيها المفتاح، فأنت تبني تسرّباً في الذاكرة.

راقب معدل الإصابة قبل أن توسّع المجمّع

معدل الإصابة وحده لا يكفي للحكم؛ تابعه إلى جانب عدد المفاتيح النشطة:

def cache_stats():
    info = r.info("stats")
    hits = info.get("keyspace_hits", 0)
    misses = info.get("keyspace_misses", 0)
    total = hits + misses
    return {
        "hit_rate": f"{hits / total * 100:.1f}%" if total else "0%",
        "hits": hits,
        "misses": misses,
        "active_keys": r.dbsize()
    }

معدل إصابة منخفض مع عدد مفاتيح كبير يعني غالباً أن تصميم المفتاح مفرط في التفصيل — ربما أدخلت معرّف الجلسة أو طابعاً زمنياً فصار كل طلب فريداً.

سيناريو تشغيلي: بوابة حجز مواعيد في الخليج

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

الترتيب العملي: شغّل TokenPool قبل فتح النافذة بثلاث دقائق حتى تمتلئ القائمة، واضبط TTL على 80 ثانية لـ reCAPTCHA v2 حتى لا تبدأ النافذة برموز على وشك الانتهاء. مع خطة ADVANCE — 90 دولاراً شهرياً و50 thread — تكفي سعة التزامن لملء المجمّع وخدمة الطلبات المباشرة معاً.

وفي فريق موزّع بين القاهرة والرياض ودبي، وحّد مفاتيح الإحصاءات على UTC؛ مفتاح captcha:stats:{date} بتوقيت محلي يقسم يوم العمل على مفتاحين.

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

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

هل يمكن استخدام الرمز المحلول أكثر من مرة؟

لا. الرمز صالح لاستهلاك واحد، وإعادة إرساله تُرفض حتى داخل نافذة صلاحيته. لهذا تستخدم الأمثلة lpop لا lindex: السحب يزيل الرمز ويمنع عاملَين من أخذ النسخة نفسها.

ماذا أستخدم إذا لم يكن Redis متاحاً في بيئتي؟

أي مخزن يدعم انتهاء الصلاحية على مستوى المفتاح يفي بالغرض: Memcached بمهلته الأصلية، أو جدول بعمود expires_at ومهمة تنظيف دورية. ما تخسره هو الذرّية الجاهزة في SET NX EX.

كم عدد الـ threads الذي أحتاجه مع مجمّع رموز؟

احسب الحد الأعلى للحلول المتزامنة لا مجموعها اليومي. يحتفظ خيط التعبئة بعملية حل نشطة واحدة لكل مجمّع، فعشرة مجمّعات تعني عشرة threads لهذه المهمة وحدها. خطة BASIC — 15 دولاراً شهرياً و5 threads — تكفي لتجربة النمط، وتناسب الأحمالَ المتوازية خطة أعلى مثل ADVANCE.

كيف أمنع تضخم ذاكرة Redis مع الوقت؟

اربط TTL بالمفتاح لحظة إنشائه، فاستدعاء expire بعد rpush قد لا يُنفَّذ إذا توقف العامل بينهما. راقب dbsize، واضبط maxmemory-policy على volatile-ttl ليتخلص Redis أولاً من الرموز الأقرب إلى الانتهاء.

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

• ابدأ مع CaptchaAI في خمس دقائق
• حلّ reCAPTCHA v2 عبر الـ API
• التعامل مع Cloudflare Turnstile
• حلّ GeeTest v3 عبر الـ API

أدلة ذات صلة

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