Histórico e análise de resolução de CAPTCHA com MongoDB

Quantas resoluções a sua aplicação fechou na última hora? Qual foi a taxa de sucesso do reCAPTCHA v2 nesta semana, comparada à do Cloudflare Turnstile? Se responder isso exige abrir logs espalhados por vários serviços, seu pipeline de resolução de CAPTCHA está sem observabilidade — e é esse buraco que o MongoDB preenche.

Cada tipo de CAPTCHA carrega campos diferentes: reCAPTCHA usa googlekey, hCaptcha usa sitekey, CAPTCHAs de imagem usam body. Um banco relacional obrigaria você a migrar o esquema a cada tipo novo; o MongoDB aceita o documento como ele é. Neste guia você monta o esquema, cria os índices certos, implementa a função de resolução com histórico em Python e Node.js, e roda consultas de agregação que respondem esse tipo de pergunta em segundos.

Por que o MongoDB funciona bem para o histórico de CAPTCHA

Além do esquema flexível, o documento espelha o formato que a API da CaptchaAI já devolve: um objeto por tarefa, com status, request (token ou erro) e os metadados que você quiser anexar — sem a camada de tradução que existiria em um banco relacional.

Times com workers em várias regiões — por exemplo, uma instância em sa-east-1 (São Paulo) para reduzir a latência até o site de origem — se beneficiam de indexar por metadata.target_domain e metadata.worker_id: dá para comparar o tempo de resolução por região sem duplicar a modelagem. Como os registros guardam pageurl e o domínio-alvo, vale revisar quanto tempo esses metadados ficam armazenados — a seção de retenção mais abaixo trata disso à luz da LGPD.

Modelo de documento para registros de CAPTCHA

Cada tentativa de resolução vira um documento. Os campos abaixo cobrem o ciclo de vida completo — do envio até a resposta final — sem impor um esquema rígido:

{
  "_id": "ObjectId",
  "captcha_id": "12345678",
  "type": "recaptcha_v2",
  "method": "userrecaptcha",
  "sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
  "pageurl": "https://example.com/form",
  "status": "solved",
  "solution": "03AGdBq26...",
  "error": null,
  "submitted_at": "2026-04-04T10:15:30.000Z",
  "solved_at": "2026-04-04T10:15:45.000Z",
  "elapsed_ms": 15000,
  "polls": 3,
  "proxy_used": true,
  "cost": 0.00299,
  "metadata": {
    "project": "price-monitor",
    "worker_id": "worker-3",
    "target_domain": "example.com"
  }
}

elapsed_ms e polls alimentam as consultas de analytics mais adiante; proxy_used e cost ajudam a reconciliar o consumo com o seu plano da CaptchaAI; e o objeto metadata é livre — adicione o que fizer sentido para o seu pipeline (projeto, worker, domínio-alvo, tenant, o que for).

Implementação em Python

A seguir: conexão, índices, a função que resolve e grava o histórico, e quatro consultas de analytics prontas para usar.

Configuração e conexão com o MongoDB

Aponte MONGO_URI para sua instância — local, Docker ou Atlas — e mantenha a chave de API da CaptchaAI em uma variável de ambiente, nunca no código:

import os
import time
from datetime import datetime, timezone
from pymongo import MongoClient, ASCENDING, DESCENDING
import requests

MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017")
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

client = MongoClient(MONGO_URI)
db = client["captcha_tracking"]
solves = db["solves"]

Criação dos índices

Quatro índices cobrem os acessos mais comuns: data de envio, tipo mais status, e busca por projeto ou domínio-alvo. O último usa TTL para expirar documentos automaticamente após 90 dias — ajuste conforme sua política de retenção:

def setup_indexes():
    solves.create_index([("submitted_at", DESCENDING)])
    solves.create_index([("type", ASCENDING), ("status", ASCENDING)])
    solves.create_index([("metadata.project", ASCENDING)])
    solves.create_index([("metadata.target_domain", ASCENDING)])
    solves.create_index(
        [("submitted_at", ASCENDING)],
        expireAfterSeconds=90 * 24 * 3600,  # Auto-delete after 90 days
        name="ttl_cleanup"
    )

setup_indexes()

Função para resolver e registrar o histórico

solve_and_store grava o documento assim que a tarefa é enviada, com status: submitted, e só depois consulta o resultado — assim, mesmo se o processo cair no meio do polling, sobra um registro do que foi enviado e quando. A cada consulta sem resultado pronto (CAPCHA_NOT_READY), o contador polls sobe; ao final, o status vira solved, error ou timeout:

def solve_and_store(sitekey, pageurl, captcha_type="recaptcha_v2", metadata=None):
    record = {
        "type": captcha_type,
        "method": "userrecaptcha",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "status": "submitted",
        "submitted_at": datetime.now(timezone.utc),
        "metadata": metadata or {}
    }

    result = solves.insert_one(record)
    doc_id = result.inserted_id

    # Submit to CaptchaAI
    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:
        solves.update_one(
            {"_id": doc_id},
            {"$set": {"status": "error", "error": data.get("request")}}
        )
        return None

    captcha_id = data["request"]
    solves.update_one(
        {"_id": doc_id},
        {"$set": {"captcha_id": captcha_id, "status": "polling"}}
    )

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

        if poll_resp.get("status") == 1:
            solved_at = datetime.now(timezone.utc)
            elapsed_ms = int(
                (solved_at - record["submitted_at"]).total_seconds() * 1000
            )
            solves.update_one({"_id": doc_id}, {"$set": {
                "status": "solved",
                "solution": poll_resp["request"],
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls
            }})
            return poll_resp["request"]

        if poll_resp.get("request") != "CAPCHA_NOT_READY":
            solves.update_one({"_id": doc_id}, {"$set": {
                "status": "error",
                "error": poll_resp.get("request"),
                "polls": polls
            }})
            return None

    solves.update_one({"_id": doc_id}, {"$set": {
        "status": "timeout", "polls": polls
    }})
    return None

Consultas de analytics

Com o histórico gravado, quatro perguntas ficam triviais de responder:

• Qual a taxa de sucesso no período? — get_success_rate agrupa por status e calcula o percentual de solved.
• Qual tipo demora mais para resolver? — get_avg_solve_time_by_type agrupa por type com média, mínimo e máximo de elapsed_ms.
• Como o volume varia ao longo do dia? — get_hourly_solve_volume agrupa por hora, pronto para um gráfico.
• Quais erros são mais frequentes? — get_error_breakdown agrupa por error nas últimas N horas.

def get_success_rate(hours=24):
    """Success rate for the last N hours."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}}},
        {"$group": {
            "_id": "$status",
            "count": {"$sum": 1}
        }}
    ]
    results = {r["_id"]: r["count"] for r in solves.aggregate(pipeline)}
    total = sum(results.values())
    solved = results.get("solved", 0)
    return (solved / total * 100) if total else 0


def get_avg_solve_time_by_type():
    """Average solve time grouped by CAPTCHA type."""
    pipeline = [
        {"$match": {"status": "solved"}},
        {"$group": {
            "_id": "$type",
            "avg_time_ms": {"$avg": "$elapsed_ms"},
            "min_time_ms": {"$min": "$elapsed_ms"},
            "max_time_ms": {"$max": "$elapsed_ms"},
            "count": {"$sum": 1}
        }},
        {"$sort": {"count": -1}}
    ]
    return list(solves.aggregate(pipeline))


def get_hourly_solve_volume(days=7):
    """Hourly solve volume for charting."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}}},
        {"$group": {
            "_id": {
                "date": {"$dateToString": {"format": "%Y-%m-%d", "date": "$submitted_at"}},
                "hour": {"$hour": "$submitted_at"}
            },
            "total": {"$sum": 1},
            "solved": {"$sum": {"$cond": [{"$eq": ["$status", "solved"]}, 1, 0]}}
        }},
        {"$sort": {"_id.date": 1, "_id.hour": 1}}
    ]
    return list(solves.aggregate(pipeline))


