SQLite: CAPTCHA-Token lokal cachen und Solves tracken

Öffnen drei parallele Worker innerhalb weniger Sekunden dasselbe Formular, geht derselbe Sitekey dreimal an die API – und belegt drei Threads für ein einziges Token. Eine SQLite-Datei neben dem Skript beendet dieses Muster: Sie hält ein gelöstes Token bis zum Ende seiner Gültigkeit bereit und protokolliert nebenbei Typ, Dauer, Anzahl der Abfragen und Ergebnis jedes Vorgangs.

Der wirtschaftliche Effekt ist dabei ein anderer, als viele erwarten. CaptchaAI rechnet pro gleichzeitigem Thread ab und nicht pro Lösung – BASIC kostet 15 $ pro Monat und enthält 5 Threads mit unbegrenzten Lösungen. Ein Cache-Treffer spart also keine Einzelgebühr, sondern gibt sofort einen Thread frei und überspringt einen kompletten Lösungsdurchlauf: Für reCAPTCHA v2 nennt CaptchaAI unter 60 Sekunden als Obergrenze, der Treffer aus der lokalen Datei kommt in Mikrosekunden. Der Hebel heißt hier Durchsatz, nicht Rechnungsbetrag.

SQLite braucht dafür weder Server noch Container noch Zugangsdaten: eine Datei, das Modul sqlite3 aus der Standardbibliothek, fertig.

Wann eine SQLite-Datei reicht – und wann nicht

Die Entscheidung hängt an einer Frage: Schreibt genau ein Host in die Datei, oder mehrere?

Bleibt alles auf einer Maschine – Entwicklungsrechner, ein kleiner VPS bei Hetzner oder netcup, ein einzelner GitLab-Runner –, ist SQLite die pragmatische Wahl. Sobald ein zweiter Host mitschreibt, gehören der Cache nach Redis und das Protokoll in eine Server-Datenbank.

Zwei Tabellen: Protokoll und Token-Cache

Das Schema trennt zwei Dinge, die oft in eine einzige Tabelle gepresst werden: die dauerhafte Historie und den kurzlebigen 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 ist das Langzeitgedächtnis: Jede Zeile hält fest, welcher Typ wann übermittelt wurde, wie viele Abfragen bis zum Ergebnis nötig waren und wie lange der Vorgang gedauert hat. token_cache ist das Gegenteil – der Primärschlüssel aus sitekey, pageurl und token verhindert Dubletten, expires_at begrenzt die Gültigkeit, used markiert ein ausgeliefertes Token als verbraucht.

Wichtig: Ein Token ist beim Ziel nur rund 90 bis 120 Sekunden gültig. Der Cache ist damit ein Deduplizierungsfenster von Sekunden – kein Tokenlager. Lösen Sie unmittelbar vor dem Absenden des Formulars und nutzen Sie den Cache für den Fall, dass mehrere Threads zeitgleich dasselbe Ziel bearbeiten.

Python: vom Setup bis zum Cleanup

Schritt 1: Verbindung öffnen und Schema anlegen

PRAGMA journal_mode=WAL ist keine Feinjustierung, sondern die Voraussetzung dafür, dass Leser und Schreiber sich nicht gegenseitig blockieren. busy_timeout=5000 gibt einem wartenden Schreibvorgang fünf Sekunden Zeit, statt sofort mit database is locked abzubrechen.

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

Schritt 2: Lösen, protokollieren, Status abfragen

Der Ablauf bleibt immer gleich: Cache prüfen, Datensatz anlegen, Aufgabe an in.php übermitteln, danach res.php im Fünf-Sekunden-Takt abfragen. Jeder Zustandswechsel landet in der Datenbank, bevor die nächste Anfrage rausgeht – nach einem Absturz bleibt so nachvollziehbar, wo ein Vorgang hängen blieb. Entscheidend ist die Unterscheidung zwischen CAPCHA_NOT_READY (weiter abfragen) und jedem anderen Rückgabewert (echter Fehler, abbrechen).

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

Schritt 3: Token für die Restlaufzeit vorhalten

Die TTL im Beispiel liegt bei 90 Sekunden und damit bewusst unter der tatsächlichen Gültigkeit – lieber ein Token zu früh verwerfen als eines ausliefern, das beim Ziel schon abgelaufen ist. get_cached_token setzt used=1 unmittelbar beim Lesen; damit erhält kein zweiter Worker dasselbe Token.

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

Schritt 4: Auswerten und aufräumen

