إدارة حالة جلسة CAPTCHA عبر العمال الموزعين

تنجح مشاركة حالة الجلسة بين العمّال الموزّعين حين يقرأ الجميع من مخزن مركزي واحد — عادةً Redis — بدل أن يحتفظ كل عامل بجلسته المعزولة. الفكرة باختصار: خزّن ملفات تعريف الارتباط والرموز المحلولة وتعيينات البروكسي في مكان مشترك، فيرى الموقع سلوكاً متماسكاً لا عشرات الجلسات المتضاربة. في هذا الدليل نبني هذا المخزن خطوة بخطوة فوق CaptchaAI بلغتي Python وNode.js، مع قفل الجلسة وتجمّع البروكسي والتخزين المؤقت للرموز.

متى تتعارض الجلسات في نظام موزّع؟

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

Worker 1 → Login → Solve CAPTCHA → Get cookie A → Submit form ✅
Worker 2 → New session → Solve CAPTCHA → Get cookie B → Submit form ✅
Worker 3 → Reuse cookie A? → Cookie expired → Solve CAPTCHA → Fail ❌

من دون حالة مشتركة، يهدر العمّال حلولاً على جلسات منتهية الصلاحية وينتجون سلوكاً يسهل على المواقع رصده. الحل ليس تقليل عدد العمّال، بل توحيد الحالة التي يعتمدون عليها.

ما الذي تتضمّنه حالة الجلسة المشتركة؟

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

بنية تخزين حالة الجلسة في Redis

نضع Redis في المنتصف كمخزن حالة واحد يقرأ منه جميع العمّال ويكتبون فيه. لكل نوع بيانات بنية Redis الأنسب: تجزئة (Hash) لملفات تعريف الارتباط، وقائمة (List) للرموز المحلولة مسبقاً، ومجموعة (Set) لتجمّع البروكسي، وسلسلة (String) لأقفال الجلسة. هكذا يتحوّل التوزيع إلى عمليات قراءة وكتابة بسيطة:

┌──────────────────────────────────────┐
│          Session State Store          │
│              (Redis)                  │
│                                      │
│  cookies:{domain} → Hash             │
│  tokens:{sitekey} → List             │
│  proxies:pool → Set                  │
│  locks:{domain}:{worker} → String    │
└─────┬──────────┬──────────┬──────────┘
      │          │          │
  ┌───▼───┐  ┌──▼────┐  ┌──▼────┐
  │Worker1│  │Worker2│  │Worker3│
  └───────┘  └───────┘  └───────┘

مفاتيح Redis منظّمة حسب النطاق ومفتاح الموقع، فيجلب كل عامل حالة نطاقه دون تعارض مع نطاق آخر.

بناء المخزن المشترك في Python

نبني الطبقات الثلاث بالترتيب: مخزن الجلسة الذي يغلّف عمليات Redis، ثم العامل الذي يقرأ الحالة ويكتبها حول نداء CaptchaAI، وأخيراً تجمّع البروكسي الذي يمنع تعارض عناوين IP.

مخزن الجلسة

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

import os
import json
import time
import redis
import requests
from datetime import datetime, timezone

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"]


class SessionStore:
    """Shared session state across distributed workers."""

    def __init__(self, domain):
        self.domain = domain
        self.cookie_key = f"session:cookies:{domain}"
        self.token_key = f"session:tokens:{domain}"

    def save_cookies(self, cookies, ttl=1800):
        """Store cookies from a successful session."""
        cookie_data = {name: value for name, value in cookies.items()}
        r.hset(self.cookie_key, mapping=cookie_data)
        r.expire(self.cookie_key, ttl)

    def get_cookies(self):
        """Retrieve shared cookies."""
        cookies = r.hgetall(self.cookie_key)
        return cookies if cookies else None

    def save_token(self, sitekey, token, ttl=80):
        """Store a solved CAPTCHA token."""
        key = f"{self.token_key}:{sitekey}"
        r.rpush(key, token)
        r.expire(key, ttl)

    def get_token(self, sitekey):
        """Pop a cached CAPTCHA token."""
        key = f"{self.token_key}:{sitekey}"
        return r.lpop(key)

    def acquire_session_lock(self, worker_id, ttl=300):
        """Ensure only one worker manages the session at a time."""
        lock_key = f"session:lock:{self.domain}"
        return r.set(lock_key, worker_id, nx=True, ex=ttl)

    def release_session_lock(self, worker_id):
        """Release session lock if this worker holds it."""
        lock_key = f"session:lock:{self.domain}"
        current = r.get(lock_key)
        if current == worker_id:
            r.delete(lock_key)

