MongoDB für CAPTCHA Solve History und Analytics

Wenn die Erfolgsquote Ihrer Automatisierung am Dienstagvormittag einbricht, hilft nur ein Datensatz pro Lösungsversuch, den Sie im Nachhinein abfragen können. MongoDB ist dafür die pragmatischste Wahl: Jeder Versuch wird ein Dokument, die Auswertung übernimmt die Aggregations-Pipeline, alte Datensätze räumt ein TTL-Index selbst weg.

Dieser Leitfaden zeigt Schema, Indizes, das Protokollieren über die CaptchaAI-API und vier Aggregationen für Erfolgsquote, Lösungszeit und Fehlercodes – in Python und Node.js.

Welche Fragen ein Lösungsprotokoll beantwortet

Diese Fragen soll die Sammlung später beantworten:

• Wie hoch war die Erfolgsquote in den letzten 24 Stunden, insgesamt und je CAPTCHA-Typ?
• Dauern reCAPTCHA-v2-Abfragen länger als gestern, oder betrifft das nur ein Projekt?
• Welche Fehlercodes häufen sich, und kommen sie aus einer einzigen Ziel-Domain?
• Wie viele Versuche laufen gleichzeitig – also wie viele Threads brauchen Sie?

MongoDB statt fester Tabellen: warum das Schema flexibel bleiben muss

Die Felder unterscheiden sich je CAPTCHA-Typ: reCAPTCHA v2 braucht googlekey und pageurl, Cloudflare Turnstile einen sitekey, GeeTest v3 die Werte gt und challenge, Bild-CAPTCHAs die Bilddaten im Feld body. In einer festen Tabelle endet das in Spalten voller Leerwerte. MongoDB speichert jedes Dokument mit genau den Feldern des jeweiligen Typs – auch ein neuer Typ wie CaptchaFox (Beta) passt ohne Schemaänderung hinein. Die Aggregation rechnet zudem serverseitig, statt 100.000 Dokumente in Ihr Skript zu holen.

Dokumentschema für CAPTCHA-Lösungen

{
  "_id": "ObjectId",
  "captcha_id": "12345678",
  "type": "recaptcha_v2",
  "method": "userrecaptcha",
  "sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
  "pageurl": "https://example.com/form",
  "status": "solved",
  "solution": "03AGdBq26...",
  "error": null,
  "submitted_at": "2026-04-04T10:15:30.000Z",
  "solved_at": "2026-04-04T10:15:45.000Z",
  "elapsed_ms": 15000,
  "polls": 3,
  "proxy_used": true,
  "cost": 0.00299,
  "metadata": {
    "project": "price-monitor",
    "worker_id": "worker-3",
    "target_domain": "example.com"
  }
}

Drei Felder tragen die spätere Auswertung: status als Zustandsmaschine (submitted → polling → solved, error oder timeout), elapsed_ms als Zeit zwischen Einreichen und Ergebnis und metadata als Kontext, der die Aufteilung nach Projekt, Worker oder Ziel-Domain erst möglich macht. Halten Sie elapsed_ms gegen die Erwartung pro Typ: Cloudflare Turnstile ist üblicherweise in unter 10 Sekunden gelöst, reCAPTCHA v2 in unter 60 Sekunden.

Das Feld cost verdient eine Einordnung: CaptchaAI rechnet pro gleichzeitigem Thread ab, nicht pro Lösung. Ein Kostenwert pro Dokument ist deshalb kein Rechnungsposten, sondern eine interne Umlage – Monatspreis geteilt durch die Lösungen im selben Zeitraum.

Python-Implementierung

Verbindung und Konfiguration

import os
import time
from datetime import datetime, timezone
from pymongo import MongoClient, ASCENDING, DESCENDING
import requests

MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017")
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

client = MongoClient(MONGO_URI)
db = client["captcha_tracking"]
solves = db["solves"]

Beide Zugangsdaten kommen aus Umgebungsvariablen: der Verbindungsstring in MONGO_URI, der Schlüssel in CAPTCHAAI_API_KEY. Datenbank und Sammlung legt MongoDB beim ersten Schreibvorgang selbst an – ein Migrationsschritt entfällt.

