File d'attente de résolution CAPTCHA en Python avec CaptchaAI

Une file d'attente est la façon la plus efficace de résoudre des centaines de CAPTCHA en Python sans bloquer votre script sur chaque réponse. Elle sépare l'envoi des tâches de la récupération des tokens : vous soumettez tout d'un coup, puis plusieurs workers interrogent les résultats en parallèle. Le débit n'est plus limité par le temps de résolution d'un seul CAPTCHA, mais par le nombre de threads de votre plan CaptchaAI et par la vitesse de résolution propre à chaque type.

Pourquoi une file d'attente accélère la résolution à grande échelle

Résoudre un CAPTCHA à la fois gaspille l'essentiel du temps à attendre la réponse de l'API. Sur un scraping de plusieurs centaines de pages — le catalogue d'une boutique e-commerce européenne, par exemple — cette attente cumulée se compte en heures. Une file d'attente lève ce goulot d'étranglement :

elle soumet tous les CAPTCHA immédiatement, sans attendre la réponse du précédent ;
elle interroge plusieurs ID de tâches en parallèle (le polling) ;
elle relance automatiquement les tentatives échouées ;
elle plafonne la simultanéité pour rester sous les limites de débit de l'API ;
elle expose un suivi de progression et des callbacks pour brancher votre traitement en aval.

La concurrence utile est bornée par vos threads : le plan BASIC ($15/mois, 5 threads) autorise 5 résolutions en vol, ADVANCE ($90/mois, 50 threads) en autorise 50. Inutile de lancer 200 workers sur un plan à 5 threads — vous ne feriez qu'accumuler des erreurs ERROR_NO_SLOT_AVAILABLE.

Pour dimensionner votre file sans saturer l'API :

partez du nombre de threads de votre plan comme plafond de simultanéité ;
lancez autant de workers que de threads, jamais davantage ;
surveillez ERROR_NO_SLOT_AVAILABLE et réduisez la cadence dès qu'il apparaît.

Quelle approche choisir ?

Trois modèles couvrent la quasi-totalité des besoins. Ce tableau résume quand privilégier chacun avant d'entrer dans le code.

Approche	Quand l'utiliser	Paramètre de simultanéité
File threadée	Code synchrone existant à faire évoluer	`max_workers`
asyncio + sémaphore	Nouveau projet, charge I/O-bound	`max_concurrent`
Producteur-consommateur	Pages découvertes en continu par un crawler	`num_consumers`

Les sections suivantes détaillent chaque modèle.

File d'attente threadée : le point de départ

Si votre code existant est synchrone, la Queue de la bibliothèque standard suffit. Chaque worker tourne dans son propre thread : il prend une tâche, la résout, puis dépose le résultat dans une seconde file. La classe ci-dessous encapsule ce schéma — submit() empile les tâches, start() lance les workers, wait() bloque jusqu'à la fin, et get_results() récupère les tokens résolus.

import time
import threading
import requests
from queue import Queue, Empty

API_KEY = "YOUR_API_KEY"


class CaptchaQueue:
    """Thread-based CAPTCHA solving queue."""

    def __init__(self, api_key, max_workers=10):
        self.api_key = api_key
        self.task_queue = Queue()
        self.result_queue = Queue()
        self.max_workers = max_workers
        self.workers = []

    def submit(self, method, callback=None, **params):
        """Add a CAPTCHA task to the queue."""
        task = {
            "method": method,
            "params": params,
            "callback": callback,
        }
        self.task_queue.put(task)

    def start(self):
        """Start worker threads."""
        for _ in range(self.max_workers):
            t = threading.Thread(target=self._worker, daemon=True)
            t.start()
            self.workers.append(t)

    def wait(self):
        """Wait for all tasks to complete."""
        self.task_queue.join()

    def get_results(self):
        """Get all available results."""
        results = []
        while not self.result_queue.empty():
            try:
                results.append(self.result_queue.get_nowait())
            except Empty:
                break
        return results

    def _worker(self):
        while True:
            try:
                task = self.task_queue.get(timeout=1)
            except Empty:
                continue

            try:
                result = self._solve(task["method"], **task["params"])
                entry = {"status": "solved", "result": result, "task": task}
                self.result_queue.put(entry)
                if task["callback"]:
                    task["callback"](result)
            except Exception as e:
                entry = {"status": "error", "error": str(e), "task": task}
                self.result_queue.put(entry)
            finally:
                self.task_queue.task_done()

    def _solve(self, method, **params):
        submit = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }, timeout=30).json()

        if submit.get("status") != 1:
            raise Exception(f"Submit error: {submit.get('request')}")

        task_id = submit["request"]
        for _ in range(30):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }, timeout=30).json()
            if result.get("status") == 1:
                return result["request"]
            if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                raise Exception("CAPTCHA unsolvable")
        raise TimeoutError("Solve timed out")


