Degradação graciosa quando a resolução de CAPTCHA falha

A resolução de CAPTCHA pode falhar: tempos limite, parâmetros incorretos, saldo zero ou limites de taxa. Se sua automação falhar na primeira falha, você perderá todo o progresso. A degradação graciosa mantém o pipeline em execução: pule, tente novamente, enfileire-se ou retorne a alternativas.

Modos de falha

Padrão 1: pular e continuar

Para operações em lote onde falhas individuais são aceitáveis:

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

Padrão 2: fila de novas tentativas

As tarefas com falha vão para uma fila de novas tentativas para processamento posterior:

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)

Padrão 3: modo degradado

Quando o serviço de resolução não estiver disponível, mude para um modo limitado:

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

Node.js: padrão combinado

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');
  }
}

Tentar novamente a matriz de decisão

• Tente novamente somente quando a falha parecer transitória e o estado da sessão circundante ainda for válido.
• Repita a etapa quando a solicitação puder ser reconstruída de forma determinística sem duplicar as ações do usuário.
• Ignore ou pause o fluxo de trabalho quando novas tentativas apenas desperdiçarem o equilíbrio ou aumentarem o risco de detecção.

Solução de problemas

Perguntas frequentes

Qual é a diferença entre degradação normal e um disjuntor?

Um disjuntor impede completamente as chamadas quando são detectadas falhas. A degradação graciosa é mais ampla: inclui comportamentos alternativos, lógica de salto e fluxos de trabalho alternativos. Eles funcionam bem juntos.

Devo sempre tentar novamente as tarefas que falharam?

Não para ERROR_BAD_PARAMETERS ou ERROR_WRONG_GOOGLEKEY - eles não terão sucesso na nova tentativa. Tente novamente apenas erros transitórios, como tempos limite e limites de taxa.

Crie automação CAPTCHA resiliente com CaptchaAI

Obtenha sua chave API emcaptchaai.com.

Guias relacionados

• Padrão de disjuntor para chamadas de API CAPTCHA
• Fila de mensagens mortas para tarefas com falha
• Implementando Lógica de Nova Tentativa