SQLite para cachear y registrar resoluciones de CAPTCHA

Si necesitas saber cuántos CAPTCHA resolvió tu script anoche, cuáles fallaron y cuánto tardó cada uno, no hace falta levantar un PostgreSQL: un archivo SQLite lo cubre. Viene con Python, no requiere proceso servidor y soporta del orden de mil escrituras por segundo con el modo WAL activado.

Aquí montamos tres piezas sobre la API de CaptchaAI: caché de tokens dentro de su TTL, registro de cada intento y consultas de análisis. El código espera tu API key en CAPTCHAAI_API_KEY.

Qué problema resuelve una base local

Un script sin persistencia es una caja negra: cuando el cliente pregunta por qué falló el proceso de madrugada, solo tienes logs de texto plano.

El segundo beneficio es económico. CaptchaAI factura por thread concurrente, no por resolución, y cada plan incluye resoluciones ilimitadas por thread: BASIC $15/mes con 5 threads, STANDARD $30/mes con 15 y ADVANCE $90/mes con 50. El límite no es el saldo, sino la concurrencia, y un caché de tokens libera threads que gastabas resolviendo dos veces el mismo formulario.

Un escenario típico de la región

Una agencia en Ciudad de México monitorea precios en marketplaces regionales para tres clientes con un plan STANDARD. Los tres procesos golpean el mismo dominio protegido por reCAPTCHA v2 en ventanas que se solapan: sin caché, cada uno pide su token y los 15 threads se saturan a media mañana. Con token_cache reutilizan el token vigente y el equipo pasa de rozar el techo a operar con margen. Igual de útil para quien vigila portales de trámites o cita previa, donde el tráfico llega a ráfagas.

Como en todo scraping, respeta los términos de servicio del sitio y la normativa de protección de datos aplicable (GDPR y LOPDGDD en España, LFPDPPP en México).

Cuándo SQLite es suficiente y cuándo no

Mientras escriban uno o varios procesos de la misma máquina, SQLite va sobrado. En cuanto aparecen varios servidores sobre el mismo archivo, se acabó.

Si caes en las tres últimas filas, el esquema de abajo sirve igual: las columnas se trasladan casi tal cual a MongoDB o a DynamoDB.

Diseño del esquema

captcha_solves es el historial y token_cache la memoria corta. Separarlas evita que la limpieza del caché toque tus métricas históricas.

CREATE TABLE IF NOT EXISTS captcha_solves (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    captcha_id TEXT,
    type TEXT NOT NULL,
    sitekey TEXT,
    pageurl TEXT,
    status TEXT NOT NULL DEFAULT 'submitted',
    solution TEXT,
    error TEXT,
    submitted_at TEXT NOT NULL DEFAULT (datetime('now')),
    solved_at TEXT,
    elapsed_ms INTEGER,
    polls INTEGER DEFAULT 0,
    project TEXT
);

CREATE INDEX IF NOT EXISTS idx_submitted_at ON captcha_solves(submitted_at);
CREATE INDEX IF NOT EXISTS idx_type_status ON captcha_solves(type, status);
CREATE INDEX IF NOT EXISTS idx_sitekey ON captcha_solves(sitekey);

-- Token cache for reuse within TTL
CREATE TABLE IF NOT EXISTS token_cache (
    sitekey TEXT NOT NULL,
    pageurl TEXT NOT NULL,
    token TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    expires_at TEXT NOT NULL,
    used INTEGER DEFAULT 0,
    PRIMARY KEY (sitekey, pageurl, token)
);

CREATE INDEX IF NOT EXISTS idx_cache_lookup
ON token_cache(sitekey, pageurl, used, expires_at);

elapsed_ms guarda el tiempo real de extremo a extremo, no el que reporta la API; polls delata un sondeo demasiado agresivo; project separa las estadísticas por cliente. La clave primaria compuesta de token_cache más la bandera used convierten esa tabla en una cola de un solo consumo.

