Pipelines CAPTCHA reutilizables para varios clientes con CaptchaAI

Un pipeline de CAPTCHA reutilizable te ahorra reescribir la misma lógica de resolución en cada proyecto. Si tu agencia gestiona scraping o automatización para varias cuentas, un único sistema modular —que reciba tareas, las envíe a CaptchaAI y guarde los tokens resueltos— se mantiene mucho mejor que un script distinto por cliente. En este tutorial lo construyes paso a paso: la arquitectura, la clase central en Python, una versión equivalente en Node.js, la configuración por cliente y el manejo de errores listo para producción.

Arquitectura del pipeline

La idea es desacoplar quién pide la resolución de quién la ejecuta. Los scrapers de cada cliente encolan tareas; unos workers las envían a CaptchaAI y sondean el resultado; los tokens resueltos quedan en un almacén del que el consumidor los recupera cuando los necesita.

┌──────────────┐    ┌───────────────┐    ┌──────────────┐
│  Client A    │──▶ │               │    │              │
│  Client B    │──▶ │  Task Queue   │──▶ │  CaptchaAI   │
│  Client C    │──▶ │               │    │  API         │
└──────────────┘    └───────────────┘    └──────────────┘
                           │                    │
                           ▼                    ▼
                    ┌───────────────┐    ┌──────────────┐
                    │  Result Store │◀── │  Polling      │
                    │  (Redis/DB)   │    │  Workers      │
                    └───────────────┘    └──────────────┘

Componentes:

• Recepción de tareas: recibe las solicitudes de resolución que envían los scrapers de cada cliente.
• Cola: almacena las tareas en búfer y aplica un límite de concurrencia por cliente.
• Workers de resolución: envían cada tarea a CaptchaAI y sondean el resultado.
• Almacén de resultados: guarda los tokens resueltos para que el consumidor los recupere.

¿Cuántos threads necesitas por cliente?

Antes de escribir código conviene dimensionar la capacidad, porque marca qué plan contratas. CaptchaAI factura por thread concurrente, no por resolución: cada thread es un CAPTCHA en curso y, cuando termina, queda libre para el siguiente. Todos los planes incluyen resoluciones ilimitadas por thread durante el mes, sin topes diarios ni recargos por tipo de CAPTCHA.

Para una agencia esto convierte el costo en un gasto fijo y predecible en USD, algo que agradecen los equipos que facturan a sus clientes en monedas locales volátiles. La cuenta es directa: suma la concurrencia máxima de todos tus clientes y elige el plan cuyo número de threads la cubra.

• BASIC ($15/mes, 5 threads): uno o dos clientes con volumen bajo.
• STANDARD ($30/mes, 15 threads): varios clientes pequeños trabajando en paralelo.
• ADVANCE ($90/mes, 50 threads): una cartera de clientes con scraping continuo.

Si la concurrencia combinada supera tus threads, las tareas esperan en la cola en lugar de fallar, así que puedes empezar con un plan modesto y subir cuando el volumen crezca. Consulta los precios actualizados en captchaai.com/pricing antes de dimensionar.

Pipeline en Python

La clase central del solver

Esta clase encapsula tres cosas: el envío de la tarea a in.php, el sondeo del resultado contra res.php y un límite de concurrencia con max_concurrent. El estado vive en una deque (la cola) y un diccionario de tareas activas, ambos protegidos por un Lock para poder llamar al pipeline desde varios hilos.

import requests
import time
from dataclasses import dataclass
from typing import Optional
from collections import deque
from threading import Lock

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"

@dataclass
class SolveRequest:
    client_id: str
    method: str
    params: dict
    callback: Optional[callable] = None

@dataclass
class SolveResult:
    client_id: str
    task_id: str
    token: Optional[str] = None
    error: Optional[str] = None


