تتبّع ميزانية الأخطاء لموثوقية حل CAPTCHA

ميزانية الأخطاء تحوّل سؤال «هل الموثوقية جيدة؟» إلى رقم واضح: كم حالة فشل يستطيع مسار حل CAPTCHA أن يتحمّلها قبل أن يهبط تحت هدف مستوى الخدمة (SLO). فبدل الاعتماد على انطباع بأنّ معدل النجاح «مقبول»، تمنحك الميزانية حدًّا كميًّا؛ تحته يمكنك التجربة والنشر بثقة، وفوقه يجب أن تتوقف عمليات النشر والتغييرات المحفوفة بالمخاطر. في هذا الدليل نبني متعقّبًا عمليًّا لميزانية الأخطاء بلغتَي Python وJavaScript، ونربطه مباشرةً بمسار الحل عبر CaptchaAI حتى يقرّر بنفسه متى يُبطئ أو يتوقف.

سنغطّي هنا ثلاثة محاور مترابطة:

• تعريف الـ SLO وحساب ميزانية الأخطاء وعتباتها الزمنية.
• بناء متعقّب عملي بلغتَي Python وJavaScript يربط النتيجة بالتنبيهات.
• ضبط تنبيهات معدل الحرق ومعالجة المشكلات الشائعة في الإنتاج.

أساسيات ميزانية الأخطاء

قبل كتابة أي سطر برمجي، اتّفق مع فريقك على أربعة مصطلحات تحكم القرار بأكمله:

المعادلة بسيطة: إذا كان الـ SLO عند 95% خلال نافذة 24 ساعة تُنفَّذ فيها 10000 عملية حل، فإن ميزانية الأخطاء هي 500 حالة فشل. طالما بقيت تحت هذا الرقم فأنت داخل الهدف، وبمجرد بلوغ الـ 500 يجب إيقاف عمليات النشر الجديدة وأي تغيير قد يزيد نسبة الفشل. الفكرة الجوهرية أنّ الفشل ليس عدوًّا يجب القضاء عليه بالكامل، بل مورد محدود تُنفقه بحكمة على التجارب والتحديثات.

لحساب ميزانيتك بنفسك، اتبع ثلاث خطوات:

1. قِس معدل نجاحك الحالي على نافذة زمنية تمثيلية لا على يوم استثنائي.
2. اطرح 2–3% منه لتحديد الـ SLO وترك هامش أمان معقول.
3. اضرب الحجم المتوقع من الطلبات في نسبة الفشل المسموح بها لتحصل على عدد حالات الفشل ضمن الميزانية.

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

متى تحتاج إلى ميزانية أخطاء؟ سيناريو من السوق

تخيّل فريق أتمتة يدير موقع مقارنة أسعار في منطقة الخليج، يعتمد على حل Cloudflare Turnstile لجمع بيانات المتاجر. خلال موسم «الجمعة البيضاء» يقفز حجم الطلبات إلى أضعافه، وترتفع معه حالات الفشل بسبب ضغط الشبكة وتغيّر صفحات المتاجر. بدون ميزانية أخطاء، سيكتشف الفريق المشكلة متأخرًا بعد أن تتراكم البيانات الناقصة. أمّا مع متعقّب ميزانية يعمل على نافذة ساعة واحدة، فيصل تنبيه فور بلوغ الاستهلاك 50% من الميزانية، ويتوقف الحل تلقائيًّا قبل أن تُهدر أرصدة الحساب على طلبات فاشلة. هنا يظهر أثر النموذج القائم على الـ threads في CaptchaAI: خطة مثل ADVANCE بسعر 90$ شهريًّا تمنح 50 thread متزامنًا مع حلول غير محدودة لكل thread، فيصبح القرار متعلقًا بموثوقية المسار لا بعدد العمليات المتبقية، وميزانية الأخطاء هي الأداة التي تقيس تلك الموثوقية لحظةً بلحظة.

Python: بناء متعقّب ميزانية الأخطاء