مهلة الرمز الافتراضية هنا 80 ثانية فقط، فالرموز المحلولة قصيرة العمر ومن الأسلم استهلاكها بسرعة.

عامل يقرأ الحالة المشتركة ويكتبها

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

class CaptchaWorker:
    def __init__(self, worker_id, domain):
        self.worker_id = worker_id
        self.store = SessionStore(domain)
        self.session = requests.Session()

    def setup_session(self):
        """Load shared cookies into this worker's session."""
        cookies = self.store.get_cookies()
        if cookies:
            for name, value in cookies.items():
                self.session.cookies.set(name, value)
            return True
        return False

    def solve_captcha(self, sitekey, pageurl):
        """Solve with token cache and session sharing."""
        # Check for cached token
        cached = self.store.get_token(sitekey)
        if cached:
            return {"solution": cached, "source": "cache"}

        # 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"]
                self.store.save_token(sitekey, token)
                return {"solution": token, "source": "api"}

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

        return {"error": "TIMEOUT"}

    def process_page(self, url, sitekey):
        """Full workflow: setup session → solve CAPTCHA → submit."""
        # Load shared session
        self.setup_session()

        # Solve CAPTCHA
        result = self.solve_captcha(sitekey, url)
        if "error" in result:
            return result

        # Submit form with token
        response = self.session.post(url, data={
            "g-recaptcha-response": result["solution"]
        })

        # Share resulting cookies
        self.store.save_cookies(dict(self.session.cookies))

        return {"status": response.status_code, "source": result["source"]}

توزيع البروكسي بين العمّال

يمنع تجمّع البروكسي أن يستخدم عاملان عنوان IP نفسه في وقت واحد. يسحب كل عامل بروكسي من المجموعة عند البدء ويعيده عند الانتهاء، فيبقى ارتباط الجلسة بعنوان IP متّسقاً طوال المهمة.

class ProxyPool:
    """Distribute proxies across workers to avoid IP conflicts."""

    def __init__(self, proxies):
        self.pool_key = "session:proxy_pool"
        self.assigned_key = "session:proxy_assigned"
        # Initialize pool
        for proxy in proxies:
            r.sadd(self.pool_key, proxy)

    def acquire_proxy(self, worker_id, ttl=600):
        """Assign an unused proxy to a worker."""
        # Check if worker already has one
        existing = r.hget(self.assigned_key, worker_id)
        if existing:
            return existing

        # Pop from available pool
        proxy = r.spop(self.pool_key)
        if proxy:
            r.hset(self.assigned_key, worker_id, proxy)
            r.expire(self.assigned_key, ttl)
            return proxy
        return None

    def release_proxy(self, worker_id):
        """Return proxy to the pool."""
        proxy = r.hget(self.assigned_key, worker_id)
        if proxy:
            r.sadd(self.pool_key, proxy)
            r.hdel(self.assigned_key, worker_id)

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

إن كان مكدّس فريقك قائماً على JavaScript، فالبنية نفسها تنتقل مباشرة إلى ioredis وaxios. مخزن الجلسة يغلّف عمليات Redis ذاتها، ودالة workerSolve تتحقق من الذاكرة المؤقتة قبل إرسال الطلب إلى CaptchaAI:

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;

class SessionStore {
  constructor(domain) {
    this.domain = domain;
    this.cookieKey = `session:cookies:${domain}`;
    this.tokenKey = `session:tokens:${domain}`;
  }

  async saveCookies(cookies, ttl = 1800) {
    const entries = Object.entries(cookies).flat();
    if (entries.length > 0) {
      await redis.hset(this.cookieKey, ...entries);
      await redis.expire(this.cookieKey, ttl);
    }
  }

