Résoudre des CAPTCHA de manière asynchrone avec aiohttp et CaptchaAI

Pour traiter plusieurs CAPTCHA à la fois en Python, la solution n'est pas de multiplier les threads : c'est d'arrêter d'attendre. Chaque résolution passe l'essentiel de son temps à patienter pendant que CaptchaAI travaille, et une boucle asyncio sait justement mettre ce temps mort à profit pour envoyer puis interroger des dizaines de tâches simultanément. aiohttp est la brique HTTP non bloquante qui rend ce schéma naturel.

Ce guide construit un petit client asynchrone de bout en bout : vous verrez comment soumettre une tâche, interroger le résultat, lancer un lot en parallèle, brancher la résolution sur un flux de scraping et plafonner la concurrence avec un sémaphore. Chaque étape s'appuie sur la même API in.php / res.php, donc le code reste court.

Prérequis

pip install aiohttp

Un client CaptchaAI asynchrone avec aiohttp

Tout part d'une petite classe qui encapsule les deux appels de l'API : submit envoie la tâche et récupère son identifiant, poll interroge le résultat toutes les cinq secondes jusqu'à un délai limite. La méthode solve enchaîne les deux, et get_balance vérifie le solde avant de lancer un gros volume.

import aiohttp
import asyncio


class AsyncCaptchaAI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"

    async def submit(self, session, params):
        """Submit a CAPTCHA task and return the task ID."""
        params["key"] = self.api_key
        async with session.get(
            f"{self.base_url}/in.php", params=params
        ) as resp:
            text = await resp.text()

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        return text.split("|")[1]

    async def poll(self, session, task_id, timeout=300):
        """Poll for the result with a timeout."""
        params = {
            "key": self.api_key,
            "action": "get",
            "id": task_id,
        }
        deadline = asyncio.get_event_loop().time() + timeout

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)

            async with session.get(
                f"{self.base_url}/res.php", params=params
            ) as resp:
                text = await resp.text()

            if text == "CAPCHA_NOT_READY":
                continue
            if text.startswith("OK|"):
                return text.split("|", 1)[1]
            raise Exception(f"Solve failed: {text}")

        raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

    async def solve(self, session, params, timeout=300):
        """Submit and poll in one call."""
        task_id = await self.submit(session, params)
        return await self.poll(session, task_id, timeout)

    async def get_balance(self, session):
        """Check account balance."""
        params = {"key": self.api_key, "action": "getbalance"}
        async with session.get(
            f"{self.base_url}/res.php", params=params
        ) as resp:
            return float(await resp.text())

Résoudre un premier CAPTCHA

Commencez par un cas isolé pour valider votre clé et votre solde. On ouvre une session aiohttp, on lit le solde, puis on résout un reCAPTCHA v2 en passant method, googlekey et pageurl. Le token renvoyé est celui que votre formulaire attend dans le champ g-recaptcha-response.

import asyncio
import os

async def main():
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Check balance
        balance = await solver.get_balance(session)
        print(f"Balance: ${balance:.2f}")

        # Solve reCAPTCHA v2
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": "https://example.com",
        })
        print(f"Token: {token[:50]}...")

asyncio.run(main())

Résoudre plusieurs CAPTCHA en parallèle

C'est ici que l'asynchrone paie. Au lieu d'attendre chaque token l'un après l'autre, vous construisez une liste de coroutines et vous les confiez à asyncio.gather. Le paramètre return_exceptions=True évite qu'une seule URL en échec fasse tomber tout le lot : chaque résultat est ensuite trié individuellement.

async def solve_batch(urls, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        tasks = [
            solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })
            for url in urls
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"FAILED {url}: {result}")
            else:
                print(f"SOLVED {url}: {len(result)} chars")

        return results


urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
    "https://example.com/page4",
    "https://example.com/page5",
]
asyncio.run(solve_batch(urls, "6Le-wvkS..."))

Ajouter la résolution CAPTCHA à un pipeline de scraping

En pratique, vous ne résolvez pas un CAPTCHA pour le plaisir : vous le résolvez pour débloquer une requête. Imaginez un worker de veille tarifaire déployé sur Scaleway ou OVHcloud à Paris qui parcourt des centaines de fiches produits. Le flux ci-dessous ne déclenche la résolution que lorsque la page renvoie réellement un g-recaptcha, puis renvoie le formulaire avec le token. Pensez à minimiser les données personnelles collectées et à vérifier vos obligations RGPD avant de stocker quoi que ce soit.

async def scrape_with_captcha(url, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Fetch the page
        async with session.get(url) as resp:
            html = await resp.text()

        # Check if page has a CAPTCHA
        if "g-recaptcha" not in html:
            return html  # No CAPTCHA, return content

        # Solve the CAPTCHA
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

        # Submit with solved token
        async with session.post(url, data={
            "g-recaptcha-response": token,
        }) as resp:
            return await resp.text()

Contrôler la concurrence avec un sémaphore

Une fois tout passé en async, le risque n'est plus le blocage mais l'excès de requêtes simultanées. Un asyncio.Semaphore fixe un plafond applicatif : vous choisissez combien de résolutions tournent en même temps, en cohérence avec le nombre de threads de votre plan CaptchaAI et la charge que votre code de scraping peut absorber.

async def solve_with_limit(urls, site_key, max_concurrent=10):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
    semaphore = asyncio.Semaphore(max_concurrent)

    async def solve_one(session, url):
        async with semaphore:
            return await solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })

    async with aiohttp.ClientSession() as session:
        tasks = [solve_one(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    solved = sum(1 for r in results if not isinstance(r, Exception))
    print(f"Solved {solved}/{len(urls)} CAPTCHAs")
    return results

Résoudre un Turnstile en asynchrone

Le même client gère les autres types pris en charge sans changement de structure : il suffit d'ajuster method et les paramètres. Pour Cloudflare Turnstile, on passe method=turnstile avec le sitekey et l'URL de la page.

async def solve_turnstile(url, sitekey):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        token = await solver.solve(session, {
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": url,
        })
        return token

Dépannage

FAQ

Comment gérer les timeouts quand je lance des centaines de résolutions en parallèle ?

Gardez le timeout par tâche généreux (300 s dans l'exemple) et laissez le sémaphore lisser le débit. Un délai trop court sous forte charge coupe des résolutions encore valides ; il vaut mieux plafonner la concurrence que raccourcir le délai.

Faut-il un proxy différent pour chaque tâche asynchrone ?

Non, pas pour l'appel à CaptchaAI lui-même : c'est le service qui résout le défi. Un proxy résidentiel ou datacenter concerne uniquement vos propres requêtes de scraping, à répartir selon la cible et votre volume.

aiohttp fonctionne-t-il dans un notebook Jupyter ?

Oui, mais Jupyter gère déjà une boucle d'événements. Appeler asyncio.run y déclenche l'erreur Event loop is closed ; installez nest_asyncio et utilisez await directement dans la cellule.

CaptchaAI facture-t-il à la résolution ou au thread ?

Au thread. Chaque plan inclut des résolutions illimitées par thread sur le mois, et le plus petit palier, BASIC ($15/mois, 5 threads), autorise déjà cinq résolutions simultanées, ce qui cadre bien avec la valeur de max_concurrent.

Guides connexes

• L'intégration de HTTPX avec CaptchaAI
• La résolution parallèle de CAPTCHA
• L'intégration de Scrapy avec CaptchaAI

Prochaines étapes

Si votre code Python enchaîne déjà des appels concurrents, obtenez votre clé CaptchaAI et faites de la résolution CAPTCHA une brique native de votre architecture async plutôt qu'un traitement mis à part.