Construindo uma fila de resolução CAPTCHA em Python com CaptchaAI

Resolver um CAPTCHA de cada vez não escala. O reCAPTCHA v2 pode levar até 60 segundos para ser resolvido, e mil páginas em série somam quase 17 horas só de espera pela API. A saída é separar o envio das tarefas da consulta dos resultados — uma fila. Com o padrão certo, dezenas ou centenas de CAPTCHAs são resolvidos em paralelo, limitados apenas pelo número de threads do seu plano CaptchaAI. Este guia mostra quatro implementações prontas para produção, da mais simples (threading) até a fila com prioridade.

Por que resolver CAPTCHA em fila muda o resultado

Enviar e esperar, um de cada vez, é o padrão mais lento possível: a maior parte do tempo o processo fica parado aguardando a resposta de res.php, sem fazer nada. Uma fila resolve isso separando responsabilidades:

• Envia todos os CAPTCHAs para in.php sem esperar a resposta do anterior
• Consulta (polling) vários IDs de tarefa em paralelo até cada um retornar status: 1
• Faz nova tentativa automaticamente quando uma tarefa falha, sem intervenção manual
• Limita a simultaneidade para não ultrapassar as threads do seu plano CaptchaAI
• Acompanha o progresso e dispara callbacks quando cada tarefa termina

Se os workers rodam fora do Brasil, o RTT até a API soma alguns segundos por chamada; hospedar os workers em uma região próxima, como sa-east-1 (São Paulo) na AWS, reduz essa latência quando o volume é alto.

Fila com threading: paralelismo simples para código síncrono

1. Use este padrão se o resto do projeto já é síncrono — cada worker roda em uma Thread própria
2. Puxa tarefas de uma Queue padrão do Python e chama a API de forma independente, sem bloquear os outros workers no res.php
3. Simples de depurar e suficiente até algumas dezenas de workers simultâneos

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]}")

Fila assíncrona com asyncio: mais throughput por menos overhead

Use asyncio quando threads do sistema operacional começam a pesar: uma única thread com Semaphore controla quantas tarefas estão em voo — menos overhead por tarefa, mais fácil escalar para centenas de CAPTCHAs. Escolha natural em stacks já assíncronas (aiohttp, FastAPI).

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())

Padrão produtor-consumidor para scraping contínuo

1. O produce descobre e enfileira as tarefas conforme o crawl avança, sem esperar que ele termine
2. Um número fixo de consume lê de uma asyncio.Queue limitada e resolve os CAPTCHAs
3. O tamanho máximo da fila evita que a memória cresça sem controle durante o crawl

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())

Fila com prioridade: o checkout não pode esperar

Nem toda tarefa tem a mesma urgência: um Cloudflare Turnstile no checkout de um fluxo de QA autorizado não pode esperar atrás de cem páginas de catálogo. A PriorityQueue resolve isso — cada tarefa carrega um número de prioridade (menor valor primeiro) e os workers puxam sempre a mais urgente primeiro.

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())

Monitoramento e métricas: enxergando a fila em produção

Meça pelo menos quatro números antes de ir para produção: tarefas enviadas, resolvidas, com falha, e o throughput por minuto. Um objeto de métricas simples como o abaixo já basta para perceber uma queda na taxa de sucesso antes que ela vire um incidente maior.

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"
        )

Solução de problemas comuns na fila

Os sintomas abaixo aparecem com frequência em filas de CAPTCHA em produção. Na maioria dos casos, a causa está na simultaneidade configurada ou em uma exceção não tratada no worker.

Perguntas frequentes

Quantas threads minha fila pode usar ao mesmo tempo?

A CaptchaAI limita a simultaneidade pelo plano, não pela sua fila: BASIC (US$ 15/mês, 5 threads) e STANDARD (US$ 30/mês, 15 threads) já bastam para testes; para volume maior, suba para ADVANCE (US$ 90/mês, 50 threads) ou além, até VIP-3 (US$ 7500/mês, 5000 threads). Configure max_workers (threading) ou max_concurrent (asyncio) próximo do número de threads do plano — passar disso só devolve mais ERROR_NO_SLOT_AVAILABLE.

Threading ou asyncio: qual escolher em um projeto novo?

Use asyncio: ele lida com a espera por I/O — a chamada à API, o polling — de forma muito mais leve que threads do sistema operacional, e já é o padrão em stacks como FastAPI e aiohttp. Threading continua sendo a opção certa quando você está integrando a fila a uma base de código síncrona existente e reescrever tudo em asyncio não compensa.

O worker travou sem lançar erro — o que verificar primeiro?

Confirme que o loop do worker está dentro de um try/except: uma exceção não tratada mata a thread silenciosamente e a tarefa fica presa na fila para sempre. Depois, olhe os timeouts do requests ou do aiohttp — uma chamada sem timeout configurado tem o mesmo efeito de travar o worker indefinidamente.

Como evito perder tarefas se o processo cair no meio da fila?

Registre o task_id retornado por in.php assim que ele chega, antes de começar o polling. Se o processo cair, você consulta res.php com os IDs já enviados em vez de reenviar tarefas que talvez já tenham sido resolvidas do lado da CaptchaAI.

Preciso de bibliotecas além de requests e aiohttp?

Não, para reproduzir os exemplos deste guia: requests cobre a fila com threading e aiohttp cobre as versões assíncronas — ambas fazem parte do ecossistema padrão de Python para chamadas HTTP e não exigem nenhuma dependência adicional.

Resumo

O padrão que você escolhe importa menos que o princípio por trás dele: nunca espere uma resposta de CAPTCHA por vez. Separe o envio da consulta, dimensione os workers pelas threads do seu plano na CaptchaAI e meça antes de aumentar a simultaneidade. Comece pelo threading se o resto do código já é síncrono, migre para asyncio à medida que o volume cresce, e reserve o produtor-consumidor para crawls contínuos.

Artigos relacionados

• Guia completo de Playwright com CaptchaAI em Python
• Como construir pipelines de CAPTCHA para clientes com a CaptchaAI
• Automação responsável com a CaptchaAI