Résolution de CAPTCHA d'images par lots : traitement de plus de 1 000 images

Pour résoudre plus de 1 000 CAPTCHA d'images sans y passer la journée, la clé tient en une idée : séparez la soumission de l'interrogation des résultats en deux phases asynchrones distinctes. Vous envoyez d'abord toutes les images en rafale contrôlée, puis vous interrogez en parallèle les tâches en attente. Ce guide décrit ce pipeline de bout en bout avec CaptchaAI, en Python puis en Node.js, avec la gestion du débit et le suivi de l'avancement qui manquent presque toujours aux scripts improvisés.

Une boucle séquentielle qui soumet une image, attend la réponse, puis passe à la suivante, plafonne à quelques images par minute : le temps de résolution de chaque image devient un temps mort pour toutes les autres. Le traitement par lots transforme ce goulot d'étranglement en débit soutenu.

Pourquoi deux phases valent mieux qu'une boucle séquentielle

L'architecture repose sur une file d'attente d'images alimentant un groupe de workers de soumission, puis un groupe de workers d'interrogation qui écrivent dans un stockage de résultats. La soumission et l'interrogation ne bloquent jamais ensemble : pendant qu'une image attend sa réponse, les threads restent libres de prendre la suivante.

[Image Queue] → [Submit Workers] → [Poll Workers] → [Results Store]
     ↓                ↓                  ↓                ↓
  1000 images    20 concurrent      Adaptive poll     CSV/JSON output
                   submits           intervals

Concrètement, la phase 1 renvoie un identifiant de tâche (task_id) pour chaque image acceptée. La phase 2 interroge régulièrement chaque task_id jusqu'à obtenir la réponse ou atteindre le timeout. Découpler les deux vous permet de régler indépendamment la concurrence des soumissions et celle du polling, qui n'ont pas les mêmes contraintes.

Le pipeline asynchrone en Python

L'implémentation Python s'appuie sur asyncio et aiohttp. Deux sémaphores bornent la concurrence : un pour les soumissions, un pour les interrogations. Chaque image est encodée en base64, envoyée à l'endpoint in.php, et son task_id est conservé pour la phase de polling.

import asyncio
import aiohttp
import base64
import json
import time
import csv
from pathlib import Path

API_KEY = "YOUR_API_KEY"
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
MAX_CONCURRENT_SUBMITS = 20
MAX_CONCURRENT_POLLS = 30
POLL_INTERVAL = 5


async def submit_image(session, sem, image_path):
    """Submit a single image CAPTCHA."""
    async with sem:
        with open(image_path, "rb") as f:
            img_b64 = base64.b64encode(f.read()).decode()

        data = {
            "key": API_KEY,
            "method": "base64",
            "body": img_b64,
            "json": "1",
        }

        async with session.post(SUBMIT_URL, data=data) as resp:
            result = await resp.json()

        if result["status"] != 1:
            return {"file": str(image_path), "error": result["request"]}

        return {
            "file": str(image_path),
            "task_id": result["request"],
            "submitted_at": time.time(),
        }


async def poll_result(session, sem, task):
    """Poll for a single task result."""
    async with sem:
        for attempt in range(24):
            await asyncio.sleep(POLL_INTERVAL)

            params = {
                "key": API_KEY,
                "action": "get",
                "id": task["task_id"],
                "json": "1",
            }

            async with session.get(RESULT_URL, params=params) as resp:
                result = await resp.json()

            if result["status"] == 1:
                return {
                    "file": task["file"],
                    "task_id": task["task_id"],
                    "answer": result["request"],
                    "solve_time": time.time() - task["submitted_at"],
                }
            if result["request"] != "CAPCHA_NOT_READY":
                return {
                    "file": task["file"],
                    "task_id": task["task_id"],
                    "error": result["request"],
                }

        return {
            "file": task["file"],
            "task_id": task["task_id"],
            "error": "TIMEOUT",
        }


