التخزين المؤقت لرموز CAPTCHA لإعادة استخدامها

ليست كل رموز CAPTCHA قابلة لإعادة الاستخدام. بعضها — مثل ملف cf_clearance أو نتيجة OCR — يبقى صالحاً لدقائق فيكفيك حلٌّ واحد لعشرات الطلبات، وبعضها كرموز reCAPTCHA يُرفض بعد أول استخدام. يبيّن هذا الدليل كيف تميّز ما يستحق التخزين المؤقت، وتضبط نافذة الصلاحية (TTL)، وتُبطل الرمز قبل حقنه منتهياً لتقلّل استدعاءات CaptchaAI API وتخفض التكلفة.

متى ينجح التخزين المؤقت لرموز CAPTCHA ومتى يفشل؟

قبل تفعيل التخزين المؤقت، اسأل أولاً إن كانت الحالة تسمح بإعادة الاستخدام أصلاً:

ما الرموز القابلة للتخزين المؤقت؟ عمر كل نوع CAPTCHA

الخلاصة: ملف cf_clearance ونتائج Image OCR هما الأجدى للتخزين المؤقت، إذ يغطي الحلّ الواحد جلسة كاملة أو صورة متكررة، بينما نافذة reCAPTCHA قصيرة وغالباً للاستخدام مرة واحدة.

مثال عملي: مراقبة أسعار على موقع محمي بـ Cloudflare

تخيّل أداةً لمراقبة الأسعار على متجر محمي بـ Cloudflare Challenge تستطلع عشرات الصفحات باستمرار. بلا تخزين مؤقت يستهلك كل طلب حلاً وخيط معالجة (Thread) مستقلاً. أما بحلّ cf_clearance مرة واحدة وإعادة استخدامه طوال الجلسة (15–30 دقيقة) فينخفض عدد عمليات الحل بأكثر من 90%؛ عندها قد تكفي باقة BASIC (15 دولاراً شهرياً، 5 خيوط) بدل ADVANCE (90 دولاراً، 50 خيطاً)، لأن الفوترة على الخيوط المتزامنة لا على كل حل.

تنفيذ التخزين المؤقت عملياً

اختر الطبقة بحسب حجم تشغيلك: ذاكرة محلية، فـ Redis للأنظمة الموزّعة، فالحلّ المسبق.

ذاكرة داخل العملية باستخدام Python

أبسط صورة للتخزين المؤقت خريطةٌ في الذاكرة تربط نوع الطلب ومعاملاته بالرمز وتاريخ انتهائه. لاحظ خريطة TTL_MAP: نافذة كل نوع أقصر من عمره الحقيقي كهامش أمان.

import time
import hashlib
from typing import Optional
import requests

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"


class TokenCache:
    def __init__(self):
        self.cache = {}

    def _key(self, method: str, params: dict) -> str:
        # Cache key from method + stable params
        stable = {k: v for k, v in sorted(params.items())
                  if k not in ("key", "json")}
        raw = f"{method}:{stable}"
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

    def get(self, method: str, params: dict) -> Optional[str]:
        key = self._key(method, params)
        entry = self.cache.get(key)
        if entry and entry["expires_at"] > time.time():
            print(f"Cache HIT: {key}")
            return entry["token"]
        if entry:
            del self.cache[key]
        return None

    def set(self, method: str, params: dict, token: str, ttl: int):
        key = self._key(method, params)
        self.cache[key] = {
            "token": token,
            "expires_at": time.time() + ttl,
        }
        print(f"Cached: {key} (TTL: {ttl}s)")

    def invalidate(self, method: str, params: dict):
        key = self._key(method, params)
        self.cache.pop(key, None)

    def cleanup(self):
        now = time.time()
        expired = [k for k, v in self.cache.items() if v["expires_at"] <= now]
        for k in expired:
            del self.cache[k]


# TTL per CAPTCHA type
TTL_MAP = {
    "userrecaptcha": 100,       # 120s lifetime, 20s safety margin
    "turnstile": 240,           # 300s lifetime, 60s margin
    "cloudflare_challenge": 900,# 15min lifetime, 5min margin
    "base64": 86400,            # OCR result never expires — cache 24h
}