Indizes anlegen, bevor die Sammlung wächst

def setup_indexes():
    solves.create_index([("submitted_at", DESCENDING)])
    solves.create_index([("type", ASCENDING), ("status", ASCENDING)])
    solves.create_index([("metadata.project", ASCENDING)])
    solves.create_index([("metadata.target_domain", ASCENDING)])
    solves.create_index(
        [("submitted_at", ASCENDING)],
        expireAfterSeconds=90 * 24 * 3600,  # Auto-delete after 90 days
        name="ttl_cleanup"
    )

setup_indexes()

Jeder Index hat einen Zweck: submitted_at absteigend für Zeitfenster-Abfragen, type plus status für die Erfolgsquote je Typ, die beiden metadata-Indizes für die Aufteilung nach Projekt und Ziel-Domain. Der letzte Eintrag ist ein TTL-Index; MongoDB löscht Dokumente, sobald submitted_at älter als 90 Tage ist – ohne Cronjob und ohne Aufräumskript. Legen Sie die Indizes an, solange die Sammlung klein ist.

In einem Schritt lösen und protokollieren

def solve_and_store(sitekey, pageurl, captcha_type="recaptcha_v2", metadata=None):
    record = {
        "type": captcha_type,
        "method": "userrecaptcha",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "status": "submitted",
        "submitted_at": datetime.now(timezone.utc),
        "metadata": metadata or {}
    }

    result = solves.insert_one(record)
    doc_id = result.inserted_id

    # 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:
        solves.update_one(
            {"_id": doc_id},
            {"$set": {"status": "error", "error": data.get("request")}}
        )
        return None

    captcha_id = data["request"]
    solves.update_one(
        {"_id": doc_id},
        {"$set": {"captcha_id": captcha_id, "status": "polling"}}
    )

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

        if poll_resp.get("status") == 1:
            solved_at = datetime.now(timezone.utc)
            elapsed_ms = int(
                (solved_at - record["submitted_at"]).total_seconds() * 1000
            )
            solves.update_one({"_id": doc_id}, {"$set": {
                "status": "solved",
                "solution": poll_resp["request"],
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls
            }})
            return poll_resp["request"]

        if poll_resp.get("request") != "CAPCHA_NOT_READY":
            solves.update_one({"_id": doc_id}, {"$set": {
                "status": "error",
                "error": poll_resp.get("request"),
                "polls": polls
            }})
            return None

    solves.update_one({"_id": doc_id}, {"$set": {
        "status": "timeout", "polls": polls
    }})
    return None

Entscheidend ist die Reihenfolge: Das Dokument entsteht vor dem API-Aufruf, mit dem Status submitted. Wer erst nach dem Ergebnis schreibt, protokolliert nur die Erfolge und wundert sich später über eine verdächtig saubere Statistik.

Der Ablauf folgt dem Zweischritt der API: in.php nimmt die Aufgabe entgegen und liefert eine captcha_id, res.php gibt das Ergebnis zurück. Bis dahin antwortet der Endpunkt mit CAPCHA_NOT_READY; jeder andere Rückgabewert ist ein echter Fehlercode. Das gelöste Token tragen Sie sofort in das Formularfeld ein, denn es ist nur rund 120 Sekunden gültig.

Auswertungen mit der Aggregations-Pipeline