async def process_batch(image_dir, output_file="results.csv"):
    """Process all images in a directory."""
    image_paths = sorted(Path(image_dir).glob("*.png")) + \
                  sorted(Path(image_dir).glob("*.jpg"))

    print(f"Found {len(image_paths)} images")

    submit_sem = asyncio.Semaphore(MAX_CONCURRENT_SUBMITS)
    poll_sem = asyncio.Semaphore(MAX_CONCURRENT_POLLS)

    async with aiohttp.ClientSession() as session:
        # Phase 1: Submit all images
        print("Submitting...")
        submit_tasks = [
            submit_image(session, submit_sem, path)
            for path in image_paths
        ]
        submissions = await asyncio.gather(*submit_tasks)

        # Separate successes and errors
        pending = [s for s in submissions if "task_id" in s]
        errors = [s for s in submissions if "error" in s]
        print(f"Submitted: {len(pending)}, Errors: {len(errors)}")

        # Phase 2: Poll all pending tasks
        print("Polling for results...")
        poll_tasks = [
            poll_result(session, poll_sem, task)
            for task in pending
        ]
        results = await asyncio.gather(*poll_tasks)

    # Combine results
    all_results = results + errors

    # Write to CSV
    with open(output_file, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=[
            "file", "task_id", "answer", "solve_time", "error"
        ])
        writer.writeheader()
        for r in all_results:
            writer.writerow({
                "file": r.get("file", ""),
                "task_id": r.get("task_id", ""),
                "answer": r.get("answer", ""),
                "solve_time": round(r.get("solve_time", 0), 2),
                "error": r.get("error", ""),
            })

    solved = sum(1 for r in results if "answer" in r)
    failed = sum(1 for r in results if "error" in r)
    print(f"Done: {solved} solved, {failed} failed, {len(errors)} submit errors")
    print(f"Results saved to {output_file}")


# Run
asyncio.run(process_batch("./captcha_images"))

Sur un lot de 1 000 images, la sortie ressemble à ceci : quelques soumissions peuvent échouer d'emblée (image illisible, réseau), le reste est interrogé puis écrit dans un CSV exploitable.

Found 1000 images
Submitting...
Submitted: 997, Errors: 3
Polling for results...
Done: 985 solved, 12 failed, 3 submit errors
Results saved to results.csv

La même logique en Node.js avec un pool de workers

Si votre pile est en JavaScript, la classe BatchProcessor reproduit la même séparation soumission/interrogation, mais découpe le travail en tranches (chunks) de la taille de la concurrence. Chaque tranche est traitée avec Promise.all, ce qui borne naturellement le nombre de tâches en vol.

const axios = require('axios');
const fs = require('fs');
const path = require('path');
const { createObjectCsvWriter } = require('csv-writer');

const API_KEY = 'YOUR_API_KEY';
const SUBMIT_URL = 'https://ocr.captchaai.com/in.php';
const RESULT_URL = 'https://ocr.captchaai.com/res.php';
const MAX_CONCURRENT = 20;
const POLL_INTERVAL_MS = 5000;

class BatchProcessor {
  constructor(concurrency = MAX_CONCURRENT) {
    this.concurrency = concurrency;
    this.results = [];
    this.processed = 0;
    this.total = 0;
  }

  async submitImage(imagePath) {
    const imgBase64 = fs.readFileSync(imagePath, { encoding: 'base64' });
    const resp = await axios.post(SUBMIT_URL, null, {
      params: {
        key: API_KEY,
        method: 'base64',
        body: imgBase64,
        json: 1,
      },
    });

    if (resp.data.status !== 1) {
      throw new Error(resp.data.request);
    }
    return resp.data.request;
  }

