كيف تحافظ على استمرار الأتمتة عند فشل حل اختبار CAPTCHA

الحل المتين لا يعني ألا يفشل حل CAPTCHA أبداً، بل أن يستمر خط الأتمتة في العمل عندما يفشل. اختصر كل سيناريو إلى واحد من أربعة قرارات: تخطَّ العنصر وتابع، أعِد المحاولة، أرسل المهمة إلى قائمة انتظار للمعالجة لاحقاً، أو انتقل إلى مسار بديل. هذا هو جوهر التدهور السلس: تحويل الفشل من انهيار مفاجئ إلى سلوك محسوب مسبقاً.

الخطأ الأكثر شيوعاً أن يتعامل السكربت مع أول استثناء بالتوقف الكامل، فتضيع نتائج ساعات من العمل بسبب مهلة عابرة أو رصيد نفد لدقائق. القاعدة العملية هي اختيار السلوك الأقل ضرراً لكل نوع مسار، لا إخفاء الفشل. مهمة معالجة دُفعية (batch) لآلاف الصفحات تتحمّل تخطي بعض العناصر مؤقتاً، بينما عملية إتمام شراء أو اختبار end-to-end حسّاس يحتاج قراراً فورياً وواضحاً بدل التخطي الصامت.

لتوضيح الفكرة عملياً: تخيّل فريقاً في القاهرة يشغّل مهمة مراقبة أسعار ليلية تمرّ على عشرات آلاف صفحات المنتجات لدى منافسين، وكل صفحة محمية بـ reCAPTCHA. لو توقّف السكربت عند أول ERROR_ZERO_BALANCE الساعة الثالثة فجراً، تضيع الدفعة كاملة ولا أحد ينتبه حتى الصباح. أما مع تدهور سلس مُصمَّم جيداً، فيدخل النظام وضعاً محدوداً، يؤجّل الصفحات المتعذّرة إلى قائمة انتظار، ويستأنفها تلقائياً فور شحن الرصيد، من دون فقدان تقدّم يُذكر.

اختر نمط التدهور بحسب طبيعة المسار

لا يوجد نمط واحد يناسب كل الحالات. الجدول التالي يربط نوع المسار بالسلوك الأنسب له:

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

أوضاع الفشل الشائعة ورموز الأخطاء

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

القاعدة البسيطة: أعِد المحاولة للمهلات وحدود المعدل، وتخطَّ أو أصلِح المصدر عند أخطاء المعلمات والمفتاح.

النمط 1: التخطي والمتابعة

للعمليات المجمّعة التي يكون فيها فقدان عنصر مفرد مقبولاً، اجعل دالة الحل تُرجع None عند الفشل بدل أن ترفع استثناءً يُوقف الحلقة. بهذا تستمر المعالجة وتحتفظ بسجل واضح للعناصر المتخطاة لمراجعتها لاحقاً:

import requests
import time

API_KEY = "YOUR_API_KEY"


def solve_or_skip(captcha_type, sitekey, page_url, max_retries=2):
    """Try to solve; return None on failure instead of crashing."""
    for attempt in range(max_retries):
        try:
            token = solve_captcha(captcha_type, sitekey, page_url)
            if token:
                return token
        except Exception as e:
            print(f"Attempt {attempt + 1} failed: {e}")

    return None  # Skip this item


def process_urls(urls):
    results = []
    skipped = []

    for url in urls:
        sitekey = extract_sitekey(url)
        if not sitekey:
            skipped.append({"url": url, "reason": "no_sitekey"})
            continue

        token = solve_or_skip("recaptcha_v2", sitekey, url)
        if token:
            data = submit_form(url, token)
            results.append({"url": url, "data": data})
        else:
            skipped.append({"url": url, "reason": "solve_failed"})

    print(f"Processed: {len(results)}, Skipped: {len(skipped)}")
    return results, skipped

احتفظ دائماً بقائمة skipped مع سبب واضح لكل عنصر؛ فهي التي تكشف لاحقاً ما إذا كان الفشل عشوائياً أم ناتجاً عن خلل منهجي في الاستخراج.

النمط 2: قائمة انتظار لإعادة المحاولة

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

from collections import deque
import json

class RetryQueue:
    def __init__(self, max_retries=3, backoff_base=60):
        self.queue = deque()
        self.max_retries = max_retries
        self.backoff_base = backoff_base

    def add(self, task):
        task["retry_count"] = task.get("retry_count", 0) + 1
        if task["retry_count"] <= self.max_retries:
            task["retry_after"] = time.time() + (
                self.backoff_base * task["retry_count"]
            )
            self.queue.append(task)
            return True
        return False  # Exceeded max retries

    def get_ready(self):
        """Get tasks ready for retry."""
        ready = []
        remaining = deque()
        now = time.time()

        while self.queue:
            task = self.queue.popleft()
            if task["retry_after"] <= now:
                ready.append(task)
            else:
                remaining.append(task)

        self.queue = remaining
        return ready

    def save(self, filepath="retry_queue.json"):
        with open(filepath, "w") as f:
            json.dump(list(self.queue), f)

    def load(self, filepath="retry_queue.json"):
        try:
            with open(filepath) as f:
                self.queue = deque(json.load(f))
        except FileNotFoundError:
            pass


# Usage
retry_q = RetryQueue()

def process_with_retry(task):
    try:
        token = solve_captcha(task["type"], task["sitekey"], task["url"])
        if token:
            return submit_form(task["url"], token)
        else:
            retry_q.add(task)
    except Exception:
        retry_q.add(task)

# Process retry queue periodically
def drain_retry_queue():
    ready = retry_q.get_ready()
    for task in ready:
        process_with_retry(task)

لاحظ أن max_retries وbackoff_base معاً يمنعان القائمة من التضخّم بلا نهاية. المهمة التي تتجاوز الحد الأقصى ينبغي أن تنتقل إلى مسار منفصل، وهنا يأتي دور قائمة الرسائل الميتة (Dead Letter Queue) لعزل الحالات التي تحتاج تدخلاً بشرياً.

النمط 3: الوضع المتدهور

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

class CaptchaSolver:
    def __init__(self, api_key):
        self.api_key = api_key
        self.degraded = False
        self.failure_count = 0
        self.failure_threshold = 5
        self.recovery_time = None

    def solve(self, captcha_type, sitekey, page_url):
        if self.degraded:
            if time.time() < self.recovery_time:
                return self._degraded_action(page_url)
            else:
                self.degraded = False
                self.failure_count = 0

        try:
            token = self._solve_api(captcha_type, sitekey, page_url)
            self.failure_count = 0
            return token
        except Exception as e:
            self.failure_count += 1
            if self.failure_count >= self.failure_threshold:
                self._enter_degraded_mode()
            raise

    def _enter_degraded_mode(self):
        self.degraded = True
        self.recovery_time = time.time() + 300  # 5 min
        print("Entering degraded mode for 5 minutes")
        # Send alert

    def _degraded_action(self, url):
        """What to do when solving is unavailable."""
        # Option A: Skip CAPTCHA pages entirely
        return None

        # Option B: Queue for later
        # retry_queue.add({"url": url, ...})
        # return None

        # Option C: Try alternative solver
        # return self._solve_with_backup_api(...)

    def _solve_api(self, captcha_type, sitekey, page_url):
        # Normal CaptchaAI API call
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
            "json": "1",
        }).json()

        if resp["status"] != 1:
            raise Exception(resp["request"])

        task_id = resp["request"]
        for _ in range(24):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": "1"
            }).json()
            if result["status"] == 1:
                return result["request"]
            if result["request"] != "CAPCHA_NOT_READY":
                raise Exception(result["request"])

        raise Exception("TIMEOUT")

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

نمط مدمج بـ Node.js

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

