أنماط معمارية لحل CAPTCHA على نطاق واسع

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

• تجمع العمال — للأحجام المتوسطة حتى نحو 1000 طلب في الساعة.
• خط الأنابيب بقائمة انتظار — لآلاف الطلبات مع ضغط خلفي منضبط.
• قاطع الدائرة — لحماية النظام عند تدهور واجهة API.
• مخزن الرموز المؤقت — لاستجابة فورية في المسارات الحساسة لزمن الانتظار.
• تجاوز الفشل متعدد المزوّدين — لأعلى درجات الاستمرارية.

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

كيف تختار نمط حل CAPTCHA على نطاق واسع؟

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

• الحجم — كم طلبًا في الساعة عند الذروة؟
• الضغط الخلفي — هل تتراكم الطلبات أسرع من قدرتك على معالجتها؟
• استقرار المزوّد — كم مرة تتدهور واجهة API أو ترتفع الأخطاء؟
• زمن الانتظار — هل المسار حسّاس للتأخير مثل صفحة الدفع؟

تبدأ معظم الأنظمة الناضجة بنمط واحد ثم تضيف الطبقات مع نمو الحجم.

النمط 1: تجمع العمال البسيط

الأنسب للأحجام المتوسطة من 100 إلى 1000 حل في الساعة، بتوزيع المهام على خيوط محدودة داخل عملية واحدة.

┌──────────┐     ┌──────────────┐     ┌────────────┐
│  Scraper  │────▶│  Thread Pool │────▶│ CaptchaAI  │
│  Tasks    │     │  (5-20       │     │   API      │
│           │◀────│   workers)   │◀────│            │
└──────────┘     └──────────────┘     └────────────┘

متى تستخدمه

• عندما يتّسع الحمل داخل عملية واحدة ولا يتجاوز بضعة آلاف في الساعة.
• عندما تريد أبسط بنية بأقل كلفة صيانة.

التنفيذ

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import requests


class SimpleWorkerPool:
    def __init__(self, api_key, max_workers=10):
        self.api_key = api_key
        self.max_workers = max_workers
        self.base = "https://ocr.captchaai.com"

    def _solve_one(self, params):
        params["key"] = self.api_key
        params["json"] = 1

        resp = requests.post(f"{self.base}/in.php", data=params).json()
        if resp["status"] != 1:
            return {"error": resp["request"]}

        task_id = resp["request"]
        time.sleep(10)

        for _ in range(60):
            result = requests.get(
                f"{self.base}/res.php",
                params={"key": self.api_key, "action": "get", "id": task_id, "json": 1},
            ).json()

            if result["request"] == "CAPCHA_NOT_READY":
                time.sleep(5)
                continue
            if result["status"] == 1:
                return {"token": result["request"]}
            return {"error": result["request"]}

        return {"error": "timeout"}

    def solve_batch(self, tasks):
        """tasks: list of (identifier, params) tuples."""
        results = {}
        with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
            futures = {
                pool.submit(self._solve_one, params): ident
                for ident, params in tasks
            }
            for future in as_completed(futures):
                ident = futures[future]
                try:
                    results[ident] = future.result()
                except Exception as e:
                    results[ident] = {"error": str(e)}
        return results

النمط 2: خط الأنابيب القائم على قائمة الانتظار

الأنسب من 1000 إلى 10000 حل في الساعة مع التحكم في الضغط الخلفي. يفصل الإرسال عن الاستطلاع عبر قوائم انتظار مستقلة، فلا يختنق النظام عند تدفق الطلبات دفعة واحدة.

┌────────┐     ┌───────────┐     ┌──────────┐     ┌───────────┐     ┌────────┐
│Producer│────▶│ Submit    │────▶│ Pending  │────▶│  Poll     │────▶│Results │
│        │     │ Queue     │     │ Queue    │     │  Workers  │     │ Queue  │
└────────┘     └───────────┘     └──────────┘     └───────────┘     └────────┘

متى تستخدمه

• عند تجاوز الحمل قدرة عملية واحدة وحاجتك لفصل الإرسال عن الاستطلاع.
• عند الحاجة إلى ضغط خلفي يمنع إغراق واجهة API عند تدفق الطلبات دفعة واحدة.

