Historial y analítica de resoluciones de CAPTCHA en MongoDB

¿Dónde conviene guardar el historial de cada CAPTCHA que resuelves y cómo lo conviertes en métricas que sirvan? MongoDB es una de las mejores opciones: su esquema flexible absorbe los campos que cambian según el tipo de CAPTCHA y su framework de agregación te da tasa de éxito, tiempos y costo sin montar un data warehouse aparte. Aquí guardas cada intento con sus metadatos y consultas patrones por fecha, tipo y tasa de error.

Por qué MongoDB encaja con los registros de resolución

Cada registro de resolución tiene campos distintos según el tipo: reCAPTCHA envía googlekey, Cloudflare Turnstile trabaja con su sitekey y los CAPTCHA de imagen mandan un body. En una base relacional acabarías con columnas nulas por todas partes o con migraciones constantes; los documentos sin esquema de MongoDB los guardan tal cual, sin migraciones de por medio.

Piensa en un equipo de datos que corre workers de monitoreo de precios sobre marketplaces regionales (tipo MercadoLibre o Amazon.es), respetando los términos de servicio de cada sitio. Con miles de resoluciones diarias por proyecto, saber qué falla y cuánto cuesta cada flujo exige un historial consultable en un solo lugar.

Diseño del documento

{
  "_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"
  }
}

Cada documento cubre el ciclo de vida completo de un intento. Vale la pena fijarte en unos pocos campos, porque son los que después alimentan las métricas:

• status: en qué punto quedó el intento (submitted, polling, solved, error, timeout). Es la base de la tasa de éxito.
• elapsed_ms y polls: cuánto tardó la resolución y cuántas veces sondeaste el resultado antes de tenerlo.
• cost: el costo de cada resolución, para sumar el gasto por proyecto o por día con un $group.
• metadata: un subdocumento libre donde etiquetas cada intento con proyecto, worker y dominio objetivo, sin tocar el esquema.

Implementación en Python

Configuración y conexión

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"]

Crear los índices

Sin índices, cada agregación recorre la colección entera. El índice TTL, además, borra solo los registros viejos.

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()

Resolver y guardar

El patrón: insertas el registro como submitted, envías la tarea a CaptchaAI, sondeas el resultado y actualizas el documento en cada cambio de estado. Si el proceso muere a mitad, queda constancia de dónde se quedó.

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

Cada llamada deja elapsed_ms y polls: la materia prima para medir tiempos y detectar cuándo un tipo empieza a tardar de más.

Consultas de analítica

Estas cuatro agregaciones cubren lo típico de un panel: tasa de éxito, tiempos por tipo, volumen por hora y errores.

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))

Implementación en JavaScript

Sobre Node.js, el modelo se traslada casi línea por línea con el driver oficial de MongoDB y 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íticas de retención

El índice TTL fija una ventana por caso de uso y automatiza la limpieza.

Buenas prácticas para consultar el historial

Unas cuantas costumbres evitan que la colección se degrade a medida que crece:

• Filtra siempre por submitted_at antes de agrupar: acotas el rango y aprovechas el índice en lugar de recorrer todo.
• Proyecta solo los campos que necesitas en cada agregación; arrastrar el solution completo infla la memoria del pipeline sin ninguna ganancia.
• Revisa db.solves.getIndexes() después de cada despliegue para confirmar que el índice TTL sigue activo.
• Separa por metadata.project cuando varios flujos comparten la misma colección: así comparas tasa de éxito y costo entre proyectos sin mezclarlos.

Resolución de problemas comunes

Preguntas frecuentes

¿Cómo mido la tasa de éxito y el costo con estos registros?

La función get_success_rate() agrupa por status y te da el porcentaje de resueltos. Para el costo, suma el campo cost por proyecto o por día con un $group. Con el gasto centralizado, comparar el costo mensual en USD frente al pago por resolución deja de ser una estimación.

¿Qué índices necesito para que las agregaciones no se ralenticen?

Los de setup_indexes(): uno sobre submitted_at, uno compuesto sobre type + status y otros sobre los campos de metadata que filtres. Sin ellos, cada agregación recorre la colección completa y el retraso se dispara al superar unos miles de documentos.

¿Puedo usar MongoDB Atlas en la nube?

Sí. Atlas admite índices TTL y pipelines de agregación igual que una instancia local. Copia la cadena de conexión de tu panel de Atlas en MONGO_URI y el resto del código no cambia.

¿Conviene guardar el token completo de la solución?

Depende: para depurar, guárdalo 24–48 horas y deja que el TTL lo borre. Para analítica a largo plazo guarda solo metadatos (tipo, hora, estado, error): el token deja de servir en cuanto vence.

¿Puedo separar las métricas por proyecto o por worker?

Sí, y para eso está el subdocumento metadata. Etiqueta cada intento con project y worker_id, indexa esos campos y añade un $match por metadata.project al inicio de cada agregación: obtienes tasa de éxito, tiempos y costo por flujo sin duplicar colecciones.

Siguientes pasos

Registra cada resolución y detecta problemas antes de que afecten a tu pipeline. Obtén tu clave API de CaptchaAI y empieza a llenar hoy tu colección de métricas.

Guías relacionadas:

• SQLite para caché local de resoluciones
• Gestión del TTL de tokens con Redis
• Tendencias de rendimiento en series temporales