SQLite para cache e rastreamento de resoluções de CAPTCHA

O rastreamento de resoluções de CAPTCHA costuma começar com print() e parar num log que ninguém consulta. Um arquivo SQLite de poucos megabytes muda isso: tempo, tentativas e falhas de cada resolução viram consultas SQL. Ele vem com o Python, não exige servidor e sustenta uma automação de máquina única abaixo de mil resoluções por hora.

O banco local cobre três funções de uma vez: histórico, cache de token dentro da validade e desduplicação de requisições simultâneas. Abaixo estão o esquema, o código, as consultas e a limpeza, em Python e em Node.js.

Quando o SQLite é suficiente — e quando não é

A fronteira não é o tamanho do arquivo, é a concorrência de escrita: o SQLite aceita muitas leituras simultâneas, mas serializa as gravações. Em um único host isso não pesa; com dois servidores gravando no mesmo arquivo em rede, o modelo quebra.

Nas três primeiras linhas, comece pelo SQLite. Migrar depois custa um .dump.

Esquema: histórico em uma tabela, cache em outra

Separar as duas responsabilidades evita o erro mais comum aqui: usar a tabela de auditoria como cache e devolver um token expirado.

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

Três colunas fazem o trabalho pesado: elapsed_ms mede o tempo entre envio e resposta, polls mostra quantas consultas de resultado foram necessárias e project separa ambientes sem exigir um banco por pipeline. Na token_cache, o índice idx_cache_lookup cobre a consulta mais frequente — token não usado e ainda válido para aquela sitekey.

Implementação em Python

Passo 1: abra a conexão e crie as tabelas

Dois PRAGMAs fazem toda a diferença no dia a dia.

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

journal_mode=WAL permite que as leituras aconteçam durante uma gravação; sem isso, qualquer consulta trava o worker. busy_timeout=5000 faz o SQLite esperar até 5 s por um lock em vez de devolver database is locked na hora.

Passo 2: envie a tarefa e registre cada tentativa

O fluxo não muda: consultar o cache, gravar a linha de auditoria, enviar a tarefa para in.php e consultar res.php até o token chegar.

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

A linha é gravada antes do envio: se a requisição falhar, o registro da tentativa continua lá — e é justamente ele que falta na hora de explicar o consumo do mês. O status percorre submitted → polling → solved / error / timeout.

Passo 3: reaproveite o token dentro da janela de validade

Um token de reCAPTCHA v2 vale cerca de 90 a 120 segundos e serve para um único envio. O cache cobre o caso em que a automação repete a submissão do formulário segundos depois.

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

O TTL de 90 s é conservador de propósito, e a coluna used marca o token como consumido já na leitura, o que evita que duas threads reaproveitem o mesmo valor. Entre máquinas, veja o gerenciamento de TTL de tokens no Redis.

Passo 4: transforme o histórico em números

Com os dados gravados, acompanhar o pipeline vira SQL, não dashboard.

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() responde ao que se pergunta em cada revisão de sprint: quantas tarefas, quantas concluídas e qual o tempo médio em 24 horas. cleanup_old_records() apaga o histórico antigo e os tokens vencidos, e o VACUUM devolve o espaço em disco, porque o SQLite não encolhe o arquivo sozinho. Agende a limpeza diária no cron.

O mesmo padrão em Node.js

Com better-sqlite3, síncrono e sem camada de callbacks, o resultado é idêntico.

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;
}

A estrutura repete a versão em Python: insere, envia, consulta e atualiza, com instruções preparadas uma vez e reaproveitadas.

Um cenário concreto: QA autorizado com workers em São Paulo

Uma equipe de QA roda testes de integração contra o próprio ambiente de staging (https://staging.example.com/checkout), com workers em São Paulo (sa-east-1). Nas primeiras noites ninguém sabe se a lentidão vem da resolução do CAPTCHA ou da aplicação.

Duas semanas de dados encerram a discussão: AVG(elapsed_ms) mostra o custo real da resolução e a coluna polls indica se o intervalo de 5 s entre consultas gera espera desnecessária. O número também dimensiona o plano — no BASIC (US$ 15/mês, 5 threads), cinco resoluções simultâneas é o teto; com fila constante, o degrau seguinte é o STANDARD (US$ 30/mês, 15 threads). A cobrança da CaptchaAI é por thread simultânea, com resoluções ilimitadas por thread.

Um lembrete de conformidade: se os testes tocam dados de pessoas reais, a LGPD (ou o RGPD, em Portugal) vale também para esse arquivo — use dados fictícios em staging.

Erros comuns e como corrigir

Perguntas frequentes

O SQLite aguenta o volume de um pipeline em produção?

Com WAL ativo, ele sustenta a ordem de 1.000 gravações por segundo em disco local. Para uma máquina única com menos de 1.000 resoluções por hora, é confiável. Acima disso, migre para PostgreSQL ou MongoDB.

Posso usar o mesmo arquivo .db com vários workers?

Sim, desde que rodem no mesmo host e com WAL habilitado: as gravações ficam serializadas e o busy_timeout faz cada worker esperar em vez de falhar. Compartilhar o arquivo por NFS não funciona — o lock deixa de ser confiável.

Vale a pena cachear tokens de reCAPTCHA?

Só dentro da janela de validade e para o mesmo par sitekey/pageurl. Como o token é de uso único e expira em 90 a 120 segundos, o ganho aparece nas retentativas imediatas do mesmo formulário.

O que esse banco muda no meu custo com a CaptchaAI?

Ele mostra a sua concorrência real. Como os planos são cobrados por thread simultânea (BASIC US$ 15/mês com 5 threads, STANDARD US$ 30/mês com 15), o dado útil é o pico de resoluções em andamento, não o total do mês. Os planos estão no site da CaptchaAI.

Devo guardar o token completo da solução?

Guarde por pouco tempo, só para depurar, e deixe a limpeza apagar os registros antigos. Depois de expirado, o token não tem utilidade e só aumenta o arquivo.

Próximos passos

Crie as duas tabelas, ligue o WAL e agende a limpeza diária. Em uma semana você terá dados para dimensionar threads com números, não com impressões.

Leituras relacionadas:

• Rastreamento de resoluções em DynamoDB serverless
• Histórico e análise de resoluções com MongoDB
• Gerenciamento de TTL de tokens no Redis
• Como evitar resoluções duplicadas com locking no banco