Dégradation gracieuse lorsque la résolution de CAPTCHA échoue

Un échec de résolution CAPTCHA n'appelle jamais une seule réponse : selon la cause, la bonne réaction est de réessayer, de s'arrêter pour corriger, de mettre en pause ou d'ignorer l'item. La dégradation gracieuse consiste à coder ces réactions de repli pour qu'un traitement par lots tienne debout quand une fraction des tâches échoue. Prenez un pipeline de scraping qui parcourt 10 000 fiches derrière un reCAPTCHA v2 : si le worker s'interrompt à la première ERROR_ZERO_BALANCE, vous perdez la progression des milliers de pages déjà traitées.

Le principe tient en une phrase : ne traitez pas tous les échecs de la même manière. Les quatre modèles ci-dessous couvrent ces cas.

Identifier le type d'échec avant de réagir

Avant de choisir une stratégie, distinguez les échecs transitoires (récupérables en réessayant) des échecs permanents (qui persisteront tant que le code ou le compte n'est pas corrigé). Rejouer un échec permanent ne fait qu'accumuler du bruit et de la dépense.

Ignorer, réessayer ou mettre en file d'attente : la bonne réponse par cas

Une fois le type d'échec identifié, la décision devient mécanique : un échec transitoire se rejoue, un échec permanent s'arrête et se corrige, un blocage de compte se met en pause en alertant une personne.

Modèle 1 : ignorer et continuer

C'est le repli le plus simple, adapté aux traitements par lots où quelques échecs isolés sont tolérables : une page manquée sur dix mille ne justifie pas d'arrêter la chaîne. La fonction renvoie None au lieu de lever une exception, et l'appelant range l'item dans une liste skipped.

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

Conservez la liste des items ignorés avec leur motif : un pic soudain de solve_failed signale souvent un changement côté site cible ou un problème de solde.

Modèle 2 : file d'attente de nouvelles tentatives avec backoff

Quand une tâche a de la valeur mais n'est pas urgente, ne l'abandonnez pas : placez-la dans une file d'attente et rejouez-la plus tard, avec un délai croissant à chaque tentative (backoff). Le compteur retry_count plafonne le nombre d'essais ; au-delà, la tâche sort de la boucle et part idéalement vers une file de lettres mortes pour inspection manuelle.

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)

Les méthodes save et load sérialisent la file sur disque : sans persistance, un redémarrage du worker efface les tâches en attente. Sur un pipeline durable, appelez save régulièrement ou remplacez la file en mémoire par Redis.

Modèle 3 : basculer en mode dégradé

Quand les échecs s'enchaînent, continuer à marteler l'API est contre-productif. Le mode dégradé compte les échecs consécutifs ; passé un seuil, il se coupe pour une durée fixe et applique une action de repli, avant de retenter automatiquement. C'est la logique de disjoncteur appliquée à la résolution.

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

La méthode _degraded_action expose trois choix : ignorer la page (Option A), remettre la tâche en file (Option B) ou rediriger vers un solveur de secours (Option C). Choisissez selon la criticité de la tâche.

Node.js : combiner file d'attente et mode dégradé

En Node.js, les deux mécanismes fusionnent dans une seule classe. Un échec ERROR_ZERO_BALANCE déclenche une coupure longue (10 minutes) ; cinq échecs consécutifs déclenchent une coupure courte (5 minutes). Dans les deux cas, les tâches sont mises en file puis rejouées à la reprise via drainRetryQueue.

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

En production : alertes et périmètre

Deux compléments rendent ces modèles exploitables. D'abord, l'alerte : un passage en mode dégradé ou une ERROR_ZERO_BALANCE doit remonter vers votre canal d'astreinte (Slack, e-mail, PagerDuty), sans quoi le pipeline tourne silencieusement à vide. Ensuite, la persistance : sur des workers hébergés chez OVHcloud, Scaleway ou en région AWS eu-west-3 (Paris), la file doit survivre aux redéploiements. Enfin, si votre flux collecte des données personnelles, minimisez ce que la file conserve et vérifiez vos obligations RGPD.

Dépannage

FAQ

Combien de nouvelles tentatives prévoir avant d'abandonner une tâche ?

Trois tentatives couvrent la plupart des échecs transitoires sans gaspiller de crédit. Au-delà, la cause est probablement permanente : envoyez la tâche vers une file de lettres mortes plutôt que de la rejouer indéfiniment.

Où stocker la file d'attente pour ne rien perdre au redémarrage ?

Sérialisez-la sur disque (méthodes save/load du modèle 2) ou dans Redis pour un pipeline multi-worker. Une file uniquement en mémoire disparaît à chaque redéploiement.

Faut-il prévoir un solveur de secours en mode dégradé ?

Utile pour un flux critique, rarement indispensable. Dans la majorité des cas, mettre en file et rejouer à la reprise suffit ; réservez une API tierce aux traitements sans tolérance à la latence.

Comment être averti quand le solde CaptchaAI tombe à zéro ?

Traitez ERROR_ZERO_BALANCE comme un événement d'astreinte : déclenchez une alerte immédiate (Slack, e-mail) au lieu d'une simple relance. Aucun backoff ne rechargera le compte à votre place.

Créez une automatisation CAPTCHA résiliente avec CaptchaAI

Obtenez votre clé API sur captchaai.com et branchez ces modèles de repli sur votre propre flux de résolution.

Guides associés

• le modèle de disjoncteur pour les appels d'API CAPTCHA
• la file de lettres mortes pour les tâches en échec
• implémenter une logique de nouvelle tentative avec l'API CaptchaAI