class CachedSolver:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.cache = TokenCache()

    def solve(self, method: str, params: dict) -> str:
        # Check cache first
        cached = self.cache.get(method, params)
        if cached:
            return cached

        # Solve via API
        token = self._api_solve(method, params)
        ttl = TTL_MAP.get(method, 60)
        self.cache.set(method, params, token, ttl)
        return token

    def _api_solve(self, method: str, params: dict) -> str:
        data = {
            "key": self.api_key,
            "method": method,
            "json": 1,
            **params
        }

        resp = requests.post(SUBMIT_URL, data=data, timeout=15)
        result = resp.json()

        if result.get("status") != 1:
            raise Exception(result.get("error_text", result.get("request")))

        task_id = result["request"]
        return self._poll(task_id)

    def _poll(self, task_id: str, max_wait: int = 120) -> str:
        elapsed = 0
        while elapsed < max_wait:
            time.sleep(5)
            elapsed += 5

            resp = requests.get(RESULT_URL, params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            }, timeout=10)
            result = resp.json()

            if result.get("status") == 1:
                return result["request"]
            if result.get("request") == "CAPCHA_NOT_READY":
                continue

            raise Exception(result.get("error_text", result.get("request")))

        raise Exception(f"Timeout: {task_id}")


# Usage
solver = CachedSolver(api_key="YOUR_API_KEY")

# First call — hits API
token1 = solver.solve("turnstile", {
    "sitekey": "0x4AAAA-SITEKEY",
    "pageurl": "https://example.com"
})
print(f"Token 1: {token1[:40]}...")

# Second call within TTL — cache hit, no API call
token2 = solver.solve("turnstile", {
    "sitekey": "0x4AAAA-SITEKEY",
    "pageurl": "https://example.com"
})
print(f"Token 2: {token2[:40]}...")
print(f"Same token: {token1 == token2}")  # True

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

المنطق ذاته في Node.js عبر Map وcrypto لاشتقاق مفتاح مستقر:

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

const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

const TTL_MAP = {
  userrecaptcha: 100,
  turnstile: 240,
  cloudflare_challenge: 900,
  base64: 86400,
};

class TokenCache {
  constructor() {
    this.cache = new Map();
  }

  _key(method, params) {
    const stable = Object.entries(params)
      .filter(([k]) => k !== "key" && k !== "json")
      .sort(([a], [b]) => a.localeCompare(b))
      .map(([k, v]) => `${k}=${v}`)
      .join("&");
    return crypto.createHash("sha256").update(`${method}:${stable}`).digest("hex").slice(0, 16);
  }

  get(method, params) {
    const key = this._key(method, params);
    const entry = this.cache.get(key);
    if (entry && entry.expiresAt > Date.now()) {
      console.log(`Cache HIT: ${key}`);
      return entry.token;
    }
    if (entry) this.cache.delete(key);
    return null;
  }

  set(method, params, token, ttlMs) {
    const key = this._key(method, params);
    this.cache.set(key, { token, expiresAt: Date.now() + ttlMs });
    console.log(`Cached: ${key} (TTL: ${ttlMs / 1000}s)`);
  }
}

class CachedSolver {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.cache = new TokenCache();
  }

  async solve(method, params) {
    const cached = this.cache.get(method, params);
    if (cached) return cached;

    const token = await this._apiSolve(method, params);
    const ttl = (TTL_MAP[method] || 60) * 1000;
    this.cache.set(method, params, token, ttl);
    return token;
  }

  async _apiSolve(method, params) {
    const resp = await axios.post(SUBMIT_URL, null, {
      params: { key: this.apiKey, method, json: 1, ...params },
      timeout: 15000,
    });

    if (resp.data.status !== 1) {
      throw new Error(resp.data.error_text || resp.data.request);
    }

    return this._poll(resp.data.request);
  }

  async _poll(taskId, maxWait = 120000) {
    let elapsed = 0;
    while (elapsed < maxWait) {
      await new Promise((r) => setTimeout(r, 5000));
      elapsed += 5000;

      const resp = await axios.get(RESULT_URL, {
        params: { key: this.apiKey, action: "get", id: taskId, json: 1 },
        timeout: 10000,
      });

      if (resp.data.status === 1) return resp.data.request;
      if (resp.data.request === "CAPCHA_NOT_READY") continue;

      throw new Error(resp.data.error_text || resp.data.request);
    }
    throw new Error("Timeout");
  }
}

