Transmissão de resultados em lote: processamento de soluções CAPTCHA à medida que chegam

Se o seu código só começa a trabalhar depois que a última tarefa do lote volta, você está pagando por um tempo ocioso que dá para eliminar em uma tarde. A resposta curta é tratar o lote como um fluxo: cada token é consumido no instante em que fica pronto, não quando a fila inteira termina.

Abaixo estão os dois padrões que sustentam esse fluxo sobre a API da CaptchaAI — um gerador assíncrono em Python e um EventEmitter em Node.js —, além do critério para decidir entre transmitir e coletar tudo.

Por que a ordem de chegada nunca é a ordem de envio

O tempo de resolução varia conforme o tipo de CAPTCHA, a carga do momento e o desafio sorteado. É normal que a décima tarefa enviada volte antes da terceira, e essa dispersão tem três consequências práticas:

• O primeiro resultado aproveitável aparece bem antes da média do lote.
• Guardar tudo até o final faz a memória crescer junto com o tamanho do lote.
• Como a ordem se perde, cada resultado precisa carregar o index da tarefa original.

Streaming, lote fechado e microlote lado a lado

Quando o streaming compensa e quando coletar tudo é melhor

O critério é o que acontece com o token depois que ele chega: se cada resultado tem uso imediato, transmita; se o valor só existe no conjunto, colete.

Python: gerador assíncrono que entrega cada resultado

Com asyncio e aiohttp o padrão cabe em uma função: asyncio.wait(..., return_when=FIRST_COMPLETED) devolve só os futures já concluídos, e o gerador faz yield de cada um na hora:

import asyncio
import aiohttp
import time

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


async def submit_task(session, task_data):
    """Submit a single CAPTCHA task."""
    params = {
        "key": API_KEY,
        "method": task_data.get("method", "userrecaptcha"),
        "json": 1,
    }
    if params["method"] == "userrecaptcha":
        params["googlekey"] = task_data["sitekey"]
        params["pageurl"] = task_data["pageurl"]
    elif params["method"] == "turnstile":
        params["sitekey"] = task_data["sitekey"]
        params["pageurl"] = task_data["pageurl"]

    async with session.post(SUBMIT_URL, data=params) as resp:
        result = await resp.json(content_type=None)
        if result.get("status") != 1:
            return None, result.get("request", "unknown")
        return result["request"], None


async def poll_task(session, task_id, timeout=300):
    """Poll until solved or timeout."""
    start = time.monotonic()
    while time.monotonic() - start < timeout:
        await asyncio.sleep(5)
        params = {"key": API_KEY, "action": "get", "id": task_id, "json": 1}
        async with session.get(RESULT_URL, params=params) as resp:
            result = await resp.json(content_type=None)

        if result.get("request") == "CAPCHA_NOT_READY":
            continue
        if result.get("status") == 1:
            return result["request"], None
        return None, result.get("request", "unknown")

    return None, "TIMEOUT"


async def solve_one(session, index, task_data, semaphore):
    """Solve a single task within concurrency limits."""
    async with semaphore:
        start = time.monotonic()
        task_id, error = await submit_task(session, task_data)
        if error:
            return {"index": index, "status": "failed", "error": error, "time": 0}

        token, error = await poll_task(session, task_id)
        elapsed = time.monotonic() - start

        if token:
            return {"index": index, "status": "solved", "token": token, "time": round(elapsed, 1)}
        return {"index": index, "status": "failed", "error": error, "time": round(elapsed, 1)}