التنفيذ

import queue
import threading
import time
import requests


class QueuePipeline:
    def __init__(self, api_key, submit_workers=5, poll_workers=10):
        self.api_key = api_key
        self.base = "https://ocr.captchaai.com"
        self.submit_queue = queue.Queue(maxsize=100)
        self.pending_queue = queue.Queue()
        self.results = {}
        self.results_lock = threading.Lock()
        self._running = False
        self.submit_workers = submit_workers
        self.poll_workers = poll_workers

    def start(self):
        self._running = True
        for _ in range(self.submit_workers):
            threading.Thread(target=self._submit_worker, daemon=True).start()
        for _ in range(self.poll_workers):
            threading.Thread(target=self._poll_worker, daemon=True).start()

    def stop(self):
        self._running = False

    def add(self, ident, params):
        self.submit_queue.put((ident, params))

    def get_result(self, ident, timeout=300):
        deadline = time.time() + timeout
        while time.time() < deadline:
            with self.results_lock:
                if ident in self.results:
                    return self.results.pop(ident)
            time.sleep(1)
        return {"error": "timeout"}

    def _submit_worker(self):
        while self._running:
            try:
                ident, params = self.submit_queue.get(timeout=1)
            except queue.Empty:
                continue
            params["key"] = self.api_key
            params["json"] = 1
            try:
                resp = requests.post(f"{self.base}/in.php", data=params).json()
                if resp["status"] == 1:
                    self.pending_queue.put((ident, resp["request"], time.time()))
                else:
                    with self.results_lock:
                        self.results[ident] = {"error": resp["request"]}
            except Exception as e:
                with self.results_lock:
                    self.results[ident] = {"error": str(e)}

    def _poll_worker(self):
        while self._running:
            try:
                ident, task_id, submitted_at = self.pending_queue.get(timeout=1)
            except queue.Empty:
                continue

            # Wait at least 10s from submission
            wait = 10 - (time.time() - submitted_at)
            if wait > 0:
                time.sleep(wait)

            try:
                resp = requests.get(
                    f"{self.base}/res.php",
                    params={"key": self.api_key, "action": "get", "id": task_id, "json": 1},
                ).json()

                if resp["request"] == "CAPCHA_NOT_READY":
                    self.pending_queue.put((ident, task_id, submitted_at))
                    time.sleep(3)
                elif resp["status"] == 1:
                    with self.results_lock:
                        self.results[ident] = {"token": resp["request"]}
                else:
                    with self.results_lock:
                        self.results[ident] = {"error": resp["request"]}
            except Exception:
                self.pending_queue.put((ident, task_id, submitted_at))
                time.sleep(5)

مثال على الاستخدام:

pipeline = QueuePipeline("YOUR_API_KEY")
pipeline.start()

# Add CAPTCHAs
pipeline.add("page_1", {"method": "turnstile", "sitekey": "KEY", "pageurl": "URL"})
pipeline.add("page_2", {"method": "userrecaptcha", "googlekey": "KEY", "pageurl": "URL"})

# Get results
result1 = pipeline.get_result("page_1")
result2 = pipeline.get_result("page_2")

pipeline.stop()

النمط 3: قاطع الدائرة لحماية النظام

يمنع حالات الفشل المتتالية عندما تتدهور واجهة API للحل، إذ يفتح الدائرة مؤقتًا بدل إغراق مزوّد متعثّر بمزيد من الطلبات.

متى تستخدمه

• عندما تكون واجهة API عرضة للتدهور المؤقت أو ارتفاع الأخطاء تحت الضغط.
• عندما تريد منع تراكم الطلبات على مزوّد متعثّر بدل إغراقه.

import time
import threading


class CircuitBreaker:
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing — reject requests
    HALF_OPEN = "half_open"  # Testing recovery

    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = self.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0
        self.lock = threading.Lock()

    def can_proceed(self):
        with self.lock:
            if self.state == self.CLOSED:
                return True
            if self.state == self.OPEN:
                if time.time() - self.last_failure_time > self.reset_timeout:
                    self.state = self.HALF_OPEN
                    return True
                return False
            # HALF_OPEN: allow one request
            return True

    def record_success(self):
        with self.lock:
            self.failure_count = 0
            self.state = self.CLOSED

    def record_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = self.OPEN


class ResilientSolver:
    def __init__(self, api_key):
        self.api_key = api_key
        self.breaker = CircuitBreaker(failure_threshold=5, reset_timeout=60)

    def solve(self, params):
        if not self.breaker.can_proceed():
            raise Exception("Circuit open — API degraded, try later")

        try:
            result = self._do_solve(params)
            self.breaker.record_success()
            return result
        except Exception as e:
            self.breaker.record_failure()
            raise

    def _do_solve(self, params):
        # Standard solve logic
        pass

النمط 4: الحل المسبق عبر مخزن مؤقت للرموز

احتفظ بمخزون جاهز من الرموز المحلولة مسبقًا لاستخدامها فورًا في المسارات الحساسة لزمن الانتظار مثل صفحات إتمام الشراء.

متى تستخدمه

• عندما يكون زمن الاستجابة حرجًا مثل صفحات إتمام الشراء وتسجيل الدخول.
• عندما يمكنك حلّ الرموز مسبقًا قبل لحظة الحاجة الفعلية إليها.

import queue
import threading
import time


class TokenBuffer:
    def __init__(self, solver, params, buffer_size=5, ttl_seconds=90):
        self.solver = solver
        self.params = params
        self.buffer = queue.Queue(maxsize=buffer_size)
        self.ttl = ttl_seconds
        self.buffer_size = buffer_size
        self._running = False

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

    def stop(self):
        self._running = False

    def get_token(self, timeout=30):
        """Get a pre-solved token. Returns None if buffer empty."""
        try:
            token, created_at = self.buffer.get(timeout=timeout)
            if time.time() - created_at > self.ttl:
                # Token expired, try next
                return self.get_token(timeout=timeout)
            return token
        except queue.Empty:
            return None

    def _fill_loop(self):
        while self._running:
            if self.buffer.qsize() < self.buffer_size:
                try:
                    token = self.solver.solve(self.params)
                    self.buffer.put((token, time.time()))
                except Exception:
                    time.sleep(5)
            else:
                time.sleep(2)

النمط 5: تجاوز الفشل بين عدة مزوّدين

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

متى تستخدمه

• عندما لا تحتمل الخدمة توقّفًا كاملًا عند تعطّل مزوّد واحد.
• عندما توازن بين عدة مزوّدين حسب الأولوية والتكلفة والسرعة.

class MultiProviderSolver:
    def __init__(self, providers):
        """providers: list of (name, solver_instance, priority) tuples."""
        self.providers = sorted(providers, key=lambda x: x[2])
        self.breakers = {name: CircuitBreaker() for name, _, _ in providers}

    def solve(self, params):
        errors = []
        for name, solver, _ in self.providers:
            if not self.breakers[name].can_proceed():
                continue
            try:
                result = solver.solve(params)
                self.breakers[name].record_success()
                return result
            except Exception as e:
                self.breakers[name].record_failure()
                errors.append(f"{name}: {e}")
        raise Exception(f"All providers failed: {'; '.join(errors)}")

دليل توسيع حل CAPTCHA على نطاق واسع

اختر بنيتك حسب شريحة الحجم لا حسب التفضيل: الجدول التالي يربط الطلبات في الساعة بالنمط المناسب وعدد العمال المقترح. ويحكم عدد الـ Threads في خطتك سقف التوازي الفعلي؛ فخطة ADVANCE بسعر 90 دولاراً تتيح 50 Thread، وخطة PREMIUM بسعر 170 دولاراً تتيح 100 Thread، مع حلول غير محدودة لكل Thread طوال الشهر.

خطوات الانتقال بين الشرائح

1. ابدأ بمكالمات مباشرة دون بنية خاصة ما دمت دون 100 طلب في الساعة.
2. انتقل إلى تجمع العمال ثم خط الأنابيب مع اقتراب حجمك من الآلاف.
3. أضف طوابير موزّعة عبر Redis أو RabbitMQ بعد تجاوز 10 آلاف طلب في الساعة.

مراقبة الأداء على نطاق واسع

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

import logging
from collections import defaultdict

logger = logging.getLogger("captcha_scale")