# Usage
queue = CaptchaQueue(API_KEY, max_workers=5)
queue.start()

# Submit multiple CAPTCHAs
urls_and_sitekeys = [
    ("https://example.com/page1", "SITEKEY_1"),
    ("https://example.com/page2", "SITEKEY_2"),
    ("https://example.com/page3", "SITEKEY_3"),
]

for url, sitekey in urls_and_sitekeys:
    queue.submit("userrecaptcha", googlekey=sitekey, pageurl=url)

queue.wait()
results = queue.get_results()
print(f"Solved {len(results)} CAPTCHAs")
for r in results:
    print(f"  {r['status']}: {r.get('result', r.get('error', ''))[:50]}")

File d'attente asynchrone avec asyncio

Pour un nouveau projet, asyncio est plus efficace : la résolution de CAPTCHA est une charge I/O-bound — on passe le temps à attendre le réseau —, exactement le terrain de jeu des coroutines.

Contrôler la simultanéité avec un sémaphore

Un asyncio.Semaphore plafonne le nombre de résolutions simultanées et remplace la gestion manuelle des threads, tandis qu'aiohttp porte les requêtes vers in.php et res.php. Réglez sa valeur sur les threads de votre plan.

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"


class AsyncCaptchaQueue:
    """Async CAPTCHA solving queue with concurrency control."""

    def __init__(self, api_key, max_concurrent=10):
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = []

    async def solve_batch(self, tasks):
        """Solve a batch of CAPTCHA tasks concurrently."""
        coros = [self._solve_task(task) for task in tasks]
        self.results = await asyncio.gather(*coros, return_exceptions=True)
        return self.results

    async def _solve_task(self, task):
        async with self.semaphore:
            return await self._solve(task["method"], **task["params"])

    async def _solve(self, method, **params):
        async with aiohttp.ClientSession() as session:
            # Submit
            async with session.post("https://ocr.captchaai.com/in.php", data={
                "key": self.api_key, "method": method, "json": 1, **params,
            }) as resp:
                data = await resp.json(content_type=None)
                if data.get("status") != 1:
                    raise Exception(f"Submit error: {data.get('request')}")
                task_id = data["request"]

            # Poll
            for _ in range(30):
                await asyncio.sleep(5)
                async with session.get("https://ocr.captchaai.com/res.php", params={
                    "key": self.api_key, "action": "get", "id": task_id, "json": 1,
                }) as resp:
                    result = await resp.json(content_type=None)
                    if result.get("status") == 1:
                        return result["request"]
                    if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                        raise Exception("CAPTCHA unsolvable")

            raise TimeoutError("Solve timed out")


# Usage
async def main():
    queue = AsyncCaptchaQueue(API_KEY, max_concurrent=5)

    tasks = [
        {"method": "userrecaptcha", "params": {"googlekey": f"SITEKEY_{i}", "pageurl": f"https://example.com/page{i}"}}
        for i in range(10)
    ]

    results = await queue.solve_batch(tasks)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i}: ERROR — {result}")
        else:
            print(f"Task {i}: {result[:50]}...")


asyncio.run(main())

Modèle producteur-consommateur pour le scraping continu

Quand les pages sont découvertes au fil de l'eau — un crawler qui alimente la file au fur et à mesure —, le modèle producteur-consommateur s'impose. Un producteur pousse les tâches dans une asyncio.Queue bornée, plusieurs consommateurs les résolvent en continu, et une valeur sentinelle None signale l'arrêt.

Minimiser les données collectées

Un scraping continu croise souvent des données personnelles. Appliquez la minimisation RGPD : ne conservez que les champs strictement nécessaires et vérifiez vos obligations avant toute journalisation.

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"