// Usage
(async () => {
  const solver = new CachedSolver("YOUR_API_KEY");

  const token1 = await solver.solve("turnstile", {
    sitekey: "0x4AAAA-SITEKEY",
    pageurl: "https://example.com",
  });
  console.log(`Token 1: ${token1.slice(0, 40)}...`);

  const token2 = await solver.solve("turnstile", {
    sitekey: "0x4AAAA-SITEKEY",
    pageurl: "https://example.com",
  });
  console.log(`Token 2: ${token2.slice(0, 40)}...`);
  console.log(`Same token: ${token1 === token2}`);
})();

مشاركة الذاكرة عبر Redis في الأنظمة الموزّعة

حين يعمل أكثر من worker على عمليات متعددة لا تكفي الذاكرة المحلية. استخدم Redis طبقةً مشتركة تدير انتهاء TTL نيابةً عنك:

import redis
import json

r = redis.Redis(host="localhost", port=6379, db=0)

def cache_token(method, params, token, ttl):
    key = f"captcha:{method}:{hash(frozenset(params.items()))}"
    r.setex(key, ttl, token)

def get_cached_token(method, params):
    key = f"captcha:{method}:{hash(frozenset(params.items()))}"
    return r.get(key)

يُزيل Redis الرموز تلقائياً عند انتهاء مهلتها، ويتيح الذاكرة المؤقتة لكل العمال.

الحلّ المسبق: مخزون جاهز من الرموز

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

from collections import deque
from threading import Thread

token_buffer = deque(maxlen=5)

def pre_solve_worker(solver, method, params):
    while True:
        if len(token_buffer) < 3:
            try:
                token = solver._api_solve(method, params)
                ttl = TTL_MAP.get(method, 60)
                token_buffer.append({
                    "token": token,
                    "expires_at": time.time() + ttl
                })
            except Exception as e:
                print(f"Pre-solve failed: {e}")
        time.sleep(2)

# Start pre-solver in background
thread = Thread(
    target=pre_solve_worker,
    args=(solver, "turnstile", {"sitekey": "0x4AAAA-KEY", "pageurl": "https://example.com"}),
    daemon=True
)
thread.start()

# Consume pre-solved tokens
def get_presolved():
    while token_buffer:
        entry = token_buffer.popleft()
        if entry["expires_at"] > time.time():
            return entry["token"]
    return None

قواعد إبطال الذاكرة المؤقتة لرموز CAPTCHA

أسقِط الرمز من الذاكرة في هذه الحالات:

• رفض الموقع المستهدف للرمز — أبطله وأعد الحل فوراً.
• انتهاء مهلة TTL — يُزال تلقائياً من الذاكرة المؤقتة.
• تغيير الخادم الوسيط (البروكسي) — أبطل رموز Cloudflare المرتبطة بعنوان IP.
• تعديل إعدادات CAPTCHA على الموقع — امسح كل الرموز المخزّنة لذلك الموقع.

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

أسئلة شائعة

ما القيمة المثالية لـ TTL عند تخزين رمز Turnstile؟

اجعلها أقصر من العمر الحقيقي بهامش أمان. رمز Turnstile يعيش نحو 300 ثانية، وقيمة نحو 240 ثانية تكفي لإرساله قبل انتهائه.

هل يمكن مشاركة الرمز نفسه بين عدة خوادم وسيطة؟

ليس مع رموز Cloudflare، لأن cf_clearance مرتبط بعنوان IP الذي حُلّ من خلاله. إذا بدّلت البروكسي فأبطل رموزه وأعد الحل من العنوان الجديد.

كيف أعرف أن موقعاً يقبل إعادة استخدام رمز reCAPTCHA؟

اختبرها عملياً: أرسل الرمز نفسه مرتين. إن قُبل الإرسال الثاني فالموقع يسمح بإعادة الاستخدام، وإن رُفض فعامله كاستخدام واحد وأوقف تخزينه.

هل يخفض التخزين المؤقت معدل النجاح؟

لا؛ فهو إعادة استخدام لرمز صالح لم تنتهِ مدته. الخطر الحقيقي هو حقن رمز منتهي الصلاحية، وهذا ما يخفض معدل النجاح لا التخزين نفسه.

خفّض تكاليف حل CAPTCHA مع CaptchaAI

ابدأ بتخزين رموز CAPTCHA وإعادة استخدامها عبر captchaai.com، وادفع مقابل الخيوط المتزامنة لا مقابل كل عملية حل.

أدلة ذات صلة

• بناء خطوط معالجة CAPTCHA من جهة العميل مع CaptchaAI
• إنشاء قائمة انتظار لحل CAPTCHA في Python