get_stats beantwortet die beiden Fragen, die im Betrieb zählen: Wie viele Vorgänge liefen in den letzten 24 Stunden, und wie lange dauerten sie im Mittel? cleanup_old_records entfernt alte Datensätze samt abgelaufener Cache-Einträge und gibt den Speicher mit VACUUM zurück – ohne diesen Schritt wächst die Datei weiter, obwohl Zeilen verschwinden. Ein täglicher Cron-Eintrag genügt.

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

Node.js: dieselbe Logik mit better-sqlite3

Für Playwright- oder Puppeteer-Stacks bietet sich better-sqlite3 an: synchrone API, keine verschachtelten Callbacks, identisches Schema.

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

Praxisbeispiel: nächtlicher Testlauf auf eigenem Server

Ein im DACH-Raum verbreiteter Aufbau: Ein GitLab-CI-Runner auf einem kleinen Hetzner-Server startet nachts eine Selenium-Suite gegen die eigene Staging-Umgebung unter https://staging.example-app.test. Vier parallele Browser durchlaufen denselben Login-Pfad – jedes Mal mit demselben Sitekey.

Ohne lokalen Cache erzeugt jeder Browser seine eigene Anfrage. Liegt die SQLite-Datei im Workspace des Runners, bedient der erste abgeschlossene Vorgang alle Threads, die innerhalb derselben TTL-Spanne anklopfen. Am Morgen liefert get_stats(hours=12) die Kennzahlen für den Lauf. Dasselbe Muster funktioniert bei einem Shopware-Testshop oder auf Worker-VMs bei IONOS.

Aufbewahrung und Datenschutz

Die Protokolltabelle ist kein neutraler Datentopf: In pageurl und project landen je nach Anwendung Kennungen, die sich einer Person zuordnen lassen – nach DSGVO Grund genug, Löschfrist und Rechtsgrundlage zu klären. Drei Punkte vor dem ersten Produktivlauf:

• Aufbewahrungsdauer automatisiert durchsetzen – 30 Tage reichen fürs Debugging meist aus.
• Tokens nicht länger speichern als nötig; nach Ablauf sind sie ohnehin wertlos.
• Die Datei liegt unverschlüsselt auf der Platte: Dateirechte einschränken (chmod 600) und *.db in die .gitignore aufnehmen.

Fehlerbehebung

Häufige Fragen

Wie lange darf ein Token im Cache bleiben?

Nie länger, als es beim Ziel gültig ist – rund 90 bis 120 Sekunden. Setzen Sie die TTL sicherheitshalber darunter, im Beispiel auf 90 Sekunden. Alles darüber liefert Tokens aus, die die Zielseite längst verwirft.

Was hilft gegen database is locked?

WAL-Modus aktivieren und ein busy_timeout setzen – beides steht in get_db(). Damit blockieren sich Leser und Schreiber nicht mehr. Bleibt der Fehler, schreibt meist doch ein zweiter Host auf dieselbe Datei, etwa über eine Netzwerkfreigabe – dafür ist SQLite nicht gedacht.

Muss ich für die Protokolldatei die DSGVO beachten?

Sobald personenbezogene Daten darin landen: ja. URLs mit Nutzer- oder Auftragskennungen zählen dazu, IP-Adressen ebenfalls. Löschfrist festlegen, mit cleanup_old_records() technisch durchsetzen und den Zweck dokumentieren.

Senkt der Cache meine CaptchaAI-Rechnung?

Nein – und das ist kein Nachteil. Abgerechnet werden gleichzeitige Threads, nicht einzelne Lösungen; jeder Plan enthält unbegrenzte Lösungen pro Thread. Der Cache senkt Thread-Belegung und Wartezeit. Wer dauerhaft an der Thread-Grenze arbeitet, wechselt auf einen größeren Plan – STANDARD (30 $ pro Monat, 15 Threads) oder ADVANCE (90 $ pro Monat, 50 Threads).

Wann sollte ich auf Redis oder PostgreSQL wechseln?

Sobald ein zweiter Prozess auf einem anderen Host schreiben muss oder mehr als rund 1.000 Lösungen pro Stunde anfallen. Der Umzug ist unspektakulär: sqlite3 captcha_solves.db ".dump" exportiert Schema und Daten fast unverändert.

Verwandte Leitfäden

• Token-TTL mit Redis verwalten
• Strukturierte Logs für den CAPTCHA-Betrieb
• Antwortformate und Fehlercodes der API