class CaptchaPipeline:
    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.max_concurrent = max_concurrent
        self.queue = deque()
        self.active = {}
        self.lock = Lock()

    def enqueue(self, request: SolveRequest):
        with self.lock:
            self.queue.append(request)

    def submit_task(self, request: SolveRequest) -> Optional[str]:
        data = {
            "key": self.api_key,
            "method": request.method,
            "json": 1,
            **request.params
        }

        try:
            resp = requests.post(SUBMIT_URL, data=data, timeout=15)
            result = resp.json()

            if result.get("status") == 1:
                return result["request"]
            else:
                print(f"[{request.client_id}] Submit error: {result.get('error_text', result.get('request'))}")
                return None
        except requests.RequestException as e:
            print(f"[{request.client_id}] Network error: {e}")
            return None

    def poll_result(self, task_id: str, max_wait: int = 120) -> Optional[str]:
        elapsed = 0
        interval = 5
        while elapsed < max_wait:
            time.sleep(interval)
            elapsed += interval

            try:
                resp = requests.get(RESULT_URL, params={
                    "key": self.api_key,
                    "action": "get",
                    "id": task_id,
                    "json": 1
                }, timeout=10)
                result = resp.json()

                if result.get("status") == 1:
                    return result["request"]
                elif result.get("request") == "CAPCHA_NOT_READY":
                    continue
                else:
                    print(f"Poll error for {task_id}: {result.get('error_text', result.get('request'))}")
                    return None
            except requests.RequestException:
                continue

        return None

    def process_queue(self):
        while self.queue or self.active:
            # Fill active slots
            with self.lock:
                while self.queue and len(self.active) < self.max_concurrent:
                    request = self.queue.popleft()
                    task_id = self.submit_task(request)
                    if task_id:
                        self.active[task_id] = request

            # Poll active tasks
            completed = []
            for task_id, request in list(self.active.items()):
                token = self.poll_result(task_id, max_wait=10)
                if token:
                    result = SolveResult(
                        client_id=request.client_id,
                        task_id=task_id,
                        token=token
                    )
                    if request.callback:
                        request.callback(result)
                    completed.append(task_id)

            with self.lock:
                for task_id in completed:
                    del self.active[task_id]

Uso con varios clientes

Encolas una tarea por cliente indicando el method —aquí userrecaptcha para reCAPTCHA v2 y turnstile para Cloudflare Turnstile— y sus parámetros. El callback recibe el token en cuanto está listo, sin bloquear el resto de la cola:

pipeline = CaptchaPipeline(api_key="YOUR_API_KEY", max_concurrent=15)