class ResilientSolver {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.retryQueue = [];
    this.failureCount = 0;
    this.degraded = false;
  }

  async solve(type, sitekey, pageUrl) {
    if (this.degraded) {
      this.retryQueue.push({ type, sitekey, pageUrl, addedAt: Date.now() });
      return null;
    }

    try {
      const token = await this._callApi(type, sitekey, pageUrl);
      this.failureCount = 0;
      return token;
    } catch (err) {
      this.failureCount++;

      if (err.message === 'ERROR_ZERO_BALANCE') {
        this._enterDegraded(600000); // 10 min
        return null;
      }

      if (this.failureCount >= 5) {
        this._enterDegraded(300000); // 5 min
      }

      this.retryQueue.push({ type, sitekey, pageUrl, addedAt: Date.now() });
      return null;
    }
  }

  _enterDegraded(durationMs) {
    this.degraded = true;
    console.warn(`Degraded mode for ${durationMs / 1000}s`);
    setTimeout(() => {
      this.degraded = false;
      this.failureCount = 0;
      this.drainRetryQueue();
    }, durationMs);
  }

  async drainRetryQueue() {
    const tasks = this.retryQueue.splice(0);
    for (const task of tasks) {
      await this.solve(task.type, task.sitekey, task.pageUrl);
    }
  }

  async _callApi(type, sitekey, pageUrl) {
    // Standard submit + poll
    const axios = require('axios');
    const submit = await axios.post('https://ocr.captchaai.com/in.php', null, {
      params: { key: this.apiKey, method: 'userrecaptcha', googlekey: sitekey, pageurl: pageUrl, json: 1 },
    });
    if (submit.data.status !== 1) throw new Error(submit.data.request);

    const taskId = submit.data.request;
    for (let i = 0; i < 24; i++) {
      await new Promise(r => setTimeout(r, 5000));
      const poll = await axios.get('https://ocr.captchaai.com/res.php', {
        params: { key: this.apiKey, action: 'get', id: taskId, json: 1 },
      });
      if (poll.data.status === 1) return poll.data.request;
      if (poll.data.request !== 'CAPCHA_NOT_READY') throw new Error(poll.data.request);
    }
    throw new Error('TIMEOUT');
  }
}

نصائح لتشغيل موثوق في الإنتاج

بعد تطبيق الأنماط، تفصلك بضع خطوات عن نظام يصمد فعلاً تحت الضغط:

• ثبّت قائمة الانتظار على قرص أو قاعدة بيانات، لا في الذاكرة فقط، حتى لا تضيع المهام المؤجّلة عند إعادة تشغيل الخدمة أو نشر إصدار جديد.
• راقب معدل الفشل كمقياس مستقل، وأطلق تنبيهاً عند دخول الوضع المتدهور بدل انتظار شكوى مستخدم. رقم بسيط مثل نسبة الأخطاء في آخر دقيقة يكفي للإنذار المبكر.
• افصل الأخطاء العابرة عن الدائمة في السجلّات، فتراكم ERROR_WRONG_GOOGLEKEY يشير إلى خلل في الاستخراج لا إلى تقلّب الشبكة.
• اضبط عتبة الأعطال ومهلة التعافي حسب حجمك؛ العتبة المنخفضة جداً توقف كل شيء عند أول تذبذب، والمهلة الطويلة تُبقيك متدهوراً بعد عودة الخدمة.

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

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

أين أخزّن قائمة إعادة المحاولة حتى لا تُفقد المهام عند إعادة تشغيل الخدمة؟

القائمة في الذاكرة تختفي مع كل إعادة تشغيل أو نشر. استخدم ملف JSON للمشاريع الصغيرة، أو Redis أو قاعدة بيانات للأحمال الأكبر. المثال أعلاه يوفّر save وload لتثبيت القائمة على ملف كنقطة بداية.

ما المدة المناسبة للبقاء في الوضع المتدهور قبل إعادة المحاولة؟

ابدأ بخمس دقائق للأعطال العامة، وامنح خطأ الرصيد الصفري مهلة أطول (عشر دقائق مثلاً) لأنه لا يزول إلا بعد شحن الرصيد يدوياً. اضبط الأرقام بناءً على مدة الانقطاعات الفعلية في بيئتك، وأضف فحص صحة يعيدك مبكراً عند تعافي الخدمة.

كيف أراقب معدل الفشل لأعرف أن النظام تدهور فعلاً؟

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

هل يستحق إعداد مزوّد حل احتياطي العناء في مشروع صغير؟

غالباً لا. للمشاريع الصغيرة، قائمة إعادة محاولة موثوقة مع تنبيه عند فشل CaptchaAI تكفي وأبسط في الصيانة. المزوّد الاحتياطي يصبح مبرَّراً حين يكون توقّف المسار مكلفاً بحيث لا يُقبل انتظار دقائق التعافي.

كيف أتعامل مع خطأ ERROR_ZERO_BALANCE دون إيقاف كل المهام؟

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

مع CaptchaAI داخل مسار واضح لإعادة المحاولة والتخطي والتنبيه، لا يتحوّل فشل CAPTCHA واحد إلى تعطّل كامل للنظام. طبّق الأنماط أعلاه لبناء أتمتة تتعافى بنفسها.

أدلة ذات صلة

• نمط قاطع الدائرة لحماية استدعاءات CAPTCHA API
• قائمة الرسائل الميتة (Dead Letter Queue) لمهام CAPTCHA الفاشلة