  async pollResult(taskId) {
    for (let i = 0; i < 24; i++) {
      await new Promise(r => setTimeout(r, POLL_INTERVAL_MS));
      const resp = await axios.get(RESULT_URL, {
        params: { key: API_KEY, action: 'get', id: taskId, json: 1 },
      });

      if (resp.data.status === 1) return resp.data.request;
      if (resp.data.request !== 'CAPCHA_NOT_READY') {
        throw new Error(resp.data.request);
      }
    }
    throw new Error('TIMEOUT');
  }

  async processOne(imagePath) {
    const startTime = Date.now();
    try {
      const taskId = await this.submitImage(imagePath);
      const answer = await this.pollResult(taskId);
      this.processed++;
      const elapsed = ((Date.now() - startTime) / 1000).toFixed(1);
      console.log(`[${this.processed}/${this.total}] ${path.basename(imagePath)}: ${answer} (${elapsed}s)`);
      return { file: imagePath, answer, solveTime: elapsed, error: '' };
    } catch (err) {
      this.processed++;
      return { file: imagePath, answer: '', solveTime: 0, error: err.message };
    }
  }

  async run(imageDir, outputFile = 'results.csv') {
    const files = fs.readdirSync(imageDir)
      .filter(f => /\.(png|jpg|jpeg|gif)$/i.test(f))
      .map(f => path.join(imageDir, f));

    this.total = files.length;
    console.log(`Processing ${this.total} images with ${this.concurrency} workers`);

    // Process in chunks
    for (let i = 0; i < files.length; i += this.concurrency) {
      const chunk = files.slice(i, i + this.concurrency);
      const chunkResults = await Promise.all(
        chunk.map(f => this.processOne(f))
      );
      this.results.push(...chunkResults);
    }

    // Write CSV
    const csvWriter = createObjectCsvWriter({
      path: outputFile,
      header: [
        { id: 'file', title: 'File' },
        { id: 'answer', title: 'Answer' },
        { id: 'solveTime', title: 'Solve Time (s)' },
        { id: 'error', title: 'Error' },
      ],
    });
    await csvWriter.writeRecords(this.results);

    const solved = this.results.filter(r => r.answer).length;
    console.log(`Done: ${solved}/${this.total} solved. Results: ${outputFile}`);
  }
}

const processor = new BatchProcessor(20);
processor.run('./captcha_images');

Garder le débit sous contrôle pour éviter les 429

À forte concurrence, le premier mur que vous rencontrez n'est pas le solveur mais la limitation de débit : trop de soumissions par seconde et l'API renvoie des réponses 429. Un limiteur simple, à fenêtre glissante d'une seconde, lisse les rafales sans brider inutilement le débit.

class RateLimiter:
    def __init__(self, max_per_second=10):
        self.max_per_second = max_per_second
        self.timestamps = []

    async def acquire(self):
        now = time.time()
        self.timestamps = [t for t in self.timestamps if now - t < 1.0]

        if len(self.timestamps) >= self.max_per_second:
            wait = 1.0 - (now - self.timestamps[0])
            if wait > 0:
                await asyncio.sleep(wait)

        self.timestamps.append(time.time())

# Use in submit loop
rate_limiter = RateLimiter(max_per_second=10)

async def submit_with_rate_limit(session, image_path):
    await rate_limiter.acquire()
    # ... submit as before

Réglez max_per_second en observant vos propres réponses : commencez prudemment (par exemple 10 soumissions par seconde) et augmentez tant que vous ne voyez pas de 429.

Suivre l'avancement d'un long traitement

Un lot de plusieurs milliers d'images tourne plusieurs minutes ; sans retour visuel, impossible de savoir s'il progresse ou s'il est bloqué. Un compteur qui affiche le rythme et une estimation du temps restant (ETA) suffit à rendre le traitement observable.

import sys

