SQLite : cache de tokens et suivi des résolutions CAPTCHA

Un pipeline de résolution CAPTCHA sans journal local est une boîte noire : impossible de savoir quel sitekey échoue, quel type traîne, ni combien de demandes visaient une page déjà résolue trente secondes plus tôt. Un fichier SQLite de quelques kilo-octets répond à ces trois questions, sans serveur à installer. Voici le schéma, le cache de tokens borné par TTL et les requêtes de suivi.

Ce que la journalisation locale vous apporte

Le diagnostic. Un elapsed_ms et un compteur de polls par ligne montrent si un ralentissement vient du service, de votre boucle d'interrogation ou d'un sitekey précis.
La réutilisation dans la fenêtre TTL. Un token reCAPTCHA reste valide 90 à 120 s. Si deux workers visent la même page dans cet intervalle, le second consomme le token déjà obtenu : un thread libéré, puisque CaptchaAI facture des threads simultanés et non des résolutions à l'unité.
La mesure. Sans historique, « ça marche bien » reste une impression ; avec deux colonnes et une agrégation, c'est un chiffre daté.

SQLite ou base serveur : où passe la frontière

Situation	SQLite	Alternative plus adaptée
Développement sur un seul poste	✅	-
Production légère (< 1 000 résolutions/heure)	✅	-
Historique des campagnes de tests QA	✅	-
Workers répartis sur plusieurs serveurs	⏳	PostgreSQL, MongoDB
Débit élevé et distribué	⏳	Redis, DynamoDB
Tableau de bord temps réel	⏳	TimescaleDB, InfluxDB

La frontière tient au nombre de machines, pas au volume : un processus unique encaisse sans peine la charge d'un worker modeste. Dès que deux serveurs doivent partager le même cache, passez à une base réseau.

Le schéma : une table de suivi, une table de cache

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

captcha_solves garde une ligne par demande, de l'état submitted jusqu'à solved, error ou timeout. token_cache ne garde que le réutilisable : sitekey, pageurl, token, expiration et un drapeau used qui empêche de servir deux fois la même valeur. Attention aux données : pageurl peut contenir un identifiant de session. Si le RGPD s'applique à votre équipe, tronquez l'URL à son chemin avant l'insertion.

Le pipeline en Python, étape par étape

Étape 1 : ouvrir la base en mode WAL

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

Deux PRAGMA font l'essentiel. journal_mode=WAL sépare lectures et écritures : votre tableau de bord interroge la base pendant qu'un worker écrit. busy_timeout=5000 fait patienter SQLite avant de renvoyer database is locked.

Étape 2 : envoyer la demande, puis journaliser chaque transition

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

L'ordre compte : le cache est interrogé avant tout appel réseau, la ligne de suivi est insérée avant l'envoi à in.php, et l'id renvoyé est écrit dès son arrivée. Si le processus meurt pendant le polling, la ligne reste en polling avec son captcha_id : vous reprenez l'interrogation plus tard au lieu de relancer une résolution.

Étape 3 : réutiliser un token encore valide

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

Le TTL par défaut est fixé à 90 s, volontairement sous la durée de vie réelle : mieux vaut redemander une résolution que d'injecter un token périmé. used=1 est posé à la lecture, pour qu'un second worker ne récupère pas la même valeur. Ajustez le TTL au type visé : la durée de validité varie d'un fournisseur à l'autre.

Étape 4 : mesurer le taux de réussite et purger

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() répond aux deux questions d'exploitation : quelle part des demandes aboutit, et en combien de temps. Programmez-la chaque nuit et conservez la sortie comme référence. cleanup_old_records() supprime les lignes anciennes, vide les entrées expirées et lance un VACUUM.

Le même schéma en Node.js

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

better-sqlite3 est synchrone : aucune promesse à enchaîner autour des écritures, et les transitions d'état restent lisibles au milieu d'un flot async. Les colonnes sont identiques à la version Python, si bien que deux workers écrits dans deux langages partagent le même fichier.

Un cas concret : tests de recette chez un éditeur SaaS

Une équipe QA lyonnaise valide chaque nuit son parcours d'inscription sur un environnement de recette hébergé chez OVHcloud. Le formulaire est protégé par reCAPTCHA v2, et la suite rejoue le même écran une trentaine de fois.

Avec token_cache et un TTL de 90 s, les tests qui tombent dans la même fenêtre réutilisent le token déjà obtenu. Le vrai gain est venu après trois semaines de journalisation : l'agrégation a montré qu'un seul sitekey de recette concentrait presque tous les états timeout, sur une page jamais redéployée. L'équipe purge à sept jours plutôt qu'à trente, les URL de recette contenant des identifiants de comptes de test.

Dépannage

Problème	Cause	Correctif
`database is locked`	Écritures simultanées sans mode WAL	Ajoutez `PRAGMA journal_mode=WAL` et un `busy_timeout`
Le fichier grossit sans limite	Aucune purge planifiée	Exécutez `cleanup_old_records()` chaque nuit
Requêtes lentes après quelques semaines	Index absents sur les colonnes filtrées	Créez les index sur `submitted_at` et `type`
Deux workers reçoivent le même token	Drapeau `used` posé trop tard	Écrivez `used=1` dans la transaction de lecture

Questions fréquentes

Faut-il activer le mode WAL dès le premier jour ?

Oui. C'est une ligne de code, et elle supprime la principale cause de database is locked dès qu'un second processus lit la base. Le mode est persistant : posé une fois sur le fichier, il vaut pour les connexions suivantes.

Comment éviter de relancer une résolution pour une page déjà traitée ?

C'est le rôle de la clé composite (sitekey, pageurl) : la fonction interroge la table locale avant tout appel réseau et ne repart vers l'API qu'à défaut d'entrée valide. Pour les accès concurrents entre workers, voyez comment éviter les résolutions en double avec un verrou en base.

Ce cache fonctionne-t-il pour tous les types de CAPTCHA ?

Il fonctionne pour ce que CaptchaAI résout : reCAPTCHA v2 et v3 (Enterprise inclus), Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3, CAPTCHA image/OCR, grilles d'images et BLS CAPTCHA, plus CaptchaFox (bêta), Friendly Captcha (bêta) et Lemin (bêta). hCaptcha et FunCaptcha (Arkose Labs) ne sont pas pris en charge ; GeeTest v4 est annoncé comme à venir.

Combien de threads pour 1 000 résolutions par heure ?

Cela dépend du temps de résolution que vous mesurez — c'est ce que donne la colonne elapsed_ms. Sur une médiane à 15 s, un thread absorbe environ 240 demandes par heure, soit de l'ordre de 5 threads pour 1 000, ce que couvre BASIC ($15/mois, 5 threads). Un cache efficace réduit ce besoin : les demandes servies localement ne consomment aucun thread.