المتعقّب التالي يحتفظ بنافذة زمنية متجدّدة من الأحداث، يحسب الميزانية المتبقية ومعدل الحرق، ويُطلق ردود نداء (callbacks) عند الانتقال بين الحالات الأربع: سليمة، تحذير، حرجة، ومستنفدة. لاحظ أنّه آمن للاستخدام مع تعدّد الخيوط عبر threading.Lock، وأنّ دالة solve_with_budget تسجّل كل محاولة نجاحًا أو فشلًا ثم تتوقف تلقائيًّا حين تنفد الميزانية:

import time
import threading
from dataclasses import dataclass, field
from collections import deque
from enum import Enum

API_KEY = "YOUR_API_KEY"


class BudgetStatus(Enum):
    HEALTHY = "healthy"          # Budget > 50% remaining
    WARNING = "warning"          # Budget 10-50% remaining
    CRITICAL = "critical"        # Budget < 10% remaining
    EXHAUSTED = "exhausted"      # Budget depleted


@dataclass
class SLOConfig:
    """Service Level Objective configuration."""
    target_success_rate: float = 0.95  # 95%
    window_seconds: int = 86400        # 24 hours
    warning_threshold: float = 0.50    # Alert at 50% budget
    critical_threshold: float = 0.10   # Alert at 10% budget


@dataclass
class ErrorBudgetEvent:
    timestamp: float
    success: bool


class ErrorBudgetTracker:
    """Tracks error budget consumption for CAPTCHA solving."""

    def __init__(self, config: SLOConfig = SLOConfig()):
        self.config = config
        self._events: deque[ErrorBudgetEvent] = deque()
        self._lock = threading.Lock()
        self._callbacks: dict[BudgetStatus, list[callable]] = {
            status: [] for status in BudgetStatus
        }
        self._last_status = BudgetStatus.HEALTHY

    def on_status_change(self, status: BudgetStatus, callback: callable):
        """Register a callback for status transitions."""
        self._callbacks[status].append(callback)

    def record(self, success: bool):
        """Record a solve attempt."""
        now = time.monotonic()
        event = ErrorBudgetEvent(timestamp=now, success=success)

        with self._lock:
            self._events.append(event)
            self._prune(now)
            new_status = self._compute_status()

            if new_status != self._last_status:
                self._last_status = new_status
                for cb in self._callbacks.get(new_status, []):
                    try:
                        cb(self.get_report())
                    except Exception as e:
                        print(f"[BUDGET] Callback error: {e}")

    def _prune(self, now: float):
        """Remove events outside the window."""
        cutoff = now - self.config.window_seconds
        while self._events and self._events[0].timestamp < cutoff:
            self._events.popleft()

    def _compute_status(self) -> BudgetStatus:
        remaining = self.remaining_fraction
        if remaining <= 0:
            return BudgetStatus.EXHAUSTED
        if remaining < self.config.critical_threshold:
            return BudgetStatus.CRITICAL
        if remaining < self.config.warning_threshold:
            return BudgetStatus.WARNING
        return BudgetStatus.HEALTHY

    @property
    def total_events(self) -> int:
        with self._lock:
            return len(self._events)

    @property
    def success_count(self) -> int:
        with self._lock:
            return sum(1 for e in self._events if e.success)

    @property
    def failure_count(self) -> int:
        with self._lock:
            return sum(1 for e in self._events if not e.success)

    @property
    def current_success_rate(self) -> float:
        total = self.total_events
        return self.success_count / total if total > 0 else 1.0

    @property
    def error_budget_total(self) -> float:
        """Total allowed failures in the window."""
        total = self.total_events
        if total == 0:
            return 0
        return total * (1 - self.config.target_success_rate)

    @property
    def error_budget_remaining(self) -> float:
        """Remaining failure allowance."""
        return max(0, self.error_budget_total - self.failure_count)

    @property
    def remaining_fraction(self) -> float:
        """Fraction of error budget remaining (0.0 to 1.0)."""
        budget = self.error_budget_total
        if budget <= 0:
            return 1.0 if self.failure_count == 0 else 0.0
        return max(0, self.error_budget_remaining / budget)

    @property
    def burn_rate(self) -> float:
        """How fast the budget is being consumed (1.0 = normal, 2.0 = 2× faster)."""
        total = self.total_events
        if total == 0:
            return 0.0
        expected_failures = total * (1 - self.config.target_success_rate)
        if expected_failures == 0:
            return 0.0
        return self.failure_count / expected_failures

    def get_report(self) -> dict:
        return {
            "status": self._last_status.value,
            "slo_target": self.config.target_success_rate,
            "current_rate": round(self.current_success_rate, 4),
            "total_events": self.total_events,
            "successes": self.success_count,
            "failures": self.failure_count,
            "budget_total": round(self.error_budget_total, 1),
            "budget_remaining": round(self.error_budget_remaining, 1),
            "budget_remaining_pct": round(self.remaining_fraction * 100, 1),
            "burn_rate": round(self.burn_rate, 2),
        }


