Résoudre plusieurs CAPTCHA en lot avec CaptchaAI

Pour traiter plusieurs CAPTCHA sans ralentir votre pipeline, ne les résolvez pas un par un : soumettez toutes les tâches d'un coup, interrogez leurs résultats en parallèle et récupérez chaque token dès qu'il est prêt. Avec l'API CaptchaAI, cela tient en quelques dizaines de lignes de Python.

Le vrai enjeu est l'orchestration : séparer l'envoi des tâches, le suivi des task_id et la collecte des résultats. Pour une équipe qui scrape des données publiques depuis des workers OVHcloud ou Scaleway, le séquentiel fait patienter chaque page ; le parallèle ramène le temps total à celui d'une seule résolution. Pensez à minimiser les données personnelles collectées (RGPD).

Ce guide construit trois briques réutilisables, de la plus simple à la plus performante :

• la concurrence par threads, pour démarrer vite sur un petit lot ;
• une version asyncio, quand le volume grimpe à plusieurs centaines de tâches ;
• le découplage soumission/polling, pour aller chercher le débit maximal.

Aucune ne demande d'infrastructure particulière : un script Python, votre clé API et une file de tâches suffisent.

Séquentiel ou parallèle : où se gagne le débit

En séquentiel, trois résolutions à 15 s prennent environ 45 s ; lancées ensemble, une quinzaine de secondes.

Sequential (slow):
  Submit #1 → Poll → Result (15s)
  Submit #2 → Poll → Result (15s)
  Submit #3 → Poll → Result (15s)
  Total: ~45s for 3 solves

Parallel (fast):
  Submit #1 ─┐
  Submit #2 ─┤→ Poll all → Results arrive
  Submit #3 ─┘
  Total: ~15s for 3 solves

Concurrence par threads : le point de départ

Pour des lots modestes, un ThreadPoolExecutor suffit : chaque thread soumet une tâche puis interroge son résultat. Le paramètre max_workers fixe le nombre de résolutions simultanées — commencez à 5, puis montez.

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

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


def submit_task(method, **params):
    """Submit a single CAPTCHA task."""
    data = {"key": API_KEY, "method": method, "json": 1}
    data.update(params)
    resp = requests.post(f"{BASE_URL}/in.php", data=data, timeout=30)
    result = resp.json()
    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")
    return result["request"]


def poll_result(task_id, timeout=120):
    """Poll until result is ready."""
    start = time.time()
    while time.time() - start < timeout:
        time.sleep(5)
        resp = requests.get(f"{BASE_URL}/res.php", params={
            "key": API_KEY, "action": "get",
            "id": task_id, "json": 1,
        }, timeout=15)
        data = resp.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError(f"Task {task_id} timeout")


def solve_one(sitekey, pageurl):
    """Submit and poll a single task."""
    task_id = submit_task("userrecaptcha", googlekey=sitekey, pageurl=pageurl)
    token = poll_result(task_id)
    return {"url": pageurl, "token": token}


def batch_solve(tasks, max_workers=10):
    """Solve multiple CAPTCHAs in parallel."""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(solve_one, t["sitekey"], t["url"]): t
            for t in tasks
        }
        for future in as_completed(futures):
            task = futures[future]
            try:
                result = future.result()
                results.append(result)
                print(f"Solved: {result['url']}")
            except Exception as e:
                print(f"Failed: {task['url']} - {e}")
                results.append({"url": task["url"], "token": None, "error": str(e)})

    return results


# Usage
tasks = [
    {"sitekey": "SITE_KEY_1", "url": "https://example.com/page1"},
    {"sitekey": "SITE_KEY_2", "url": "https://example.com/page2"},
    {"sitekey": "SITE_KEY_3", "url": "https://example.com/page3"},
]

results = batch_solve(tasks, max_workers=5)
print(f"Solved {sum(1 for r in results if r.get('token'))}/{len(tasks)}")

