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.
| Cenário | SQLite | Alternativa mais adequada |
|---|---|---|
| Desenvolvimento em máquina única | ✅ | - |
| Produção pequena (menos de 1.000 resoluções/hora) | ✅ | - |
| Registro de resultados de testes | ✅ | - |
| Produção em vários servidores | ⏳ | PostgreSQL, MongoDB |
| Alto throughput distribuído | ⏳ | Redis, DynamoDB |
| Painel de análise em tempo real | ⏳ | TimescaleDB, InfluxDB |
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
| Problema | Causa | Correção |
|---|---|---|
database is locked |
Gravações simultâneas sem WAL | Adicione PRAGMA journal_mode=WAL |
| O arquivo cresce sem parar | Sem rotina de limpeza | Execute cleanup_old_records() diariamente |
| Consultas lentas com muitos registros | Índices ausentes | Crie índices em submitted_at e type |
| O cache devolve token vencido | Entradas expiradas não removidas | Filtre por expires_at e limpe antes da leitura |
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:
