Construyendo una cola de resolución de CAPTCHA en Python con CaptchaAI

¿Necesitas resolver cientos de CAPTCHA sin que tu scraper se quede esperando uno por uno? La respuesta es una cola. En lugar de enviar un CAPTCHA, bloquear el proceso hasta que llegue el token y recién entonces pasar al siguiente, una cola separa el envío de la tarea de la recuperación del resultado. Así envías todo de golpe y sondeas los resultados en paralelo, aprovechando que CaptchaAI resuelve del lado del servidor mientras tu código sigue trabajando.

Esta guía recorre cuatro formas de construir esa cola en Python —threads, asyncio, productor-consumidor y prioridades— más un módulo de métricas. Todos los ejemplos usan la API de CaptchaAI con los endpoints in.php y res.php.

Por qué una cola cambia el rendimiento

Resolver un CAPTCHA de forma síncrona desperdicia el tiempo en espera: envías la tarea y tu hilo queda inactivo entre cinco y quince segundos hasta que llega el token. Multiplícalo por cien tareas y tienes minutos perdidos. Una cola bien diseñada te da:

• Envío inmediato de todas las tareas, sin esperar a que termine la anterior
• Sondeo de varios task ID en paralelo
• Reintentos automáticos de las resoluciones que fallan
• Control de concurrencia para respetar el límite de solicitudes de la API
• Seguimiento del progreso y callbacks cuando cada token llega

El ciclo común: enviar y sondear

Todos los patrones comparten el mismo ciclo de dos pasos:

1. Envías la tarea con un POST a in.php, que devuelve un task ID.
2. Consultas res.php cada pocos segundos hasta que el estado sea 1 y el campo request contenga el token.

La diferencia entre una cola lenta y una rápida no está en ese ciclo, sino en cuántos corres a la vez: cada clase repite esa lógica en _solve.

Cola básica con threads

Es el enfoque más directo cuando tu código ya es síncrono. Un grupo de hilos worker toma tareas de una Queue, resuelve cada CAPTCHA y deja el resultado en una segunda cola. max_workers fija cuántas resoluciones ocurren en paralelo: empieza bajo y súbelo mientras la API no se sature.

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

Cola asíncrona con asyncio

Para proyectos nuevos, asyncio suele ser más eficiente: la resolución de CAPTCHA es I/O-bound (casi todo es espera de red), justo donde asyncio brilla. Un asyncio.Semaphore cumple el papel de max_workers y limita cuántas corrutinas golpean la API a la vez. Con return_exceptions=True, un fallo aislado no tumba el lote entero.

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

Patrón productor-consumidor para scraping continuo

Los dos patrones anteriores asumen que conoces todas las tareas de antemano. En un scraping continuo las páginas se descubren sobre la marcha: un crawler encuentra URLs nuevas mientras los workers todavía resuelven las viejas. Ahí encaja el patrón productor-consumidor: el productor alimenta la cola con tareas nuevas y un grupo de consumidores las resuelve en paralelo. El valor centinela None indica a cada consumidor que puede detenerse cuando ya no queda trabajo.

Haz scraping solo de sitios y datos que tengas permiso de consultar, y respeta los términos de servicio y la normativa de protección de datos aplicable.

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

Cola de prioridad

No todos los CAPTCHA valen lo mismo: en un flujo de QA propio, el CAPTCHA del checkout que estás validando importa más que el de una página de catálogo. Una asyncio.PriorityQueue atiende primero las tareas con el número de prioridad más bajo, así las críticas no esperan detrás del trabajo rutinario.

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

Monitoreo y métricas

Una cola sin métricas es una caja negra. Este dataclass registra lo esencial —enviadas, resueltas, fallidas— y calcula tasa de éxito, tiempo medio y throughput por minuto.

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

Cuándo elegir cada enfoque

Los cuatro patrones resuelven el mismo problema desde ángulos distintos, y puedes combinarlos:

Threads, throughput y costo en CaptchaAI

En CaptchaAI, un thread es un CAPTCHA en curso: mientras se resuelve ocupa un thread y, al terminar, queda libre para el siguiente. Por eso max_workers o el semáforo no deberían superar los threads de tu plan; si lo hacen, verás ERROR_NO_SLOT_AVAILABLE.

La correspondencia entre concurrencia y plan es directa:

1. BASIC ($15/mes, 5 threads): pruebas y volúmenes pequeños
2. STANDARD ($30/mes, 15 threads): un scraper de un solo proyecto
3. ADVANCE ($90/mes, 50 threads): hasta 50 resoluciones en paralelo
4. PREMIUM ($170/mes, 100 threads): pipelines de producción con picos

Cada plan incluye resoluciones ilimitadas por thread durante el mes, sin cargo por CAPTCHA ni recargo por tipo. Para una agencia en Ciudad de México o un freelancer que factura en pesos, ese costo fijo mensual en USD es más predecible que pagar por resolución.

Solución de problemas

Preguntas frecuentes

¿Cómo sé cuántos threads necesita mi volumen?

Depende de tu tiempo de resolución y de tu objetivo por hora. Cada thread resuelve un CAPTCHA a la vez en ciclos de cinco a quince segundos, así que calcula cuántas caben en paralelo y elige el plan cuyos threads lo cubran.

¿La cola reintenta sola los CAPTCHA que fallan?

Sí, si lo programas. Los ejemplos capturan las excepciones; para reintentar, vuelve a poner la tarea fallida en la cola con un contador de intentos y un retroceso exponencial.

¿Sirve la misma cola para Turnstile o GeeTest v3?

Sí. La cola es agnóstica al tipo: solo cambias el parámetro method (turnstile, geetest, userrecaptcha) y los parámetros propios de cada CAPTCHA. CaptchaAI resuelve reCAPTCHA v2/v3, Cloudflare Turnstile y Challenge, GeeTest v3 e imagen/OCR con el mismo flujo.

¿Qué hago si una resolución tarda demasiado?

Cada método _solve corta a los 30 intentos con esperas de cinco segundos: unos dos minutos y medio en total. Ajusta ese límite si tu caso lo necesita; al superarse, la tarea lanza un TimeoutError que puedes reintentar o descartar.

Resumen

Una cola separa el envío del sondeo y convierte decenas de esperas secuenciales en resoluciones paralelas. Elige threads si tu código ya es síncrono, asyncio para Python moderno, productor-consumidor para scraping continuo y prioridad cuando algunas tareas no pueden esperar. Combínalo con los threads adecuados de CaptchaAI y el throughput dejará de ser tu cuello de botella.

Artículos relacionados

• Crea un pipeline de pruebas automatizadas con CaptchaAI
• Cómo construir pipelines de CAPTCHA para clientes con CaptchaAI