La résolution de CAPTCHA peut échouer : délais d'attente, paramètres incorrects, solde nul ou limites de débit. Si votre automatisation plante au premier échec, vous perdez toute progression. Une dégradation progressive maintient le pipeline en marche : ignorez, réessayez, mettez en file d'attente ou revenez à des alternatives.
Comment choisir entre ignorer, réessayer ou mettre en file d'attente ?
| Type d'échec | Réponse la plus utile | Pourquoi |
|---|---|---|
| Erreur ponctuelle ou délai d'attente isolé | Réessayer | Le contexte reste souvent récupérable |
| Mauvais paramètre ou mauvaise clé de site | Arrêter puis corriger l'extraction | Réessayer ne fera qu'amplifier le bruit et le coût |
| Solde nul ou dépendance indisponible | Mettre en pause et alerter | Il faut une action externe avant de repartir |
| Tâche utile mais non urgente | Mettre en file d'attente | Vous conservez la progression sans bloquer tout le pipeline |
| Tâche non critique dans un lot volumineux | Ignorer et continuer | Le coût d'un blocage global dépasse la valeur d'un seul item |
Modes de défaillance
| Échec | Code d'erreur | Stratégie de rétablissement |
|---|---|---|
| Délai d'attente | CAPCHA_NOT_READY (dépassé les sondages) |
Réessayez avec un nouveau défi |
| Mauvais paramètres | ERROR_BAD_PARAMETERS |
Connectez-vous et ignorez - correction de l'extraction |
| Mauvaise clé de site | ERROR_WRONG_GOOGLEKEY |
Réextraire la clé du site |
| Solde nul | ERROR_ZERO_BALANCE |
Pause, alerte, attendez la recharge |
| Tarif limité | ERROR_TOO_MUCH_REQUESTS |
Reculez de façon exponentielle |
| API en panne | Erreur de connexion | Disjoncteur + réessai |
Modèle 1 : Sauter et continuer
Pour les opérations par lots où des échecs individuels sont acceptables :
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
Modèle 2 : file d'attente des nouvelles tentatives
Les tâches ayant échoué sont placées dans une file d'attente de nouvelles tentatives pour un traitement ultérieur :
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)
Modèle 3 : mode dégradé
Lorsque le service de résolution n'est pas disponible, passez en mode limité :
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 : modèle combiné
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');
}
}
Dépannage
| Problème | Parce que | Corriger |
|---|---|---|
| Toutes les tâches ignorées | Mode dégradé déclenché de manière trop agressive | Augmenter le seuil de défaillance |
| La file d'attente des nouvelles tentatives s'allonge pour toujours | Les tâches ne réussissent jamais | Définir le nombre maximum de tentatives ; passer à la file d'attente des lettres mortes |
| Récupération trop lente | Délai d'attente dégradé long | Réduisez le temps de récupération ; ajouter une sonde de vérification de l'état |
| Tâches en file d'attente perdues au redémarrage | File d'attente en mémoire | Conserver la file d'attente vers un fichier ou une base de données |
FAQ
Quelle est la différence entre une dégradation gracieuse et un disjoncteur ?
Un disjoncteur empêche complètement les appels lorsque des pannes sont détectées. La dégradation gracieuse est plus large : elle inclut des comportements de repli, une logique de saut et des flux de travail alternatifs. Ils fonctionnent bien ensemble.
Dois-je toujours réessayer les tâches ayant échoué ?
Pas pour ERROR_BAD_PARAMETERS ou ERROR_WRONG_GOOGLEKEY – ceux-ci ne réussiront pas lors d’une nouvelle tentative. Réessayez uniquement les erreurs transitoires telles que les délais d'attente et les limites de débit.
Créez une automatisation CAPTCHA résiliente avec CaptchaAI
Obtenez votre clé API surcaptchaai.com.
Guides associés
- Modèle de disjoncteur pour les appels d'API CAPTCHA
- File d'attente de lettres mortes pour les tâches ayant échoué
- Implémentation de la logique de nouvelle tentative