class ProducerConsumerQueue:
    """Continuous CAPTCHA solving with producer-consumer pattern."""

    def __init__(self, api_key, queue_size=100, num_consumers=5):
        self.api_key = api_key
        self.queue = asyncio.Queue(maxsize=queue_size)
        self.num_consumers = num_consumers
        self.solved_count = 0
        self.error_count = 0
        self.running = True

    async def produce(self, tasks):
        """Producer: feed CAPTCHA tasks into the queue."""
        for task in tasks:
            await self.queue.put(task)
        # Signal consumers to stop
        for _ in range(self.num_consumers):
            await self.queue.put(None)

    async def consume(self, result_handler):
        """Consumer: solve CAPTCHAs and call result handler."""
        async with aiohttp.ClientSession() as session:
            while True:
                task = await self.queue.get()
                if task is None:
                    self.queue.task_done()
                    break

                try:
                    result = await self._solve(session, task["method"], **task["params"])
                    self.solved_count += 1
                    if result_handler:
                        await result_handler(task, result)
                except Exception as e:
                    self.error_count += 1
                    print(f"Error: {e}")
                finally:
                    self.queue.task_done()

    async def run(self, tasks, result_handler=None):
        """Run the producer-consumer pipeline."""
        # Start producer
        producer = asyncio.create_task(self.produce(tasks))

        # Start consumers
        consumers = [
            asyncio.create_task(self.consume(result_handler))
            for _ in range(self.num_consumers)
        ]

        # Wait for everything to finish
        await producer
        await asyncio.gather(*consumers)

        print(f"Complete: {self.solved_count} solved, {self.error_count} errors")

    async def _solve(self, session, method, **params):
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }) as resp:
            data = await resp.json(content_type=None)
            if data.get("status") != 1:
                raise Exception(f"Submit: {data.get('request')}")
            task_id = data["request"]

        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }) as resp:
                result = await resp.json(content_type=None)
                if result.get("status") == 1:
                    return result["request"]
        raise TimeoutError("Timed out")


# Usage
async def handle_result(task, token):
    url = task["params"]["pageurl"]
    print(f"Solved for {url}: {token[:30]}...")


async def main():
    queue = ProducerConsumerQueue(API_KEY, num_consumers=5)

    tasks = [
        {"method": "userrecaptcha", "params": {"googlekey": f"SITEKEY_{i}", "pageurl": f"https://example.com/page{i}"}}
        for i in range(20)
    ]

    await queue.run(tasks, result_handler=handle_result)


asyncio.run(main())

File d'attente prioritaire : traiter d'abord ce qui compte

Toutes les tâches ne se valent pas. Sur une boutique en ligne, le CAPTCHA d'une page de paiement (checkout) doit passer avant celui d'une page produit, elle-même prioritaire sur une page d'information. Une asyncio.PriorityQueue ordonne les tâches par niveau — plus le nombre est bas, plus la priorité est haute.

import asyncio
from dataclasses import dataclass, field

API_KEY = "YOUR_API_KEY"


@dataclass(order=True)
class PriorityTask:
    priority: int
    task: dict = field(compare=False)


class PriorityCaptchaQueue:
    """CAPTCHA queue with priority levels."""

    def __init__(self, api_key, num_workers=5):
        self.api_key = api_key
        self.queue = asyncio.PriorityQueue()
        self.num_workers = num_workers
        self.results = {}

    async def submit(self, task_id, method, priority=5, **params):
        """Submit with priority (lower number = higher priority)."""
        await self.queue.put(PriorityTask(
            priority=priority,
            task={"id": task_id, "method": method, "params": params},
        ))

    async def process(self):
        """Process all queued tasks by priority."""
        workers = [asyncio.create_task(self._worker()) for _ in range(self.num_workers)]

        # Wait for queue to drain
        await self.queue.join()

        # Cancel workers
        for w in workers:
            w.cancel()

        return self.results

    async def _worker(self):
        import aiohttp
        async with aiohttp.ClientSession() as session:
            while True:
                item = await self.queue.get()
                task = item.task
                try:
                    result = await self._solve(session, task["method"], **task["params"])
                    self.results[task["id"]] = {"status": "solved", "token": result}
                except Exception as e:
                    self.results[task["id"]] = {"status": "error", "error": str(e)}
                finally:
                    self.queue.task_done()

    async def _solve(self, session, method, **params):
        import aiohttp
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }) as resp:
            data = await resp.json(content_type=None)
            if data.get("status") != 1:
                raise Exception(data.get("request"))
            task_id = data["request"]

        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }) as resp:
                result = await resp.json(content_type=None)
                if result.get("status") == 1:
                    return result["request"]
        raise TimeoutError()


