Résoudre des CAPTCHA en parallèle avec ThreadPoolExecutor en Python

Un lot de 200 CAPTCHA traités les uns après les autres immobilise votre script près d'une heure. Réparti sur dix threads, le même lot se termine en cinq à six minutes, sans toucher à votre logique métier. ThreadPoolExecutor, livré avec la bibliothèque standard de Python, suffit pour cela : une résolution passe l'essentiel de son temps à attendre le réseau, et c'est la charge pour laquelle un pool de threads est conçu.

Le parcours ci-dessous part d'un solveur synchrone bâti sur l'API CaptchaAI, puis l'industrialise : dimensionnement du pool, session par thread, timeouts, progression.

Pourquoi un pool de threads suffit pour la résolution de CAPTCHA

La résolution est une charge I/O-bound : le code envoie une tâche, puis attend. Pendant cette attente, l'interpréteur libère le GIL, donc les threads travaillent réellement en parallèle. Le plafond n'est pas votre machine, mais le nombre de résolutions simultanées autorisées par votre plan.

Calez max_workers sur les threads de votre plan

Chez CaptchaAI, un thread correspond à une résolution en cours : la facturation porte sur cette simultanéité, chaque thread acceptant un nombre illimité de résolutions dans le mois. Les paliers, en dollars US : BASIC ($15/mois, 5 threads), STANDARD ($30/mois, 15 threads), ADVANCE ($90/mois, 50 threads), PREMIUM ($170/mois, 100 threads).

Ouvrir 50 workers sur un plan qui en autorise 5 n'accélère rien. Démarrez à 10, montez par paliers, surveillez le taux d'erreur.

Étape 1 : le solveur synchrone, puis son premier pool

La fonction ci-dessous traite un seul CAPTCHA : elle envoie la tâche à in.php, puis interroge res.php toutes les 5 secondes jusqu'au token. Le pool s'ajoute autour, as_completed traitant chaque résultat dès son arrivée.

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def solve_captcha(sitekey, pageurl):
    """Synchronous CAPTCHA solve — submit and poll."""
    # Submit
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        raise RuntimeError(data.get("request", "Submit failed"))

    captcha_id = data["request"]

    # Poll for result
    for _ in range(60):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": captcha_id,
            "json": 1
        }).json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(result.get("request", "Unknown error"))

    raise TimeoutError("Solve timeout after 300s")


# Batch solve with ThreadPoolExecutor
tasks = [
    {"sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "pageurl": f"https://example.com/page/{i}"}
    for i in range(20)
]

start = time.time()

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {
        executor.submit(solve_captcha, t["sitekey"], t["pageurl"]): t
        for t in tasks
    }

    solved = 0
    failed = 0

    for future in as_completed(futures):
        task = futures[future]
        try:
            solution = future.result()
            solved += 1
            print(f"[OK] {task['pageurl']}: {solution[:30]}...")
        except Exception as e:
            failed += 1
            print(f"[ERR] {task['pageurl']}: {e}")

elapsed = time.time() - start
print(f"\nDone: {solved} solved, {failed} failed in {elapsed:.1f}s")

Le try/except autour de future.result() n'est pas décoratif : sans lui, la première exception remonte dans la boucle et vous perdez le décompte des tâches restantes.

Étape 2 : une session HTTP par thread

Une connexion TCP neuve à chaque requête coûte plusieurs allers-retours TLS, multipliés par deux appels et par tâche. Une requests.Session rangée dans un threading.local() donne à chaque thread la sienne, sans partage concurrent.

import threading

# Thread-local storage for sessions
thread_local = threading.local()


def get_session():
    """Get or create a thread-local session."""
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
        # Configure connection pooling
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=10,
            pool_maxsize=10,
            max_retries=2
        )
        thread_local.session.mount("https://", adapter)
    return thread_local.session


def solve_captcha_pooled(sitekey, pageurl):
    """Solve using thread-local connection pooling."""
    session = get_session()

    resp = session.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        raise RuntimeError(data.get("request"))

    captcha_id = data["request"]

    for _ in range(60):
        time.sleep(5)
        result = session.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": captcha_id,
            "json": 1
        }).json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(result.get("request"))

    raise TimeoutError("Solve timeout")

Gardez le pool de connexions de l'HTTPAdapter au moins égal à max_workers : en dessous, vos threads se disputent les mêmes sockets.

Étape 3 : executor.map() quand seul le lot compte

Quand la gestion d'erreur tient dans le worker lui-même, map() est plus court et conserve l'ordre d'entrée — pratique si vous réinjectez les tokens dans une liste d'URL ordonnée.

def solve_task(task):
    """Wrapper that returns result dict."""
    try:
        solution = solve_captcha_pooled(task["sitekey"], task["pageurl"])
        return {"url": task["pageurl"], "solution": solution, "error": None}
    except Exception as e:
        return {"url": task["pageurl"], "solution": None, "error": str(e)}


with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_task, tasks))