class ScaleMetrics:
    def __init__(self):
        self.counts = defaultdict(int)
        self.times = defaultdict(list)

    def record(self, captcha_type, success, elapsed):
        key = f"{captcha_type}_{'ok' if success else 'fail'}"
        self.counts[key] += 1
        self.times[captcha_type].append(elapsed)

    def report(self):
        for captcha_type in set(k.rsplit("_", 1)[0] for k in self.counts):
            ok = self.counts.get(f"{captcha_type}_ok", 0)
            fail = self.counts.get(f"{captcha_type}_fail", 0)
            total = ok + fail
            rate = (ok / total * 100) if total else 0
            times = self.times.get(captcha_type, [])
            avg_time = sum(times) / len(times) if times else 0
            logger.info(
                f"{captcha_type}: {total} solves, {rate:.1f}% success, {avg_time:.1f}s avg"
            )

مؤشرات يجب مراقبتها

• معدل الحل لكل نوع CAPTCHA لاكتشاف التدهور مبكرًا.
• وقت الاستجابة المتوسط ومقارنته بسقف SLA المتوقع.
• معدل ERROR_NO_SLOT_AVAILABLE كإشارة على بلوغ حد التوازي.

أخطاء شائعة عند التوسّع

أغلب مشكلات التوسّع سببها غياب الحدود والمراقبة لا الكود نفسه:

• تشغيل عدد Threads أكبر مما تتيحه خطتك، فتظهر أخطاء ERROR_NO_SLOT_AVAILABLE.
• إهمال الضغط الخلفي حتى تتراكم الطلبات وتستنزف الذاكرة.
• الاعتماد على مزوّد واحد دون مسار احتياطي عند التعطّل.

القاعدة العملية

أضف طبقة تعقيد واحدة فقط عند ظهور عنق الزجاجة الفعلي، لا قبل ذلك.

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

كيف أختار بين تجمع العمال وخط الأنابيب القائم على قائمة الانتظار؟

إذا اتسع الحمل داخل عملية واحدة ولم يتجاوز بضعة آلاف في الساعة، فتجمع العمال (النمط 1) أبسط وأسهل صيانة. وعند الحاجة إلى ضغط خلفي وفصل الإرسال عن الاستطلاع، فخط الأنابيب (النمط 2) هو الأمتن.

كم عدد الـ Threads الذي أحتاجه في خطة CaptchaAI للأحجام العالية؟

يحدد عدد الـ Threads سقف التوازي لا عدد الحلول. تبدأ الخطط من BASIC بسعر 15 دولاراً و5 Threads، وتصل إلى ENTERPRISE بسعر 300 دولار و200 Thread، مع حلول غير محدودة لكل Thread. قدّر ذروتك بالطلبات في الساعة واقسمها على متوسط زمن الحل لتعرف العدد اللازم تقريبًا.

كيف أوسّع الحل إلى ما بعد 10 آلاف طلب في الساعة؟

انتقل من قوائم الانتظار داخل العملية الواحدة إلى طوابير موزّعة عبر Redis أو RabbitMQ على عدة أجهزة. هذا يفصل الإنتاج عن المعالجة ويتيح إضافة عمال أفقيًا دون إعادة تصميم.

ماذا يحدث لبنيتي عند تدهور واجهة API وكيف يساعد قاطع الدائرة؟

عند ارتفاع الأخطاء يوقف قاطع الدائرة (النمط 3) الإرسال مؤقتًا بدل تكديس الطلبات على مزوّد متعثّر. اقرِنه بتجاوز الفشل متعدد المزوّدين (النمط 5) لتحويل الحمل إلى مزوّد احتياطي، وراقب معدل ERROR_NO_SLOT_AVAILABLE كمؤشر مبكر على بلوغ التوازي.

أدلة ذات صلة

• اختبار نقاط نهاية API لنماذج الويب عبر حل CAPTCHA
• معالجة 10 آلاف مهمة حل CAPTCHA في الساعة

ابدأ من النمط الأبسط الذي يغطّي حجمك الحالي، ثم رقِّ بنيتك تدريجيًا مع CaptchaAI كلما تحوّل نمو الحجم إلى سبب مباشر لمشكلات التشغيل.