Hướng Dẫn Thực Hành

Theo dõi ngân sách lỗi cho CAPTCHA Giải quyết độ tin cậy

Quy trình giải CAPTCHA của bạn hướng tới tỷ lệ thành công 95%. Tuần trước là 94,2%. Đó có phải là một vấn đề? Nếu không có ngân sách sai sót, bạn không thể trả lời câu hỏi đó một cách định lượng. Ngân sách lỗi cho bạn biết chính xác số lượng lỗi bạn có thể chịu đựng trước khi độ tin cậy giảm xuống dưới mức SLO của bạn - và phải làm gì khi hết ngân sách.

Lỗi cơ bản về ngân sách

Khái niệm Định nghĩa Ví dụ
CHƯA Tỷ lệ thành công mục tiêu Giải quyết thành công 95%
Lỗi ngân sách Tỷ lệ thất bại cho phép 5% tổng số lần giải quyết có thể thất bại
Tốc độ ghi Ngân sách được tiêu thụ nhanh như thế nào 2× có nghĩa là ngân sách đã cạn trong một nửa thời lượng
Cửa sổ Thời gian đo Lăn 24 giờ hoặc 7 ngày

Nếu SLO của bạn là 95% trong khoảng thời gian 24 giờ với 10.000 lần giải quyết, thì ngân sách lỗi của bạn là 500 lần thất bại. Khi bạn gặp phải 500 lần thất bại, việc triển khai mới hoặc các thay đổi rủi ro sẽ dừng lại.

Python: Trình theo dõi ngân sách lỗi

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: Trình theo dõi ngân sách lỗi

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());

Cảnh báo tốc độ ghi

Tốc độ ghi Ý nghĩa hành động
< 1,0 Tiêu thụ chậm hơn dự kiến Không cần hành động
1.0 Đang trên đà xả hơi ở cuối cửa sổ Giám sát chặt chẽ
2.0 Ngân sách đã cạn kiệt trong một nửa thời hạn Hãy điều tra và làm chậm lại
5.0+ Tiêu thụ ngân sách nhanh chóng Tạm dừng các giải pháp không quan trọng

Khắc phục sự cố

Vấn đề Nguyên nhân Cách xử lý
Ngân sách cạn kiệt quá nhanh SLO quá chặt so với điều kiện thực tế Đặt SLO thực tế dựa trên dữ liệu lịch sử
Ngân sách không bao giờ tiêu dùng SLO quá hào phóng Thắt chặt SLO để thúc đẩy cải thiện độ tin cậy
Chuyển đổi trạng thái giữa các trạng thái Cửa sổ quá ngắn Sử dụng khoảng thời gian đo dài hơn (24h so với 1h)
Tốc độ ghi sai lệch ở âm lượng thấp Một vài sự kiện làm lệch phép tính Yêu cầu số lượng sự kiện tối thiểu trước khi tính tỷ lệ đốt cháy
Bộ nhớ theo dõi ngân sách tăng lên Sự kiện không được cắt tỉa Xác minh _prune chạy trên mọi lệnh gọi record()

Câu hỏi thường gặp

SLO thực tế để giải CAPTCHA là gì?

Phụ thuộc vào loại CAPTCHA. reCAPTCHA v2 thường đạt tỷ lệ giải quyết 90–95%. Cloudflare Turnstile có thể cao hơn. CAPTCHA hình ảnh khác nhau. Bắt đầu bằng cách đo lường tỷ lệ thành công hiện tại của bạn, sau đó đặt SLO 2–3% dưới mức cơ sở đó để tạo ra mức sai sót có ý nghĩa.

Điều gì sẽ xảy ra khi hết số lượng lỗi?

Các tùy chọn từ ít nhất đến tích cực nhất: cảnh báo nhóm, điều tiết các yêu cầu mới, tạm dừng các giải pháp không cần thiết, chuyển sang xử lý CAPTCHA thủ công. Đừng bao giờ âm thầm bỏ qua một khoản ngân sách đã cạn kiệt.

Làm cách nào để xử lý số lượng lỗi trên nhiều loại CAPTCHA?

Theo dõi ngân sách riêng cho từng loại. reCAPTCHA có thể có SLO 93% trong khi Turnstile có 97%. Việc tổng hợp chúng vào một ngân sách sẽ che giấu các vấn đề cụ thể về loại hình.

bài viết liên quan

Các bước tiếp theo

Theo dõi độ tin cậy giải CAPTCHA của bạn một cách định lượng —lấy khóa API CaptchaAI của bạnvà thực hiện theo dõi ngân sách lỗi.

Hướng dẫn liên quan:

Os comentários estão desativados para este artigo.