async def stream_results(tasks, max_concurrent=20):
    """
    Async generator that yields each result as it completes.
    Results arrive in completion order, not submission order.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        pending = set()
        for i, task in enumerate(tasks):
            coro = solve_one(session, i, task, semaphore)
            pending.add(asyncio.ensure_future(coro))

        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            for future in done:
                yield future.result()


async def main():
    tasks = [
        {"sitekey": "SITE_KEY", "pageurl": f"https://example.com/page{i}"}
        for i in range(50)
    ]

    solved = 0
    failed = 0

    async for result in stream_results(tasks, max_concurrent=15):
        # Process each result immediately
        if result["status"] == "solved":
            solved += 1
            print(f"  [{solved + failed}/{len(tasks)}] Task {result['index']} SOLVED in {result['time']}s")

            # Use token immediately — don't wait for batch
            # await submit_form(result["token"])
            # await save_to_database(result)
        else:
            failed += 1
            print(f"  [{solved + failed}/{len(tasks)}] Task {result['index']} FAILED: {result['error']}")

    print(f"\nDone: {solved} solved, {failed} failed")


asyncio.run(main())

A única dependência externa é o cliente HTTP assíncrono:

pip install aiohttp

Alinhe a simultaneidade às threads do seu plano

O max_concurrent=15 do exemplo espelha as threads contratadas. A CaptchaAI cobra por thread simultânea, com resoluções ilimitadas no mês: o STANDARD (US$ 30/mês, 15 threads) cobre esse ajuste, e o BASIC (US$ 15/mês, 5 threads) pede max_concurrent=5. Abrir mais tarefas do que o plano permite não acelera o lote — o excedente só espera na fila do seu processo.

Node.js: streaming de resultados com EventEmitter

No Node.js o mesmo comportamento sai mais natural com eventos. A classe abaixo mantém uma janela fixa de tarefas ativas, emite result a cada conclusão e done no fim do lote:

const { EventEmitter } = require("events");

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

class CaptchaStream extends EventEmitter {
  constructor(maxConcurrent = 15) {
    super();
    this.maxConcurrent = maxConcurrent;
    this.active = 0;
    this.queue = [];
    this.total = 0;
    this.completed = 0;
  }

  async submitAndPoll(index, taskData) {
    const params = new URLSearchParams({
      key: API_KEY,
      method: taskData.method || "userrecaptcha",
      googlekey: taskData.sitekey,
      pageurl: taskData.pageurl,
      json: "1",
    });

    const start = Date.now();
    const submitResp = await (await fetch(SUBMIT_URL, { method: "POST", body: params })).json();

    if (submitResp.status !== 1) {
      return { index, status: "failed", error: submitResp.request, time: 0 };
    }

    const taskId = submitResp.request;
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const url = `${RESULT_URL}?key=${API_KEY}&action=get&id=${taskId}&json=1`;
      const poll = await (await fetch(url)).json();

      if (poll.request === "CAPCHA_NOT_READY") continue;
      const elapsed = ((Date.now() - start) / 1000).toFixed(1);
      if (poll.status === 1) return { index, status: "solved", token: poll.request, time: elapsed };
      return { index, status: "failed", error: poll.request, time: elapsed };
    }
    return { index, status: "failed", error: "TIMEOUT", time: ((Date.now() - start) / 1000).toFixed(1) };
  }

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

    const { index, taskData } = this.queue.shift();
    this.active++;

    try {
      const result = await this.submitAndPoll(index, taskData);
      this.emit("result", result);
    } catch (err) {
      this.emit("result", { index, status: "failed", error: err.message });
    } finally {
      this.active--;
      this.completed++;

      if (this.completed === this.total) {
        this.emit("done");
      } else {
        this.processNext();
      }
    }
  }

  start(tasks) {
    this.total = tasks.length;
    this.queue = tasks.map((taskData, index) => ({ index, taskData }));

    // Launch initial batch
    const initial = Math.min(this.maxConcurrent, tasks.length);
    for (let i = 0; i < initial; i++) {
      this.processNext();
    }
    return this;
  }
}

// Usage
const tasks = Array.from({ length: 50 }, (_, i) => ({
  sitekey: "SITE_KEY",
  pageurl: `https://example.com/page${i}`,
}));

const stream = new CaptchaStream(15);
let solved = 0, failed = 0;

stream.on("result", (result) => {
  if (result.status === "solved") {
    solved++;
    console.log(`[${solved + failed}/${tasks.length}] Task ${result.index} SOLVED (${result.time}s)`);
    // Use token immediately
    // submitForm(result.token);
  } else {
    failed++;
    console.log(`[${solved + failed}/${tasks.length}] Task ${result.index} FAILED: ${result.error}`);
  }
});

stream.on("done", () => {
  console.log(`\nComplete: ${solved} solved, ${failed} failed`);
});

stream.start(tasks);

Cenário: 500 desafios na janela noturna de QA

Um time em São Paulo roda a regressão do formulário de cadastro em staging.example.com toda madrugada, com os workers em sa-east-1 para encurtar o RTT até a API. São 500 desafios por execução, em uma janela de manutenção curta. No modelo "espere tudo", o relatório só começa depois que a tarefa mais lenta retorna; com streaming, cada envio parte assim que o token chega e uma sequência de falhas do mesmo tipo aparece nos primeiros minutos.

Um lembrete de escopo: os exemplos usam dados fictícios em ambiente próprio. Se o pipeline passar a tocar dados de pessoas reais, considere as obrigações da LGPD sobre registro e retenção — para depurar streaming bastam índice, status e tempo, sem armazenar o token.

Erros comuns no streaming e como corrigir

Checklist antes de levar para produção

• Limite de simultaneidade igual ou inferior às threads do plano.
• Timeout por tarefa, para que nenhuma pendência segure o gerador.
• Handler idempotente: numa retomada, o mesmo resultado pode voltar.
• Checkpoint em modo append a cada resultado, com o index junto.
• Tempo de resolução por tarefa registrado — é o que mostra se vale subir de plano.

Perguntas frequentes

Qual limite de simultaneidade devo usar?

O número de threads do seu plano, nunca mais que isso. Como a cobrança é por thread simultânea, o teto do plano define o throughput real do lote. Se a fila interna vive cheia, o gargalo é o plano, não o código.

E se uma tarefa nunca resolver?

Ela precisa falhar sozinha. O poll_task do exemplo tem timeout=300, devolve TIMEOUT e libera a vaga do semáforo — sem isso, um único future pendente segura o gerador. Registre esses casos com o index para reenviar depois.

Preciso de um navegador para rodar esse pipeline?

Não. Os dois exemplos conversam diretamente com in.php e res.php por HTTP, sem Selenium nem Playwright. O navegador só entra se o passo seguinte for preencher o campo do token em uma página real.

Como retomo um lote interrompido no meio?

Grave cada resultado em um arquivo de checkpoint assim que ele chegar. Na retomada, descarte os índices já concluídos e monte a lista apenas com o que restou — o gerador não muda.

Dá para misturar tipos de CAPTCHA no mesmo lote?

Sim. O method é definido por tarefa: reCAPTCHA v2 e v3, Cloudflare Turnstile, GeeTest v3 e desafios de imagem/OCR convivem no mesmo lote. Como o tempo de resolução muda bastante entre tipos, é nesse cenário misto que o streaming mais aparece.

Comece pelo primeiro lote

Crie sua conta em captchaai.com, rode um lote pequeno em streaming e compare o tempo até o primeiro token com o do seu fluxo atual.

Guias relacionados

• Processamento de CAPTCHA em streaming com Kafka
• Fila com prioridade para lotes de CAPTCHA
• Lotes de CAPTCHA a partir de um CSV em Python
• Resolução de CAPTCHA em paralelo