def get_error_breakdown(hours=24):
    """Error frequency by error code."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}, "status": "error"}},
        {"$group": {"_id": "$error", "count": {"$sum": 1}}},
        {"$sort": {"count": -1}}
    ]
    return list(solves.aggregate(pipeline))

Implementação em Node.js

A mesma lógica — conexão, índices, resolução com histórico e taxa de sucesso — em Node.js, com o driver oficial do MongoDB e axios:

const { MongoClient } = require("mongodb");
const axios = require("axios");

const MONGO_URI = process.env.MONGO_URI || "mongodb://localhost:27017";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

let db, solves;

async function connect() {
  const client = await MongoClient.connect(MONGO_URI);
  db = client.db("captcha_tracking");
  solves = db.collection("solves");

  await solves.createIndex({ submitted_at: -1 });
  await solves.createIndex({ type: 1, status: 1 });
  await solves.createIndex({ "metadata.project": 1 });
  await solves.createIndex(
    { submitted_at: 1 },
    { expireAfterSeconds: 90 * 24 * 3600 }
  );
}

async function solveAndStore(sitekey, pageurl, type = "recaptcha_v2", metadata = {}) {
  const submittedAt = new Date();
  const { insertedId } = await solves.insertOne({
    type, method: "userrecaptcha", sitekey, pageurl,
    status: "submitted", submitted_at: submittedAt, metadata,
  });

  const submit = await axios.post("https://ocr.captchaai.com/in.php", null, {
    params: { key: API_KEY, method: "userrecaptcha", googlekey: sitekey, pageurl, json: 1 },
  });

  if (submit.data.status !== 1) {
    await solves.updateOne({ _id: insertedId }, { $set: { status: "error", error: submit.data.request } });
    return null;
  }

  const captchaId = submit.data.request;
  await solves.updateOne({ _id: insertedId }, { $set: { captcha_id: captchaId, status: "polling" } });

  let polls = 0;
  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    polls++;
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
    });

    if (poll.data.status === 1) {
      const solvedAt = new Date();
      await solves.updateOne({ _id: insertedId }, { $set: {
        status: "solved", solution: poll.data.request,
        solved_at: solvedAt, elapsed_ms: solvedAt - submittedAt, polls,
      }});
      return poll.data.request;
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      await solves.updateOne({ _id: insertedId }, { $set: { status: "error", error: poll.data.request, polls } });
      return null;
    }
  }

  await solves.updateOne({ _id: insertedId }, { $set: { status: "timeout", polls } });
  return null;
}

async function getSuccessRate(hours = 24) {
  const cutoff = new Date(Date.now() - hours * 3600 * 1000);
  const pipeline = [
    { $match: { submitted_at: { $gte: cutoff } } },
    { $group: { _id: "$status", count: { $sum: 1 } } },
  ];
  const results = await solves.aggregate(pipeline).toArray();
  const total = results.reduce((s, r) => s + r.count, 0);
  const solved = results.find((r) => r._id === "solved")?.count || 0;
  return total ? ((solved / total) * 100).toFixed(1) : 0;
}

Política de retenção de dados

Quanto tempo manter um histórico de CAPTCHA depende do uso: alguns dias bastam para depuração; 90 dias é um ponto de partida razoável para métricas de produto e reconciliação de custo com o seu plano da CaptchaAI. Como os documentos guardam pageurl e target_domain — potencialmente dados pessoais, dependendo do contexto — vale revisar a retenção à luz da LGPD (ou do RGPD, em Portugal). O índice TTL abaixo torna esse descarte automático, sem job de limpeza manual:

Problemas comuns e como resolver

Perguntas frequentes

O índice TTL apaga os documentos exatamente na hora em que expiram?

Não. O monitor de TTL roda em ciclos de cerca de 60 segundos, então a exclusão acontece pouco depois do prazo em expireAfterSeconds — margem irrelevante para os fins deste guia.

Preciso guardar o token completo da resposta do CAPTCHA?

Para depuração, guarde o token por 24–48 horas e deixe o TTL limpar depois. Para analytics de longo prazo, guarde só os metadados — tipo, horário, status e erro —, já que o token perde valor assim que expira.

Qual a diferença entre manter esse histórico no MongoDB e usar um cache de token no Redis?

São complementares. O Redis (veja o gerenciamento de TTL de token no Redis) serve para consulta rápida de um token ainda válido; o MongoDB guarda o histórico completo para análise e comparação entre tipos de CAPTCHA ao longo do tempo.

Posso usar o MongoDB Atlas em vez de uma instância própria?

Sim. O Atlas suporta índices TTL e pipelines de agregação normalmente — só apontar MONGO_URI para a string de conexão do painel.

Vários workers podem gravar ao mesmo tempo sem conflito?

Sim. Cada resolução gera seu próprio insert_one (insertOne em Node.js), sem necessidade de lock: mesmo com dezenas de workers em paralelo — inclusive em regiões diferentes, como sa-east-1 — cada documento é independente. O gargalo real costuma ser o maxPoolSize da conexão, não o MongoDB (veja a tabela acima).

Próximos passos

Registre cada resolução de CAPTCHA e identifique quedas na taxa de sucesso antes que elas afetem seu pipeline — obtenha sua chave de API da CaptchaAI.

Guias relacionados:

• Cache local de CAPTCHA com SQLite
• Gerenciamento de TTL de token no Redis
• Tendências de desempenho em séries temporais