# Client A — reCAPTCHA v2
pipeline.enqueue(SolveRequest(
    client_id="client_a",
    method="userrecaptcha",
    params={
        "googlekey": "6Le-SITEKEY-A",
        "pageurl": "https://client-a-target.com/form"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

# Client B — Turnstile
pipeline.enqueue(SolveRequest(
    client_id="client_b",
    method="turnstile",
    params={
        "sitekey": "0x4AAAA-SITEKEY-B",
        "pageurl": "https://client-b-target.com/login"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

pipeline.process_queue()

Pipeline en Node.js

La misma lógica en Node.js, con promesas en lugar de hilos: enqueue devuelve una promesa que se resuelve cuando el token está listo, y Promise.allSettled deja que varias tareas avancen en paralelo sin que el fallo de una tumbe a las demás.

const axios = require("axios");

const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

class CaptchaPipeline {
  constructor(apiKey, maxConcurrent = 10) {
    this.apiKey = apiKey;
    this.maxConcurrent = maxConcurrent;
    this.queue = [];
    this.activeCount = 0;
  }

  enqueue(clientId, method, params) {
    return new Promise((resolve, reject) => {
      this.queue.push({ clientId, method, params, resolve, reject });
      this._processNext();
    });
  }

  async _processNext() {
    if (this.activeCount >= this.maxConcurrent || this.queue.length === 0) return;

    this.activeCount++;
    const task = this.queue.shift();

    try {
      const token = await this._solve(task);
      task.resolve({ clientId: task.clientId, token });
    } catch (err) {
      task.reject(err);
    } finally {
      this.activeCount--;
      this._processNext();
    }
  }

  async _solve(task) {
    const submitResp = await axios.post(SUBMIT_URL, null, {
      params: {
        key: this.apiKey,
        method: task.method,
        json: 1,
        ...task.params,
      },
      timeout: 15000,
    });

    if (submitResp.data.status !== 1) {
      throw new Error(submitResp.data.error_text || submitResp.data.request);
    }

    const taskId = submitResp.data.request;
    return this._poll(taskId);
  }

  async _poll(taskId, maxWait = 120000) {
    const interval = 5000;
    let elapsed = 0;

    while (elapsed < maxWait) {
      await new Promise((r) => setTimeout(r, interval));
      elapsed += interval;

      try {
        const resp = await axios.get(RESULT_URL, {
          params: {
            key: this.apiKey,
            action: "get",
            id: taskId,
            json: 1,
          },
          timeout: 10000,
        });

        if (resp.data.status === 1) return resp.data.request;
        if (resp.data.request !== "CAPCHA_NOT_READY") {
          throw new Error(resp.data.error_text || resp.data.request);
        }
      } catch (err) {
        if (err.response) throw err;
      }
    }

    throw new Error(`Timeout waiting for task ${taskId}`);
  }
}

// Usage
(async () => {
  const pipeline = new CaptchaPipeline("YOUR_API_KEY", 15);

  const results = await Promise.allSettled([
    pipeline.enqueue("client_a", "userrecaptcha", {
      googlekey: "6Le-SITEKEY-A",
      pageurl: "https://client-a-target.com/form",
    }),
    pipeline.enqueue("client_b", "turnstile", {
      sitekey: "0x4AAAA-SITEKEY-B",
      pageurl: "https://client-b-target.com/login",
    }),
  ]);

  results.forEach((r) => {
    if (r.status === "fulfilled") {
      console.log(`[${r.value.clientId}] Token: ${r.value.token.slice(0, 40)}...`);
    } else {
      console.error(`Failed: ${r.reason.message}`);
    }
  });
})();

Configuración por cliente

Centraliza los ajustes de cada cliente —salida de red autorizada, método por defecto y límite de concurrencia— en un único diccionario. Así cambias el comportamiento de un cliente sin tocar la lógica del pipeline:

CLIENT_CONFIG = {
    "client_a": {
        "proxy": "host:port:user:pass",
        "proxytype": "HTTP",
        "max_concurrent": 5,
        "default_method": "userrecaptcha"
    },
    "client_b": {
        "proxy": None,
        "proxytype": None,
        "max_concurrent": 10,
        "default_method": "turnstile"
    }
}

def build_params(client_id, params):
    config = CLIENT_CONFIG.get(client_id, {})
    if config.get("proxy"):
        params["proxy"] = config["proxy"]
        params["proxytype"] = config["proxytype"]
    return params

Estrategia de manejo de errores

Un pipeline multicliente debe degradar con elegancia: el fallo de un cliente no puede tumbar la cola de los demás. Mapea cada código de error de la API a una respuesta concreta:

Resolución de problemas

Estos son los síntomas más habituales cuando el pipeline atiende a varios clientes a la vez y cómo corregirlos:

Preguntas frecuentes

Dudas frecuentes al montar un pipeline que da servicio a varias cuentas al mismo tiempo:

¿Qué plan de CaptchaAI conviene para una agencia con varios clientes?

Depende de la concurrencia total, no del número de clientes. Como CaptchaAI factura por thread con resoluciones ilimitadas, suma el max_concurrent de todas tus cuentas y contrata el plan cuyos threads la cubran; STANDARD ($30/mes, 15 threads) sirve para varios clientes pequeños y ADVANCE ($90/mes, 50 threads) para una cartera con scraping continuo.

¿Puedo combinar reCAPTCHA v2 y Turnstile en el mismo pipeline?

Sí. Cada tarea lleva su propio method, así que un mismo pipeline atiende a clientes con distintos tipos sin código adicional:

• userrecaptcha para reCAPTCHA v2 (y sus variantes invisible y Enterprise).
• turnstile para Cloudflare Turnstile.
• default_method por cliente en la configuración, para no repetirlo en cada tarea.

¿Cómo evito que se mezclen los tokens entre clientes?

El almacén de resultados es la parte que más se comparte por error. Dos reglas lo evitan:

• Indexa cada resultado por la combinación client_id + task_id, nunca solo por task_id.
• Usa una clave API distinta por cliente (o el parámetro soft_id) para separar de paso la facturación y el seguimiento.

¿Cómo conservo la cola si el proceso se reinicia de noche?

Persiste la cola fuera de la memoria del proceso para que un reinicio nocturno no borre el trabajo pendiente:

• Guarda las tareas en Redis o en una base de datos a medida que entran.
• Al arrancar, recarga las tareas sin terminar y reanuda el sondeo donde lo dejaste.

Empieza a construir tu pipeline con CaptchaAI

Crea tu cuenta en captchaai.com, elige el plan cuyos threads cubran tu concurrencia y resuelve el primer CAPTCHA de tus clientes desde un único sistema modular.

Guías relacionadas

• Guía de inicio rápido de CaptchaAI
• Cómo implementar lógica de reintentos en la API
• Procesamiento distribuido con colas de Redis