Votre débit reste plafonné par les threads de votre plan CaptchaAI : cinq résolutions en parallèle sur BASIC ($15/mois, 5 threads), cinquante sur ADVANCE ($90/mois, 50 threads). Au-delà de ce quota, les soumissions patientent.

Passer à asyncio pour les gros volumes

Au-delà de quelques dizaines de tâches, les threads coûtent cher en mémoire. asyncio avec aiohttp gère des centaines de connexions sur une seule boucle d'événements ; un Semaphore borne la concurrence pour rester sous la limite de votre plan. Ce modèle prend l'avantage quand :

• vous dépassez la cinquantaine de tâches simultanées ;
• la latence réseau domine le temps de résolution ;
• vous voulez limiter l'empreinte mémoire de vos workers.

import asyncio
import aiohttp
import time

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


async def submit_task_async(session, method, **params):
    data = {"key": API_KEY, "method": method, "json": 1}
    data.update(params)
    async with session.post(f"{BASE_URL}/in.php", data=data) as resp:
        result = await resp.json()
        if result.get("status") != 1:
            raise RuntimeError(f"Submit error: {result.get('request')}")
        return result["request"]


async def poll_result_async(session, task_id, timeout=120):
    start = time.time()
    while time.time() - start < timeout:
        await asyncio.sleep(5)
        params = {
            "key": API_KEY, "action": "get",
            "id": task_id, "json": 1,
        }
        async with session.get(f"{BASE_URL}/res.php", params=params) as resp:
            data = await resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                return data["request"]
    raise TimeoutError(f"Task {task_id} timeout")


async def solve_one_async(session, sitekey, pageurl):
    task_id = await submit_task_async(
        session, "userrecaptcha",
        googlekey=sitekey, pageurl=pageurl,
    )
    token = await poll_result_async(session, task_id)
    return {"url": pageurl, "token": token}


async def batch_solve_async(tasks, max_concurrent=20):
    """Solve many CAPTCHAs concurrently with asyncio."""
    semaphore = asyncio.Semaphore(max_concurrent)
    results = []

    async def solve_with_limit(task):
        async with semaphore:
            try:
                result = await solve_one_async(
                    session, task["sitekey"], task["url"],
                )
                return result
            except Exception as e:
                return {"url": task["url"], "token": None, "error": str(e)}

    async with aiohttp.ClientSession() as session:
        coros = [solve_with_limit(t) for t in tasks]
        results = await asyncio.gather(*coros)

    return results


# Usage
tasks = [
    {"sitekey": "KEY", "url": f"https://example.com/page{i}"}
    for i in range(50)
]

results = asyncio.run(batch_solve_async(tasks, max_concurrent=20))
solved = sum(1 for r in results if r.get("token"))
print(f"Solved: {solved}/{len(tasks)}")

Séparer la soumission du polling

Pour aller plus loin, découplez les phases : envoyez d'abord toutes les tâches, collectez leurs task_id, puis lancez une seule boucle d'interrogation. Une résolution lente ne bloque alors plus tout le lot.

import requests
import time

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"


def batch_submit(tasks):
    """Submit all tasks first, return task IDs."""
    submitted = []
    for task in tasks:
        try:
            data = {
                "key": API_KEY,
                "method": "userrecaptcha",
                "googlekey": task["sitekey"],
                "pageurl": task["url"],
                "json": 1,
            }
            resp = requests.post(f"{BASE_URL}/in.php", data=data, timeout=30)
            result = resp.json()
            if result.get("status") == 1:
                submitted.append({
                    "task_id": result["request"],
                    "url": task["url"],
                })
            time.sleep(0.1)  # Brief delay between submits
        except Exception as e:
            print(f"Submit failed for {task['url']}: {e}")
    return submitted