  async getCookies() {
    return await redis.hgetall(this.cookieKey);
  }

  async saveToken(sitekey, token, ttl = 80) {
    const key = `${this.tokenKey}:${sitekey}`;
    await redis.rpush(key, token);
    await redis.expire(key, ttl);
  }

  async getToken(sitekey) {
    return await redis.lpop(`${this.tokenKey}:${sitekey}`);
  }

  async acquireLock(workerId, ttl = 300) {
    const result = await redis.set(`session:lock:${this.domain}`, workerId, "NX", "EX", ttl);
    return result === "OK";
  }

  async releaseLock(workerId) {
    const current = await redis.get(`session:lock:${this.domain}`);
    if (current === workerId) await redis.del(`session:lock:${this.domain}`);
  }
}

async function workerSolve(store, sitekey, pageurl) {
  const cached = await store.getToken(sitekey);
  if (cached) return { solution: cached, source: "cache" };

  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) {
      await store.saveToken(sitekey, poll.data.request);
      return { solution: poll.data.request, source: "api" };
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") return { error: poll.data.request };
  }
  return { error: "TIMEOUT" };
}

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

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

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

أنماط إدارة حالة الجلسة الموزّعة

اختر النمط بحسب طبيعة الموقع المستهدف ومستوى الإنتاجية المطلوب؛ والأنماط الأربعة قابلة للدمج في النظام نفسه:

معالجة الأخطاء الشائعة

معظم أعطال الأنظمة الموزّعة تعود إلى حالة غير متزامنة أو مهلات صلاحية غير مضبوطة. يربط الجدول التالي كل عَرَض بسببه والإجراء المناسب:

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

ما أنواع CAPTCHA التي تدعمها CaptchaAI في سير عمل موزّع كهذا؟

تحلّ CaptchaAI reCAPTCHA v2 وv3 (بما فيها Enterprise)، وCloudflare Turnstile وCloudflare Challenge، وGeeTest v3، واختبارات الصورة/OCR والشبكة الصورية، وBLS. أما hCaptcha وFunCaptcha (Arkose Labs) فغير مدعومة حالياً، ودعم GeeTest v4 «قادم قريباً» وليس متاحاً بعد. أنواع CaptchaFox وFriendly Captcha وLemin متاحة في مرحلة بيتا. والنمط الموزّع هنا ينطبق على أي نوع مدعوم؛ ما يتغيّر هو حقل method فقط.

هل يمكن إعادة استخدام رمز CAPTCHA نفسه على أكثر من عامل؟

لا في العادة. رمز CAPTCHA المحلول مصمَّم للاستخدام مرة واحدة وضمن سياق الصفحة نفسها، ولذلك تكون مهلته في المخزن قصيرة جداً (نحو 80 ثانية). تجمّع الرموز يفيد حين تُحلّ عدة رموز مسبقاً لمفتاح الموقع نفسه وتُوزَّع، لا حين يشترك عاملان في رمز واحد.

لماذا Redis تحديداً لتخزين حالة الجلسة؟

لأن Redis يوفّر بنى بيانات جاهزة (Hash وList وSet وString) وعمليات ذرّية ومهلات صلاحية مدمجة، وهي بالضبط ما تحتاجه مشاركة الحالة: قفل ذرّي عبر SET NX، ومهلة تلقائية للرموز، وتجمّع بروكسي عبر SPOP. يمكن استبداله بمخزن آخر، لكنه يختصر معظم منطق التزامن.

كيف أمنع تعارض عاملَين على تسجيل الدخول نفسه؟

استخدم قفل الجلسة: أول عامل ينجح في acquire_session_lock عبر SET NX يتولّى تسجيل الدخول وتخزين ملفات تعريف الارتباط، والبقية ينتظرون ثم يقرأون الجلسة الجاهزة. ومهلة القفل (300 ثانية افتراضياً) تحرّره تلقائياً إذا تعطّل حامله.

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

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

أدلة ذات صلة

• استخراج بيانات الويب مع CAPTCHA في Python
• استخراج بيانات الويب مع CAPTCHA في Node.js