Implementación en Python

Conexión y arranque

El único PRAGMA innegociable es journal_mode=WAL: sin él verás database is locked en cuanto lances dos workers.

import os
import time
import sqlite3
from datetime import datetime, timedelta, timezone
import requests

DB_PATH = os.environ.get("CAPTCHA_DB", "captcha_solves.db")
API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")  # Better concurrent read performance
    conn.execute("PRAGMA busy_timeout=5000")
    return conn


def init_db():
    conn = get_db()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS captcha_solves (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            captcha_id TEXT,
            type TEXT NOT NULL,
            sitekey TEXT,
            pageurl TEXT,
            status TEXT NOT NULL DEFAULT 'submitted',
            solution TEXT,
            error TEXT,
            submitted_at TEXT NOT NULL DEFAULT (datetime('now')),
            solved_at TEXT,
            elapsed_ms INTEGER,
            polls INTEGER DEFAULT 0,
            project TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_submitted_at ON captcha_solves(submitted_at);
        CREATE INDEX IF NOT EXISTS idx_type_status ON captcha_solves(type, status);

        CREATE TABLE IF NOT EXISTS token_cache (
            sitekey TEXT NOT NULL,
            pageurl TEXT NOT NULL,
            token TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            expires_at TEXT NOT NULL,
            used INTEGER DEFAULT 0,
            PRIMARY KEY (sitekey, pageurl, token)
        );
        CREATE INDEX IF NOT EXISTS idx_cache_lookup
        ON token_cache(sitekey, pageurl, used, expires_at);
    """)
    conn.close()

init_db()

Las sentencias son idempotentes, así que ejecutarlas en cada arranque no tiene coste.

Resolver y registrar en el mismo paso

El orden importa: primero el caché y solo si no hay token vigente se inserta el registro y se envía la tarea a in.php. Así el historial refleja resoluciones reales, no aciertos de caché.

def solve_recaptcha(sitekey, pageurl, project=None):
    conn = get_db()

    # Check cache first
    cached = get_cached_token(conn, sitekey, pageurl)
    if cached:
        conn.close()
        return cached

    # Insert tracking record
    now = datetime.now(timezone.utc).isoformat()
    cursor = conn.execute(
        "INSERT INTO captcha_solves (type, sitekey, pageurl, submitted_at, project) "
        "VALUES (?, ?, ?, ?, ?)",
        ("recaptcha_v2", sitekey, pageurl, now, project)
    )
    row_id = cursor.lastrowid
    conn.commit()

    # 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:
        conn.execute(
            "UPDATE captcha_solves SET status=?, error=? WHERE id=?",
            ("error", data.get("request"), row_id)
        )
        conn.commit()
        conn.close()
        return None

    captcha_id = data["request"]
    conn.execute(
        "UPDATE captcha_solves SET captcha_id=?, status=? WHERE id=?",
        (captcha_id, "polling", row_id)
    )
    conn.commit()

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

        if result.get("status") == 1:
            solved_at = datetime.now(timezone.utc).isoformat()
            submitted = datetime.fromisoformat(now)
            elapsed = int((datetime.now(timezone.utc) - submitted).total_seconds() * 1000)

            conn.execute(
                "UPDATE captcha_solves SET status=?, solution=?, solved_at=?, "
                "elapsed_ms=?, polls=? WHERE id=?",
                ("solved", result["request"], solved_at, elapsed, polls, row_id)
            )
            # Cache the token
            cache_token(conn, sitekey, pageurl, result["request"])
            conn.commit()
            conn.close()
            return result["request"]

        if result.get("request") != "CAPCHA_NOT_READY":
            conn.execute(
                "UPDATE captcha_solves SET status=?, error=?, polls=? WHERE id=?",
                ("error", result.get("request"), polls, row_id)
            )
            conn.commit()
            conn.close()
            return None

    conn.execute(
        "UPDATE captcha_solves SET status=?, polls=? WHERE id=?",
        ("timeout", polls, row_id)
    )
    conn.commit()
    conn.close()
    return None

El registro se actualiza en cada transición: submitted, polling con el captcha_id devuelto y finalmente solved, error o timeout. Esa columna te dice si falló el envío, el sondeo o el tiempo de espera. Para Cloudflare Turnstile o GeeTest v3 basta cambiar el method: la columna type ya separa las estadísticas.

Caché de tokens con TTL

Un token de reCAPTCHA vive unos 90–120 segundos; el TTL por defecto de 90 es conservador a propósito: mejor descartar uno válido que enviar uno caducado.

def cache_token(conn, sitekey, pageurl, token, ttl_seconds=90):
    expires_at = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat()
    conn.execute(
        "INSERT OR REPLACE INTO token_cache (sitekey, pageurl, token, expires_at) "
        "VALUES (?, ?, ?, ?)",
        (sitekey, pageurl, token, expires_at)
    )


def get_cached_token(conn, sitekey, pageurl):
    now = datetime.now(timezone.utc).isoformat()
    row = conn.execute(
        "SELECT token FROM token_cache "
        "WHERE sitekey=? AND pageurl=? AND used=0 AND expires_at > ? "
        "ORDER BY expires_at ASC LIMIT 1",
        (sitekey, pageurl, now)
    ).fetchone()

    if row:
        conn.execute(
            "UPDATE token_cache SET used=1 WHERE token=?",
            (row["token"],)
        )
        conn.commit()
        return row["token"]
    return None

La lectura filtra por used=0 y expires_at > now, y ordena por vencimiento ascendente. Marcar used=1 al entregarlo evita que dos hilos reciban el mismo token. Para un caché compartido entre máquinas, mira la gestión del TTL de tokens con Redis.

Consultas de analítica y limpieza

def get_stats(hours=24):
    conn = get_db()
    cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()

    total = conn.execute(
        "SELECT COUNT(*) FROM captcha_solves WHERE submitted_at >= ?", (cutoff,)
    ).fetchone()[0]

    solved = conn.execute(
        "SELECT COUNT(*) FROM captcha_solves WHERE submitted_at >= ? AND status='solved'",
        (cutoff,)
    ).fetchone()[0]

    avg_time = conn.execute(
        "SELECT AVG(elapsed_ms) FROM captcha_solves "
        "WHERE submitted_at >= ? AND status='solved'",
        (cutoff,)
    ).fetchone()[0]

    conn.close()
    return {
        "total": total,
        "solved": solved,
        "success_rate": (solved / total * 100) if total else 0,
        "avg_time_ms": round(avg_time) if avg_time else 0
    }


def cleanup_old_records(days=30):
    conn = get_db()
    cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    conn.execute("DELETE FROM captcha_solves WHERE submitted_at < ?", (cutoff,))
    conn.execute("DELETE FROM token_cache WHERE expires_at < ?",
                 (datetime.now(timezone.utc).isoformat(),))
    conn.execute("VACUUM")
    conn.commit()
    conn.close()

get_stats() es tu informe diario: con la tasa de éxito y el tiempo medio detectas en minutos una degradación que tardarías días en notar. cleanup_old_records() debe correr a diario desde cron; sin su VACUUM el archivo crece y nunca encoge.

La versión en Node.js

En JavaScript, better-sqlite3 ofrece una API síncrona que encaja mejor que los drivers con callbacks.

const Database = require("better-sqlite3");
const axios = require("axios");

const db = new Database(process.env.CAPTCHA_DB || "captcha_solves.db");
const API_KEY = process.env.CAPTCHAAI_API_KEY;

db.pragma("journal_mode = WAL");
db.exec(`
  CREATE TABLE IF NOT EXISTS captcha_solves (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    captcha_id TEXT, type TEXT NOT NULL, sitekey TEXT, pageurl TEXT,
    status TEXT DEFAULT 'submitted', solution TEXT, error TEXT,
    submitted_at TEXT DEFAULT (datetime('now')),
    solved_at TEXT, elapsed_ms INTEGER, polls INTEGER DEFAULT 0
  );
  CREATE INDEX IF NOT EXISTS idx_submitted ON captcha_solves(submitted_at);
`);

async function solveAndStore(sitekey, pageurl) {
  const submittedAt = new Date().toISOString();
  const insert = db.prepare(
    "INSERT INTO captcha_solves (type, sitekey, pageurl, submitted_at) VALUES (?, ?, ?, ?)"
  );
  const { lastInsertRowid } = insert.run("recaptcha_v2", sitekey, pageurl, submittedAt);

  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) {
    db.prepare("UPDATE captcha_solves SET status=?, error=? WHERE id=?")
      .run("error", submit.data.request, lastInsertRowid);
    return null;
  }

  const captchaId = submit.data.request;
  db.prepare("UPDATE captcha_solves SET captcha_id=?, status=? WHERE id=?")
    .run(captchaId, "polling", lastInsertRowid);

  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 elapsed = Date.now() - new Date(submittedAt).getTime();
      db.prepare(
        "UPDATE captcha_solves SET status=?, solution=?, solved_at=?, elapsed_ms=?, polls=? WHERE id=?"
      ).run("solved", poll.data.request, new Date().toISOString(), elapsed, polls, lastInsertRowid);
      return poll.data.request;
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      db.prepare("UPDATE captcha_solves SET status=?, error=?, polls=? WHERE id=?")
        .run("error", poll.data.request, polls, lastInsertRowid);
      return null;
    }
  }

  db.prepare("UPDATE captcha_solves SET status=?, polls=? WHERE id=?")
    .run("timeout", polls, lastInsertRowid);
  return null;
}

La estructura es la misma que en Python; portar el caché son unas quince líneas.

Problemas frecuentes

Preguntas frecuentes

¿Cuántas resoluciones por hora aguanta SQLite antes de quedarse corto?

Con WAL activado, del orden de 1.000 escrituras por segundo: muy por encima de cualquier pipeline razonable. El límite práctico es la concurrencia. Si superas las ~1.000 resoluciones por hora o escribes desde más de un servidor, migra a PostgreSQL o MongoDB.

¿El caché de tokens me ahorra dinero en el plan?

No directamente: CaptchaAI factura por thread concurrente, no por resolución. Lo que ahorra es concurrencia; cada token reutilizado deja un thread libre, así que sostienes más carga con el mismo plan BASIC ($15/mes, 5 threads) o STANDARD ($30/mes, 15 threads).

¿Puedo compartir el archivo .db entre varios contenedores Docker?

No es recomendable: SQLite depende del bloqueo de archivos del sistema operativo, y ese bloqueo no es fiable sobre volúmenes de red ni montajes compartidos. Deja el archivo dentro de un único contenedor que actúe como servicio de registro.

¿Cómo paso de SQLite a una base de producción sin rehacer el código?

Exporta con sqlite3 captcha_solves.db ".dump" e importa en PostgreSQL o MongoDB; el esquema se traslada casi literalmente. Aísla el acceso a datos detrás de una capa fina y cambiar de motor tocará un solo archivo. Para evitar resoluciones duplicadas entre procesos, revisa el patrón de deduplicación con bloqueo en base de datos.

Artículos relacionados

• Seguimiento de resoluciones sin servidor con DynamoDB

Próximos pasos

Crea las dos tablas, envuelve tu llamada a in.php con la función de seguimiento y deja correr el proceso una semana: la primera consulta a get_stats() te dirá más que cualquier estimación. Obtén tu API key de CaptchaAI y registra tu primera resolución hoy.

Guías relacionadas:

• Historial y analítica de CAPTCHA con MongoDB
• Gestión del TTL de tokens con Redis
• Deduplicar solicitudes de resolución