# --- Integration with solver ---

budget = ErrorBudgetTracker(SLOConfig(
    target_success_rate=0.95,
    window_seconds=3600,  # 1-hour window for demo
))

# Register alerts
budget.on_status_change(BudgetStatus.WARNING, lambda r:
    print(f"[ALERT] Budget warning: {r['budget_remaining_pct']}% remaining"))

budget.on_status_change(BudgetStatus.CRITICAL, lambda r:
    print(f"[ALERT] Budget critical: {r['budget_remaining_pct']}% remaining"))

budget.on_status_change(BudgetStatus.EXHAUSTED, lambda r:
    print(f"[ALERT] Budget EXHAUSTED — throttle new requests"))


def solve_with_budget(params: dict) -> str:
    """Solve CAPTCHA while tracking error budget."""
    import requests

    if budget._last_status == BudgetStatus.EXHAUSTED:
        raise RuntimeError("Error budget exhausted — solving paused")

    try:
        submit_params = {**params, "key": API_KEY, "json": 1}
        resp = requests.post(
            "https://ocr.captchaai.com/in.php", data=submit_params, timeout=30
        ).json()
        if resp.get("status") != 1:
            budget.record(False)
            raise RuntimeError(f"Submit: {resp.get('request')}")

        task_id = resp["request"]
        start = time.monotonic()
        while time.monotonic() - start < 180:
            time.sleep(5)
            poll = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id, "json": 1,
            }, timeout=15).json()

            if poll.get("request") == "CAPCHA_NOT_READY":
                continue
            if poll.get("status") == 1:
                budget.record(True)
                return poll["request"]

            budget.record(False)
            raise RuntimeError(f"Solve: {poll.get('request')}")

        budget.record(False)
        raise RuntimeError("Timeout")

    except Exception:
        budget.record(False)
        raise


# Usage
for i in range(100):
    try:
        token = solve_with_budget({
            "method": "turnstile",
            "sitekey": "0x4XXXXXXXXXXXXXXXXX",
            "pageurl": "https://example.com",
        })
    except RuntimeError as e:
        if "exhausted" in str(e):
            print(f"Stopped at iteration {i}")
            break

print(budget.get_report())

JavaScript: متعقّب ميزانية الأخطاء

النسخة نفسها لبيئة Node.js أو المتصفح. المنطق متطابق: نافذة زمنية متجدّدة، حساب للميزانية المتبقية، ومعدل حرق يُقارَن بالحدّين الحرج والتحذيري. استدعِ record(true) عند نجاح الحل وrecord(false) عند فشله، واربط ردّ نداء بحالة exhausted لإيقاف إرسال الطلبات الجديدة:

class ErrorBudgetTracker {
  #events = [];
  #config;
  #callbacks = {};

