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

aiohttp est l'un des meilleurs choix quand vous devez lancer un grand nombre de requêtes HTTP non bloquantes en Python. Si votre pipeline doit aussi gérer des CAPTCHA, l'approche synchrone devient vite un goulot d'étranglement. En combinant aiohttp et CaptchaAI, vous pouvez soumettre plusieurs tâches, poller les résultats en parallèle et garder une boucle d'événements fluide.

Ce guide montre comment structurer un client async minimal, résoudre un CAPTCHA unique, lancer des batchs en concurrence et garder le contrôle du débit avec un sémaphore.

Prérequis

pip install aiohttp

Client async CaptchaAI

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 seul CAPTCHA

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

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 a un flux de scraping

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

Quand tout passe en async, le risque n'est plus le blocage mais l'excès de concurrence. Un sémaphore vous aide à garder un volume soutenable pour votre application, votre budget et votre pipeline de scraping.

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

Exemple avec Turnstile

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

Pourquoi utiliser aiohttp plutôt que httpx ?

aiohttp reste une référence solide pour les charges à forte concurrence en Python. httpx fonctionne aussi très bien, mais si vous cherchez un modèle éprouvé pour du polling async et des gros volumes de requêtes, aiohttp est un excellent point de départ.

Combien de résolutions parallèles puis-je lancer ?

Cela dépend de votre budget, du type de CAPTCHA et du reste de votre pipeline. CaptchaAI accepte un niveau de concurrence élevé, mais il reste pertinent de fixer un plafond applicatif avec un sémaphore.

Puis-je réutiliser la même session pour plusieurs résolutions ?

Oui, et c'est recommandé. Une session aiohttp réutilisée conserve les pools de connexions et réduit le coût des requêtes suivantes.

Guides connexes

• Intégration HTTPX + CaptchaAI
• Resolution parallele de CAPTCHA
• Intégration Scrapy + CaptchaAI

Prochaines étapes

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