class ProgressTracker:
    def __init__(self, total):
        self.total = total
        self.completed = 0
        self.solved = 0
        self.failed = 0
        self.start_time = time.time()

    def update(self, success=True):
        self.completed += 1
        if success:
            self.solved += 1
        else:
            self.failed += 1

        elapsed = time.time() - self.start_time
        rate = self.completed / elapsed if elapsed > 0 else 0
        eta = (self.total - self.completed) / rate if rate > 0 else 0

        sys.stdout.write(
            f"\r[{self.completed}/{self.total}] "
            f"Solved: {self.solved} | Failed: {self.failed} | "
            f"Rate: {rate:.1f}/s | ETA: {eta:.0f}s"
        )
        sys.stdout.flush()

Dimensionner les threads et estimer le coût

La facturation CaptchaAI se fait par thread simultané, pas par résolution : chaque plan inclut un nombre de threads et des résolutions illimitées pendant le mois. Un thread correspond à un CAPTCHA en cours ; dès qu'une image est résolue, ce thread reprend la suivante. Votre débit réel est donc borné par le nombre de threads de votre plan, exactement la variable que règlent MAX_CONCURRENT_SUBMITS et MAX_CONCURRENT_POLLS dans le code.

Pour un pipeline qui vise 20 à 30 tâches en vol, visez au minimum le plan ADVANCE ($90/mois, 50 threads) ; PREMIUM ($170/mois, 100 threads) laisse de la marge si vous montez la concurrence. Le plan BASIC ($15/mois, 5 threads) suffit pour tester le script sur quelques dizaines d'images, mais bridera un vrai lot de 1 000. Les CAPTCHA Image/OCR font partie des types les plus rapides et les moins coûteux à traiter, ce qui rend le traitement par lots particulièrement rentable sur ce format. La facturation est en dollars US ; pour la grille complète, consultez la page tarifaire officielle.

Collecte des images et conformité RGPD

Si vos images proviennent de vos propres collectes ou d'un environnement que vous êtes autorisé à automatiser, appliquez le principe de minimisation : ne conservez que les images strictement nécessaires et purgez le CSV de résultats une fois le traitement terminé. Pour les équipes en France, en Belgique ou en Suisse, cela relève des obligations RGPD courantes — vérifiez vos propres obligations avant de stocker des données rattachables à des personnes.

Côté infrastructure, héberger le worker au plus près de vos données réduit la latence réseau qui s'ajoute au temps de résolution : une région européenne comme eu-west-3 (Paris), ou un hébergeur comme OVHcloud ou Scaleway, constitue un point de départ naturel pour un lot lancé depuis l'Europe.

Dépannage

FAQ

Faut-il vraiment tout soumettre avant d'interroger les résultats ?

Oui, c'est ce qui rend le lot rapide. En soumettant d'abord toutes les images, vous laissez le solveur travailler en parallèle pendant que vous interrogez ; une boucle qui attend chaque réponse avant la soumission suivante annule ce gain.

Quel plan CaptchaAI choisir pour traiter des lots ?

Cela dépend de votre concurrence, puisque la facturation est basée sur les threads. Pour 20 à 30 tâches en vol, ADVANCE ($90/mois, 50 threads) est un bon point de départ ; consultez la grille tarifaire pour les autres paliers.

Que faire si beaucoup de tâches renvoient CAPCHA_NOT_READY ?

C'est normal au début du polling : la réponse n'est pas encore prête. Le code réessaie automatiquement à chaque POLL_INTERVAL. Si le message persiste jusqu'au timeout, augmentez le nombre de tentatives ou vérifiez que les images ne sont pas trop complexes.

Comment reprendre un lot interrompu sans tout resoumettre ?

Écrivez les task_id dès la phase de soumission dans un fichier, puis relancez uniquement la phase de polling sur les identifiants sans réponse. Comme chaque task_id reste valide côté serveur, vous évitez de payer une seconde soumission pour des images déjà envoyées.

Traitez des milliers de CAPTCHA avec CaptchaAI

Obtenez votre clé API sur captchaai.com et lancez votre premier lot.

Guides associés

• Résolution simultanée avec asyncio en Python
• Résolution parallèle de CAPTCHA
• Limites de débit et throttling