Construindo pipelines CAPTCHA de cliente com CaptchaAI

Como lidar com CAPTCHA quando sua agência atende vários clientes ao mesmo tempo? A resposta não é escrever uma rotina de resolução para cada projeto novo — é montar um único pipeline compartilhado, com fila, workers e isolamento de dados por client_id, que qualquer conta nova simplesmente entra. Este guia mostra a arquitetura completa com a API da CaptchaAI, do enfileiramento de tarefas ao tratamento de erros, com exemplos funcionais em Python e Node.js.

Arquitetura de um pipeline de CAPTCHA multicliente

Em vez de multiplicar scripts de resolução por projeto, centralize a lógica em quatro peças que qualquer cliente novo reaproveita:

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

As quatro peças do pipeline

Pipeline em Python: da fila ao token resolvido

A classe abaixo faz o trabalho pesado: enfileira tarefas por cliente, envia cada uma para a API da CaptchaAI e faz o polling do resultado sem travar as demais tarefas que esperam na fila.

Classe principal do solver

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]

A fila roda em memória neste exemplo; em produção, prefira uma fila persistente (Redis, banco de dados) para sobreviver a reinícios — voltamos a esse ponto na seção sobre observabilidade mais abaixo.

Uso multicliente

Cada cliente entra no mesmo pipeline apenas trocando method e params — não é preciso duplicar a classe para cada projeto novo:

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-staging.example.com/qa-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 em Node.js: fila assíncrona com Promises

A versão em Node.js troca o polling bloqueante por async/await e resolve cada tarefa como uma Promise independente — o que facilita plugar o pipeline em um servidor Express ou em uma fila de mensagens já existente.

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-staging.example.com/qa-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}`);
    }
  });
})();

Configuração por cliente: proxy, limites e método padrão

Guarde as preferências de cada cliente — proxy, tipo de proxy, simultaneidade máxima e método padrão de resolução — em um dicionário central. Isso evita hardcodar valores dentro da lógica de envio e permite adicionar um cliente novo em uma linha:

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

Se a maioria dos seus clientes está no Brasil, hospedar os workers em uma região próxima — por exemplo, sa-east-1 da AWS, em São Paulo — reduz o tempo de rede antes mesmo da chamada chegar à API da CaptchaAI, o que ajuda quando o max_concurrent de um cliente já está no limite.

Estratégia de tratamento de erros da API CaptchaAI

Trate cada código de erro de forma diferente — parar a fila inteira por um erro que afeta um cliente só desperdiça o throughput dos demais:

• ERROR_ZERO_BALANCE — pare a fila e alerte todos os clientes
• ERROR_NO_SLOT_AVAILABLE — reenfileire a tarefa com atraso
• ERROR_WRONG_CAPTCHA_ID — descarte a tarefa e registre o erro
• ERROR_CAPTCHA_UNSOLVABLE — tente novamente uma vez e depois marque como falha
• Timeout de rede — repita com backoff exponencial (máximo de 3 tentativas)

Solução de problemas comuns no pipeline

Observabilidade, isolamento de dados e LGPD

Um pipeline multicliente concentra tokens e metadados de vários clientes no mesmo Result Store — isso é conveniente para operar, mas exige cuidado se algum cliente estiver sujeito à LGPD (Lei Geral de Proteção de Dados), no Brasil, ou à RGPD, no caso de clientes em Portugal. Três práticas resolvem a maior parte do risco:

• Particione o armazenamento de resultados por client_id desde o primeiro dia — não misture logs de clientes diferentes na mesma tabela sem uma coluna de isolamento.
• Configure retenção curta para tokens já consumidos: um token resolvido perde o valor depois de usado, então não há motivo para guardá-lo indefinidamente.
• Registre erros com o task_id, nunca com dados do formulário do cliente final — isso mantém os logs úteis para debug sem expor dados pessoais desnecessários.

Nenhuma dessas práticas depende de um recurso exclusivo da CaptchaAI — são decisões de arquitetura do seu próprio pipeline, e valem tanto para um cliente só quanto para dez.

Perguntas frequentes

Um único plano CaptchaAI atende vários clientes ao mesmo tempo?

Sim. A CaptchaAI cobra por thread simultânea, não por cliente nem por solve — então um plano como ADVANCE (US$ 90/mês, 50 threads) pode atender vários clientes pequenos ao mesmo tempo, desde que a soma dos max_concurrent de cada um não ultrapasse o total de threads contratado.

hCaptcha funciona nos pipelines dos meus clientes?

Não — o CaptchaAI não resolve hCaptcha atualmente. Se algum cliente depender especificamente dele, deixe essa limitação clara antes de fechar o projeto e trate esse fluxo à parte.

Como evito misturar tokens entre clientes no mesmo pipeline?

Indexe cada resultado por client_id e task_id no Result Store, nunca só pelo task_id. É o erro mais comum em pipelines compartilhados — e está listado na tabela de solução de problemas acima.

Preciso de uma chave de API separada por cliente?

Não é obrigatório, mas simplifica o faturamento e o rastreamento de consumo por cliente. Se preferir manter uma única chave para todos, use o parâmetro soft_id da CaptchaAI para separar o consumo internamente.

Monte seu pipeline de CAPTCHA multicliente com a CaptchaAI

Crie sua conta na CaptchaAI, gere sua chave de API e coloque o exemplo em Python ou Node.js deste guia para rodar com o primeiro cliente ainda hoje. Acesse captchaai.com para começar.

Guias relacionados

• Resolução paralela de CAPTCHA
• Como implementar lógica de nova tentativa
• Fila distribuída com Redis para CAPTCHA
• Script de verificação de integridade (health check)