  constructor(config = {}) {
    this.#config = {
      targetRate: config.targetRate || 0.95,
      windowMs: config.windowMs || 3600_000,
      warningThreshold: config.warningThreshold || 0.5,
      criticalThreshold: config.criticalThreshold || 0.1,
    };
    this.lastStatus = "healthy";
  }

  on(status, callback) {
    this.#callbacks[status] = this.#callbacks[status] || [];
    this.#callbacks[status].push(callback);
  }

  record(success) {
    const now = Date.now();
    this.#events.push({ time: now, success });
    this.#prune(now);

    const newStatus = this.#computeStatus();
    if (newStatus !== this.lastStatus) {
      this.lastStatus = newStatus;
      for (const cb of this.#callbacks[newStatus] || []) {
        cb(this.report());
      }
    }
  }

  #prune(now) {
    const cutoff = now - this.#config.windowMs;
    while (this.#events.length && this.#events[0].time < cutoff) {
      this.#events.shift();
    }
  }

  #computeStatus() {
    const frac = this.remainingFraction;
    if (frac <= 0) return "exhausted";
    if (frac < this.#config.criticalThreshold) return "critical";
    if (frac < this.#config.warningThreshold) return "warning";
    return "healthy";
  }

  get total() { return this.#events.length; }
  get successes() { return this.#events.filter((e) => e.success).length; }
  get failures() { return this.#events.filter((e) => !e.success).length; }
  get currentRate() { return this.total ? this.successes / this.total : 1; }

  get budgetTotal() {
    return this.total * (1 - this.#config.targetRate);
  }

  get budgetRemaining() {
    return Math.max(0, this.budgetTotal - this.failures);
  }

  get remainingFraction() {
    const bt = this.budgetTotal;
    if (bt <= 0) return this.failures === 0 ? 1 : 0;
    return Math.max(0, this.budgetRemaining / bt);
  }

  get burnRate() {
    const expected = this.total * (1 - this.#config.targetRate);
    return expected > 0 ? this.failures / expected : 0;
  }

  report() {
    return {
      status: this.lastStatus,
      currentRate: Math.round(this.currentRate * 10000) / 10000,
      total: this.total,
      failures: this.failures,
      budgetRemainingPct: Math.round(this.remainingFraction * 1000) / 10,
      burnRate: Math.round(this.burnRate * 100) / 100,
    };
  }
}

// Usage
const budget = new ErrorBudgetTracker({ targetRate: 0.95, windowMs: 3600_000 });

budget.on("warning", (r) => console.log(`[WARN] ${r.budgetRemainingPct}% budget left`));
budget.on("exhausted", (r) => console.log("[ALERT] Budget exhausted!"));

// Record results from your solver
budget.record(true);   // success
budget.record(false);  // failure
console.log(budget.report());

تنبيهات معدل الحرق

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

القاعدة العملية أن تجمع بين إنذارين متكاملين:

• إنذار سريع مرتبط بمعدل حرق مرتفع على مدى قصير، يلتقط الاستنزاف الحادّ والمفاجئ.
• إنذار بطيء مرتبط بمعدل حرق معتدل على مدى أطول، يلتقط التسرّب المستمر الذي يمرّ دون أن يُلاحَظ.

استكشاف الأخطاء ومعالجتها

معظم مشكلات تتبّع ميزانية الأخطاء تعود إلى ضبط الـ SLO أو حجم النافذة، لا إلى الكود نفسه:

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

ما الفرق بين ميزانية الأخطاء وبند SLA في العقد؟

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

كيف أختار طول النافذة الزمنية المناسبة؟

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

كيف تؤثّر خطة CaptchaAI وعدد الـ threads في ميزانية الأخطاء؟

لأن CaptchaAI يفوتر حسب الـ threads المتزامنة مع حلول غير محدودة لكل thread، فإن ميزانية الأخطاء لا تتعلق بعدد العمليات المتبقية بل بمعدل نجاحها. زيادة عدد الـ threads ترفع الإنتاجية لكنها قد تكشف أعطالًا كامنة أسرع، لذا راقب معدل الحرق بعد أي ترقية للخطة أو زيادة في التوازي.

لماذا يصبح معدل الحرق غير موثوق عند انخفاض عدد الطلبات؟

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

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

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

أدلة ذات صلة

• دليل البدء السريع مع CaptchaAI
• تحليل سجلّات حل CAPTCHA باستخدام ELK Stack