# Usage
async def main():
    pq = PriorityCaptchaQueue(API_KEY, num_workers=3)

    # High priority — checkout pages
    await pq.submit("checkout_1", "turnstile", priority=1, sitekey="KEY", pageurl="https://shop.com/checkout")

    # Normal priority — product pages
    for i in range(5):
        await pq.submit(f"product_{i}", "userrecaptcha", priority=5, googlekey="KEY", pageurl=f"https://shop.com/p/{i}")

    # Low priority — info pages
    for i in range(3):
        await pq.submit(f"info_{i}", "userrecaptcha", priority=10, googlekey="KEY", pageurl=f"https://shop.com/info/{i}")

    results = await pq.process()
    for task_id, result in results.items():
        print(f"{task_id}: {result['status']}")


asyncio.run(main())

Surveiller le débit et le taux de réussite

Une file en production a besoin de métriques : nombre de tâches soumises, résolues et échouées, temps moyen de résolution et débit par minute. La classe QueueMetrics calcule le taux de réussite et le débit à la volée — des chiffres prêts à être poussés vers vos logs ou votre tableau de bord.

Astuce : publiez report() à intervalle régulier dans votre journalisation pour repérer une baisse de débit avant qu'elle ne devienne un incident.

import time
from dataclasses import dataclass, field


@dataclass
class QueueMetrics:
    submitted: int = 0
    solved: int = 0
    failed: int = 0
    total_solve_time: float = 0.0
    start_time: float = field(default_factory=time.time)

    @property
    def avg_solve_time(self):
        return self.total_solve_time / self.solved if self.solved else 0

    @property
    def success_rate(self):
        total = self.solved + self.failed
        return (self.solved / total * 100) if total else 0

    @property
    def throughput(self):
        elapsed = time.time() - self.start_time
        return self.solved / elapsed * 60 if elapsed > 0 else 0

    def report(self):
        return (
            f"Submitted: {self.submitted} | "
            f"Solved: {self.solved} | "
            f"Failed: {self.failed} | "
            f"Avg time: {self.avg_solve_time:.1f}s | "
            f"Success: {self.success_rate:.1f}% | "
            f"Throughput: {self.throughput:.0f}/min"
        )

Dépannage

Symptôme	Cause	Correctif
La file grossit mais les tâches ne se terminent jamais	Trop de workers saturent l'API	Réduisez `max_workers` / `max_concurrent`
`ERROR_NO_SLOT_AVAILABLE`	Limite de simultanéité de l'API atteinte (threads du plan)	Espacez les soumissions ou passez à un plan avec plus de threads
Tâches bloquées dans la file	Un worker est mort sur une exception non gérée	Enveloppez la boucle du worker dans un `try`/`except`
La mémoire grimpe au fil du temps	Résultats jamais consommés	Appelez `get_results()` régulièrement
La file asyncio se fige	`await` oublié sur un appel asynchrone	Vérifiez que chaque coroutine est bien attendue

Questions fréquentes

Combien de threads puis-je exécuter en parallèle ?

Autant que votre plan en autorise. La concurrence est bornée par les threads : BASIC ($15/mois, 5 threads) permet 5 résolutions simultanées, ADVANCE ($90/mois, 50 threads) en permet 50. CaptchaAI gère la file côté serveur ; dès que vous voyez ERROR_NO_SLOT_AVAILABLE, vous avez dépassé votre allocation.

Que se passe-t-il quand une tâche échoue ou expire ?

Le worker capture l'exception, enregistre l'erreur dans la file de résultats et passe à la tâche suivante — la file ne s'arrête pas. Pour les échecs transitoires, réinjectez la tâche avec un backoff exponentiel plutôt que de la relancer immédiatement.

Puis-je mettre en file des types de CAPTCHA différents en même temps ?

Oui. Chaque tâche porte son propre paramètre method, donc une même file peut mélanger reCAPTCHA v2, reCAPTCHA v3, Cloudflare Turnstile et GeeTest v3. Le token renvoyé varie selon le type ; adaptez simplement votre traitement en aval au type soumis.

Comment répartir la file sur plusieurs machines ?

Remplacez la file en mémoire par un broker partagé comme Redis ou Kafka : les producteurs y déposent les tâches, des workers déployés sur plusieurs machines (par exemple chez OVHcloud ou Scaleway) les consomment. La logique de résolution reste identique ; seule la source de la file change.

En résumé

Une file d'attente de résolution CAPTCHA sépare la soumission de l'interrogation et débloque la résolution en parallèle avec CaptchaAI. Choisissez le threading pour intégrer du code synchrone existant, asyncio pour un projet Python moderne, et le modèle producteur-consommateur pour un scraping continu. Dans tous les cas, calez votre nombre de workers sur les threads de votre plan.