ThreadPoolExecutor en Python para resolver CAPTCHA en paralelo

Cuando un scraper tiene que resolver cientos de CAPTCHA por lote, hacerlo de uno en uno desperdicia casi todo el tiempo esperando respuestas de red. ThreadPoolExecutor ataca justo ese cuello de botella: paraleliza las llamadas a la API de CaptchaAI dentro de tu código síncrono actual, sin reescribir la cadena de llamadas como asyncio. Vas a construir el patrón por capas:

• Un pool de hilos básico que resuelve un lote completo.
• Reutilización de conexiones con una Session por hilo.
• Protección con tiempos de espera y seguimiento de progreso.
• Cómo dimensionar los workers según tu plan de threads.

Por qué ThreadPoolExecutor encaja para resolver CAPTCHA

Resolver un CAPTCHA es una carga I/O-bound: tu código pasa la mayor parte del tiempo esperando la respuesta HTTP de CaptchaAI, no consumiendo CPU. Durante esa espera, Python libera el GIL, así que varios hilos avanzan de verdad en paralelo. Por eso ThreadPoolExecutor rinde tan bien aquí y, además, entra en proyectos que ya tienes escritos sin tocar su arquitectura:

Implementación básica del pool de hilos

El patrón es simple: una función síncrona que envía la tarea y sondea el resultado, y un ThreadPoolExecutor que la ejecuta sobre todo el lote. Con as_completed procesas cada resultado en cuanto está listo:

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def solve_captcha(sitekey, pageurl):
    """Synchronous CAPTCHA solve — submit and poll."""
    # Submit
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        raise RuntimeError(data.get("request", "Submit failed"))

    captcha_id = data["request"]

    # Poll for result
    for _ in range(60):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": captcha_id,
            "json": 1
        }).json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(result.get("request", "Unknown error"))

    raise TimeoutError("Solve timeout after 300s")


# Batch solve with ThreadPoolExecutor
tasks = [
    {"sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "pageurl": f"https://example.com/page/{i}"}
    for i in range(20)
]

start = time.time()

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {
        executor.submit(solve_captcha, t["sitekey"], t["pageurl"]): t
        for t in tasks
    }

    solved = 0
    failed = 0

    for future in as_completed(futures):
        task = futures[future]
        try:
            solution = future.result()
            solved += 1
            print(f"[OK] {task['pageurl']}: {solution[:30]}...")
        except Exception as e:
            failed += 1
            print(f"[ERR] {task['pageurl']}: {e}")

elapsed = time.time() - start
print(f"\nDone: {solved} solved, {failed} failed in {elapsed:.1f}s")

Con max_workers=10 mantienes diez resoluciones en vuelo a la vez. El diccionario futures te deja recuperar la tarea original de cada resultado, algo clave cuando un lote mezcla varias páginas.

Reutiliza conexiones con una Session por hilo

Abrir una conexión TCP nueva en cada solicitud tira rendimiento a la basura. Comparte un requests.Session por hilo con almacenamiento thread-local, de modo que cada worker reutilice su pool de conexiones sin pisarse con los demás:

import threading

# Thread-local storage for sessions
thread_local = threading.local()


def get_session():
    """Get or create a thread-local session."""
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
        # Configure connection pooling
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=10,
            pool_maxsize=10,
            max_retries=2
        )
        thread_local.session.mount("https://", adapter)
    return thread_local.session


def solve_captcha_pooled(sitekey, pageurl):
    """Solve using thread-local connection pooling."""
    session = get_session()

    resp = session.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        raise RuntimeError(data.get("request"))

    captcha_id = data["request"]

    for _ in range(60):
        time.sleep(5)
        result = session.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": captcha_id,
            "json": 1
        }).json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(result.get("request"))

    raise TimeoutError("Solve timeout")

El almacenamiento thread-local es la pieza clave: cada worker recibe su propia Session la primera vez que la pide, y así el pool de conexiones nunca se comparte entre hilos.

map() para lotes sencillos

Si no necesitas manejar errores tarea por tarea, executor.map() es más directo. Envuelve la resolución en una función que devuelve siempre un dict, y así ningún fallo aislado rompe el lote completo:

def solve_task(task):
    """Wrapper that returns result dict."""
    try:
        solution = solve_captcha_pooled(task["sitekey"], task["pageurl"])
        return {"url": task["pageurl"], "solution": solution, "error": None}
    except Exception as e:
        return {"url": task["pageurl"], "solution": None, "error": str(e)}


with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_task, tasks))

solved = [r for r in results if r["solution"]]
failed = [r for r in results if r["error"]]
print(f"Solved: {len(solved)}, Failed: {len(failed)}")

A diferencia de as_completed, map() conserva el orden de entrada: los resultados salen alineados con la lista tasks.

Protección con tiempos de espera

Un hilo colgado en una página lenta puede bloquear el pool entero. Ponle dos límites: uno global sobre as_completed y otro por tarea sobre future.result():

from concurrent.futures import TimeoutError as FuturesTimeout

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {
        executor.submit(solve_captcha_pooled, t["sitekey"], t["pageurl"]): t
        for t in tasks
    }

    for future in as_completed(futures, timeout=600):  # 10 min global timeout
        task = futures[future]
        try:
            solution = future.result(timeout=120)  # 2 min per task
            print(f"[OK] {task['pageurl']}")
        except FuturesTimeout:
            print(f"[TIMEOUT] {task['pageurl']}")
        except Exception as e:
            print(f"[ERR] {task['pageurl']}: {e}")

Seguimiento del progreso en tiempo real

En un lote largo conviene ver cuánto lleva completado. Un contador protegido por un Lock te da una barra de progreso segura entre hilos:

import threading

progress_lock = threading.Lock()
progress = {"done": 0, "total": 0}


def solve_with_progress(task):
    result = solve_task(task)
    with progress_lock:
        progress["done"] += 1
        pct = progress["done"] / progress["total"] * 100
        print(f'\r  Progress: {progress["done"]}/{progress["total"]} ({pct:.0f}%)', end="")
    return result


progress["total"] = len(tasks)

with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_with_progress, tasks))

print()  # Newline after progress

ThreadPoolExecutor frente a asyncio

Ambos paralelizan I/O, pero cuestan esfuerzos distintos de adoptar. ThreadPoolExecutor se cuela en código síncrono existente; asyncio te obliga a que toda la cadena sea async:

# ThreadPoolExecutor — drop into existing sync code
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_task, tasks))

# asyncio — requires async function chain
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [solve_async(session, t) for t in task_list]
        results = await asyncio.gather(*tasks)

Elige ThreadPoolExecutor cuando:

• Tu base de código actual es síncrona.
• Usas bibliotecas sin soporte async (Selenium, algunos ORM).
• Quieres paralelismo rápido sin reestructurar nada.

Elige asyncio cuando:

• Construyes desde cero.
• Buscas la máxima eficiencia (menos hilos del sistema operativo).
• Ya trabajas en un framework async (FastAPI, aiohttp).

Cómo elegir el número de workers

Cada worker equivale a una resolución simultánea, y ahí entra un detalle de facturación útil: CaptchaAI cobra por thread concurrente, con resoluciones ilimitadas por thread. Conviene alinear max_workers con los threads de tu plan para no pedir más concurrencia de la que pagas. Por ejemplo, el plan BASIC ($15/mes, 5 threads) da margen para 5 workers, STANDARD ($30/mes, 15 threads) para 15, y ADVANCE ($90/mes, 50 threads) sostiene los 50 workers de la fila inferior:

Más workers implican más conexiones simultáneas a la API. Empieza en 10 y súbelo mientras vigilas la tasa de error.

Resolución de problemas

Preguntas frecuentes

¿Cuántos workers debería usar según mi plan de CaptchaAI?

Tantos como threads incluya tu plan. Como CaptchaAI factura por thread concurrente con resoluciones ilimitadas, poner max_workers por encima de tus threads solo genera cola, no más velocidad. Con ADVANCE ($90/mes, 50 threads) puedes llegar a 50 workers; si empiezas, BASIC ($15/mes, 5 threads) cubre lotes pequeños sin desperdicio.

¿Funciona ThreadPoolExecutor junto a Selenium?

Sí, y es uno de sus mejores usos. Selenium es síncrono y no encaja con asyncio, así que un pool de hilos es la vía natural para resolver varios CAPTCHA mientras controlas varias sesiones de navegador en paralelo.

¿Cómo evito saturar la API con demasiadas conexiones?

Combina un max_workers acorde a tus threads con agrupación de conexiones (HTTPAdapter con pool_maxsize) y reintentos moderados. Si aparecen picos de ConnectionError, reduce workers antes de subirlos de nuevo.

¿El GIL limita el rendimiento real de los hilos?

No para este caso. En trabajo I/O-bound como las solicitudes HTTP y time.sleep, Python libera el GIL y tus hilos avanzan de verdad en paralelo durante las llamadas de red. El GIL solo frena el paralelismo ligado a CPU.

Siguientes pasos

Paraleliza tu resolución de CAPTCHA: consigue tu API key de CaptchaAI y suelta ThreadPoolExecutor dentro de tu pipeline.

Guías relacionadas:

• Resolución de CAPTCHA en paralelo
• Rendimiento en paralelo frente a secuencial
• Cómo resolver 10.000 tareas por hora