def get_success_rate(hours=24):
    """Success rate for the last N hours."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}}},
        {"$group": {
            "_id": "$status",
            "count": {"$sum": 1}
        }}
    ]
    results = {r["_id"]: r["count"] for r in solves.aggregate(pipeline)}
    total = sum(results.values())
    solved = results.get("solved", 0)
    return (solved / total * 100) if total else 0


def get_avg_solve_time_by_type():
    """Average solve time grouped by CAPTCHA type."""
    pipeline = [
        {"$match": {"status": "solved"}},
        {"$group": {
            "_id": "$type",
            "avg_time_ms": {"$avg": "$elapsed_ms"},
            "min_time_ms": {"$min": "$elapsed_ms"},
            "max_time_ms": {"$max": "$elapsed_ms"},
            "count": {"$sum": 1}
        }},
        {"$sort": {"count": -1}}
    ]
    return list(solves.aggregate(pipeline))


def get_hourly_solve_volume(days=7):
    """Hourly solve volume for charting."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}}},
        {"$group": {
            "_id": {
                "date": {"$dateToString": {"format": "%Y-%m-%d", "date": "$submitted_at"}},
                "hour": {"$hour": "$submitted_at"}
            },
            "total": {"$sum": 1},
            "solved": {"$sum": {"$cond": [{"$eq": ["$status", "solved"]}, 1, 0]}}
        }},
        {"$sort": {"_id.date": 1, "_id.hour": 1}}
    ]
    return list(solves.aggregate(pipeline))