solved = [r for r in results if r["solution"]]
failed = [r for r in results if r["error"]]
print(f"Solved: {len(solved)}, Failed: {len(failed)}")

Étape 4 : deux niveaux de timeout

Un thread bloqué occupe une place dans le pool aussi longtemps que le processus vit. Posez deux garde-fous : un timeout global sur as_completed, un timeout par tâche sur future.result().

from concurrent.futures import TimeoutError as FuturesTimeout

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {
        executor.submit(solve_captcha_pooled, t["sitekey"], t["pageurl"]): t
        for t in tasks
    }

    for future in as_completed(futures, timeout=600):  # 10 min global timeout
        task = futures[future]
        try:
            solution = future.result(timeout=120)  # 2 min per task
            print(f"[OK] {task['pageurl']}")
        except FuturesTimeout:
            print(f"[TIMEOUT] {task['pageurl']}")
        except Exception as e:
            print(f"[ERR] {task['pageurl']}: {e}")

Un plafond de 120 s par tâche libère vite un thread perdu ; le timeout global borne la durée du lot dans votre ordonnanceur.

Étape 5 : un callback de progression lisible

Sur plusieurs centaines de tâches, l'absence de retour visuel fait croire à un script figé. Un compteur protégé par un threading.Lock évite l'affichage entrelacé.

import threading

progress_lock = threading.Lock()
progress = {"done": 0, "total": 0}


def solve_with_progress(task):
    result = solve_task(task)
    with progress_lock:
        progress["done"] += 1
        pct = progress["done"] / progress["total"] * 100
        print(f'\r  Progress: {progress["done"]}/{progress["total"]} ({pct:.0f}%)', end="")
    return result


progress["total"] = len(tasks)

with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_with_progress, tasks))

print()  # Newline after progress

ThreadPoolExecutor ou asyncio : comment trancher

# ThreadPoolExecutor — drop into existing sync code
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_task, tasks))

# asyncio — requires async function chain
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [solve_async(session, t) for t in task_list]
        results = await asyncio.gather(*tasks)

Restez sur ThreadPoolExecutor si :

• votre base de code est synchrone ;
• vous dépendez de bibliothèques qui ignorent l'async (Selenium, certains ORM) ;
• vous voulez du parallélisme sans refonte.

Passez à asyncio si :

• vous démarrez de zéro ;
• l'efficacité maximale compte (moins de threads système) ;
• vous êtes déjà dans un framework asynchrone (FastAPI, aiohttp).

Cas concret : 3 000 formulaires qualifiés chaque nuit

Une équipe QA lyonnaise contrôle chaque nuit 3 000 formulaires de son propre catalogue depuis un serveur OVHcloud (une instance AWS eu-west-3 à Paris ferait le même travail). En séquentiel, à 15 s de résolution moyenne, le lot dépasse douze heures et déborde de la fenêtre de nuit ; avec 25 workers alignés sur un plan qui les autorise, il tombe à une trentaine de minutes.

Côté conformité : journalisez l'identifiant de tâche et l'horodatage, pas le contenu des formulaires, et limitez les données personnelles conservées dans vos logs, conformément au RGPD.

Dépannage

FAQ

Combien de threads mon plan CaptchaAI autorise-t-il en parallèle ?

Selon le palier : 5 threads sur BASIC ($15/mois), 15 sur STANDARD ($30/mois), 50 sur ADVANCE ($90/mois), jusqu'à 5 000 sur VIP-3 ($7,500/mois). max_workers se règle sur ce nombre.

Que se passe-t-il si max_workers dépasse les threads de mon plan ?

Rien ne casse, mais les résolutions au-delà du quota attendent leur tour pendant que vos timeouts par tâche s'écoulent, ce qui gonfle le taux d'échec apparent.

Puis-je piloter Selenium depuis un ThreadPoolExecutor ?

Oui, à condition qu'un driver n'appartienne qu'à un seul thread : une instance WebDriver n'est pas thread-safe. Stockez-la dans un threading.local(), comme la session de l'étape 2.

Comment rejouer uniquement les tâches en échec ?

Collectez les résultats dont la clé error n'est pas vide, puis relancez-les dans un second pool plus petit, avec un backoff exponentiel. En cas d'échec répété, vérifiez le sitekey et l'URL de la page.

ThreadPoolExecutor permet-il de traiter hCaptcha en parallèle ?

Non — hCaptcha et FunCaptcha ne sont pas pris en charge par CaptchaAI, et GeeTest v4 est annoncé comme à venir. Le modèle vaut pour les types pris en charge : reCAPTCHA v2 et v3, Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3 et les CAPTCHA image ou OCR.

Prochaines étapes

Parallélisez votre file dès le prochain lot : récupérez votre clé API CaptchaAI et branchez le pool sur votre pipeline.

Guides associés :

• La résolution de CAPTCHA en parallèle, vue d'ensemble
• Comparer parallèle et séquentiel sur un même lot
• Tenir 10 000 tâches par heure