def batch_poll(submitted, timeout=120):
    """Poll all submitted tasks until complete."""
    pending = {s["task_id"]: s for s in submitted}
    results = []
    start = time.time()

    while pending and time.time() - start < timeout:
        time.sleep(5)
        for task_id in list(pending.keys()):
            try:
                resp = requests.get(f"{BASE_URL}/res.php", params={
                    "key": API_KEY, "action": "get",
                    "id": task_id, "json": 1,
                }, timeout=15)
                data = resp.json()
                if data["request"] != "CAPCHA_NOT_READY":
                    info = pending.pop(task_id)
                    results.append({
                        "url": info["url"],
                        "token": data["request"],
                    })
            except Exception:
                pass

    # Mark remaining as failed
    for task_id, info in pending.items():
        results.append({"url": info["url"], "token": None, "error": "timeout"})

    return results


# Usage
tasks = [
    {"sitekey": "KEY", "url": f"https://example.com/page{i}"}
    for i in range(20)
]

submitted = batch_submit(tasks)
print(f"Submitted {len(submitted)} tasks")

results = batch_poll(submitted)
solved = sum(1 for r in results if r.get("token"))
print(f"Solved: {solved}/{len(tasks)}")

Choisir votre niveau de concurrence

Les chiffres ci-dessous reposent sur des mesures observées et des retours d'utilisateurs. Les résultats varient selon l'environnement, le volume et le moment de la journée.

Pour calibrer sans tâtonner :

• restez sous le nombre de threads facturés sur votre plan CaptchaAI ;
• augmentez la concurrence par paliers plutôt que d'un seul bond ;
• surveillez le taux d'erreurs 429 : c'est votre signal de saturation.

Suivre chaque résultat sans les mélanger

Ce qui casse le plus de pipelines par lots n'est pas la vitesse, c'est l'association entre un résultat et sa tâche. Trois règles suffisent à la garantir :

1. Indexez chaque tâche par son task_id dès la soumission, jamais par son rang dans la liste d'envoi.
2. Conservez l'URL (ou l'identifiant métier) à côté du task_id, pour retrouver le contexte quand le résultat arrive.
3. Traitez un échec comme un résultat à part entière : un token à None accompagné d'un code d'erreur, pas une exception avalée en silence.

Ce petit contrat rend le lot rejouable : en cas de coupure réseau, vous relancez uniquement les tâches en échec au lieu de repasser tout le lot.

Dépannage du traitement par lots

Questions fréquentes

Comment éviter les erreurs 429 sur un gros lot ?

Espacez les envois — un délai de 100 ms suffit souvent — et bornez la concurrence avec un Semaphore ou un max_workers.

Quel plan CaptchaAI pour 50 résolutions en parallèle ?

Il vous faut au moins autant de threads que de résolutions en vol. Pour 50 en parallèle, visez ADVANCE ($90/mois, 50 threads) ; BASIC ($15/mois, 5 threads) plafonnera à 5.

Puis-je mélanger reCAPTCHA v2 et v3 dans le même lot ?

Oui. Chaque tâche est indépendante : reCAPTCHA v2, reCAPTCHA v3 et Cloudflare Turnstile peuvent cohabiter dans le même lot. Indexez les résultats par task_id.

Le traitement par lots coûte-t-il plus cher ?

Non. La facturation se fait au thread simultané, pas à la résolution : un lot ne coûte pas plus que les mêmes tâches lancées une à une. Vous gagnez seulement du temps.

À retenir

Le traitement par lots ne change rien à la façon dont un CAPTCHA est résolu : il change la façon dont vous orchestrez des dizaines de résolutions en même temps.

• Démarrez avec un ThreadPoolExecutor et cinq à dix tâches en parallèle.
• Passez à asyncio dès que le volume ou la latence l'exigent.
• Découplez soumission et polling pour lisser le débit, et indexez tout par task_id.
• Calez la concurrence sur les threads de votre plan, jamais au-dessus.

Guides connexes

• Gérer la limitation de débit et les réponses 429
• Mettre en place une logique de nouvelle tentative

Prêt à traiter vos lots à haute cadence ? Essayez CaptchaAI et mesurez votre propre débit en parallèle.