def get_error_breakdown(hours=24):
    """Error frequency by error code."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}, "status": "error"}},
        {"$group": {"_id": "$error", "count": {"$sum": 1}}},
        {"$sort": {"count": -1}}
    ]
    return list(solves.aggregate(pipeline))

Die vier Funktionen decken den Alltag ab: get_success_rate() für den Anteil gelöster Versuche im Zeitfenster, get_avg_solve_time_by_type() für Lösungszeiten je Typ inklusive Minimum und Maximum, get_hourly_solve_volume() für Stundenwerte im Diagramm und get_error_breakdown() für die Häufigkeit der Fehlercodes.

Die Spanne sagt mehr als der Mittelwert: 14 Sekunden im Schnitt bei einem Maximum von 55 Sekunden sind etwas anderes als 14 bei 18. Dashboards lesen die Aggregationen direkt aus – MongoDB Charts, Metabase oder Grafana.

Dieselbe Logik in Node.js

const { MongoClient } = require("mongodb");
const axios = require("axios");

const MONGO_URI = process.env.MONGO_URI || "mongodb://localhost:27017";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

let db, solves;

async function connect() {
  const client = await MongoClient.connect(MONGO_URI);
  db = client.db("captcha_tracking");
  solves = db.collection("solves");

  await solves.createIndex({ submitted_at: -1 });
  await solves.createIndex({ type: 1, status: 1 });
  await solves.createIndex({ "metadata.project": 1 });
  await solves.createIndex(
    { submitted_at: 1 },
    { expireAfterSeconds: 90 * 24 * 3600 }
  );
}

async function solveAndStore(sitekey, pageurl, type = "recaptcha_v2", metadata = {}) {
  const submittedAt = new Date();
  const { insertedId } = await solves.insertOne({
    type, method: "userrecaptcha", sitekey, pageurl,
    status: "submitted", submitted_at: submittedAt, metadata,
  });

  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) {
    await solves.updateOne({ _id: insertedId }, { $set: { status: "error", error: submit.data.request } });
    return null;
  }

  const captchaId = submit.data.request;
  await solves.updateOne({ _id: insertedId }, { $set: { captcha_id: captchaId, status: "polling" } });

  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 solvedAt = new Date();
      await solves.updateOne({ _id: insertedId }, { $set: {
        status: "solved", solution: poll.data.request,
        solved_at: solvedAt, elapsed_ms: solvedAt - submittedAt, polls,
      }});
      return poll.data.request;
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      await solves.updateOne({ _id: insertedId }, { $set: { status: "error", error: poll.data.request, polls } });
      return null;
    }
  }

  await solves.updateOne({ _id: insertedId }, { $set: { status: "timeout", polls } });
  return null;
}

async function getSuccessRate(hours = 24) {
  const cutoff = new Date(Date.now() - hours * 3600 * 1000);
  const pipeline = [
    { $match: { submitted_at: { $gte: cutoff } } },
    { $group: { _id: "$status", count: { $sum: 1 } } },
  ];
  const results = await solves.aggregate(pipeline).toArray();
  const total = results.reduce((s, r) => s + r.count, 0);
  const solved = results.find((r) => r._id === "solved")?.count || 0;
  return total ? ((solved / total) * 100).toFixed(1) : 0;
}

Die Node.js-Fassung schreibt in dieselbe Sammlung und benutzt dieselben Feldnamen – sobald Python- und Node.js-Worker unterschiedliche Schlüssel setzen, zerfällt jede Aggregation in zwei Auswertungen.

Aufbewahrung der Lösungsdaten und Datenschutz

Wie lange die Datensätze bleiben, setzt der TTL-Index um:

Für Leser im DACH-Raum kommt ein zweiter Punkt hinzu: pageurl und metadata.target_domain können personenbezogene Angaben enthalten, etwa Vorgangsnummern in einer URL. Kürzen Sie URLs auf Pfad oder Domain und halten Sie die Aufbewahrungsfrist schriftlich fest. Der TTL-Index belegt technisch, dass Daten nach einer definierten Frist verschwinden – statt „irgendwann“.

Aus der Praxis: drei Projekte, ein Thread-Budget

Eine Agentur in Hamburg betreibt Preis-Monitoring für drei Shopware-Kunden. Die Worker laufen als nächtlicher GitLab-CI-Job und schreiben mit metadata.project in dieselbe Sammlung. Nach zwei Wochen beantwortet die Aggregation, worüber vorher gestritten wurde: Kunde B verursacht knapp zwei Drittel aller Versuche, hat aber die niedrigere Fehlerquote – die Ziel-Domain von Kunde A reagiert schlicht langsamer.

Wichtiger ist die zweite Auswertung: Aus submitted_at und solved_at ergibt sich die höchste Zahl gleichzeitig offener Versuche – und genau die ist die Abrechnungsgröße, denn CaptchaAI berechnet gleichzeitige Threads bei unbegrenzten Lösungen pro Thread. Liegt die Spitze bei vier parallelen Versuchen, genügt BASIC (15 $/Monat, 5 Threads); wandert sie mit einem vierten Kunden Richtung 40, passt ADVANCE (90 $/Monat, 50 Threads). Preise in US-Dollar.

Typische Fehlerbilder

Häufige Fragen

Welche Indizes brauche ich wirklich?

Zwei genügen zum Start: submitted_at absteigend für Zeitfenster-Abfragen und type plus status für die Erfolgsquote je Typ. Die metadata-Indizes lohnen sich, sobald mehrere Projekte in derselben Sammlung liegen.

Wie erkenne ich, ob mein Thread-Budget ausreicht?

Über die Zeitstempel. Zählen Sie je Minute die gleichzeitig offenen Versuche und nehmen Sie das Tagesmaximum. Liegt es dauerhaft an der Thread-Zahl Ihres Plans und steigt zugleich elapsed_ms, warten Anfragen auf einen freien Thread.

Darf ich pageurl und Ziel-Domain dauerhaft speichern?

Technisch ja, sinnvoll selten. URLs können personenbezogene Parameter enthalten; eine kurze TTL-Frist und das Kürzen auf Pfad oder Domain sind der pragmatische Weg. Für die Auswertung genügen Typ, Status, Dauer und Fehlercode.

Was mache ich mit Datensätzen, die auf polling stehen bleiben?

Sie sind ein Signal, kein Datenmüll: Der Worker wurde während der Abfrage beendet. Ein kleiner Aufräumlauf setzt solche Einträge nach einer Karenzzeit auf timeout. Ihre Häufigkeit ist ein Frühindikator für instabile Worker oder zu kurze CI-Laufzeiten.

Läuft das auch mit MongoDB Atlas?

Ja. TTL-Indizes und Aggregations-Pipelines verhalten sich dort identisch; Sie tragen nur den Connection-String aus dem Atlas-Dashboard in MONGO_URI ein.

Weiterlesen

• Ergebnisse ohne Datenbankserver lokal in SQLite ablegen
• Erfolgsquoten laufend mit Prometheus und Grafana überwachen
• Langfristige Trends der Lösungszeiten als Zeitreihe auswerten