Erstellen eines CaptchaAI-Nutzungs-Dashboards und einer Überwachung

Für ein brauchbares Nutzungs-Dashboard brauchen Sie weder eine Zeitreihendatenbank noch einen Agenten auf jedem Worker: eine CSV-Zeile pro Lösungsversuch, drei Auswertungsfunktionen und eine Abfrage des Guthabens genügen. Dieser Leitfaden baut genau das in Python. Am Ende beantworten Ihre eigenen Zahlen die drei Fragen, die im Betrieb zählen: Läuft es stabil, wie lange dauert eine Lösung gerade, und reicht der gebuchte Plan für die Last?

Aus der Praxis: Ein Team in Hamburg betreibt seine Scraping-Worker auf einem Hetzner-VPS und deployt über GitLab CI. Gebucht ist ADVANCE (90 $/Monat, 50 Threads). Die Frage in der Monatsrunde lautet nie „Wie viele CAPTCHAs haben wir gelöst?“, sondern „Waren die 50 Threads jemals gleichzeitig belegt?“ – ohne eigene Messwerte reine Raterei.

Warum Sie die CaptchaAI-Nutzung trotz Thread-Abrechnung überwachen sollten

CaptchaAI rechnet pro gleichzeitigem Thread ab, nicht pro Lösung, und jeder Plan enthält unbegrenzte Lösungen pro Thread. Die Frage „Was kostet uns eine einzelne Lösung?“ verliert damit an Gewicht – dafür rücken drei andere nach vorn:

• Auslastung: Wie viele Threads waren gleichzeitig belegt? Nur das entscheidet, ob ein Plan-Wechsel etwas bringt.
• Qualität: Wie entwickelt sich die Erfolgsquote je Methode? Ein Einbruch bei genau einem Typ ist fast immer ein Integrationsproblem.
• Latenz: Wie lange dauert eine Lösung im Median und im P95? Diese Zahl bestimmt den Durchsatz Ihrer Pipeline.

Keine dieser Fragen beantwortet die Kontoansicht eines Anbieters – nur Ihr eigenes Log.

Welche Kennzahlen ein CaptchaAI-Dashboard abdecken muss

Alles davon steckt in einer Zeile pro Versuch: Zeitstempel, Methode, Dauer, Status, Fehlercode, Task-ID.

Schritt 1: Der Metriksammler

Der Sammler schreibt jeden Versuch sofort in eine CSV und hält eine Zusammenfassung der laufenden Sitzung im Speicher. Das Sperren über threading.Lock ist Pflicht, sobald mehrere Threads gleichzeitig lösen – bei Thread-basierter Abrechnung der Normalfall.

import time
import csv
import datetime
import threading
from collections import defaultdict


class MetricsCollector:
    """Collect and store CaptchaAI solve metrics."""

    def __init__(self, log_file="captchaai_metrics.csv"):
        self.log_file = log_file
        self.lock = threading.Lock()
        self.session_stats = defaultdict(lambda: {
            "count": 0, "success": 0, "error": 0,
            "timeout": 0, "total_time": 0,
        })
        self._init_log()

    def _init_log(self):
        try:
            with open(self.log_file, "r"):
                pass
        except FileNotFoundError:
            with open(self.log_file, "w", newline="") as f:
                writer = csv.writer(f)
                writer.writerow([
                    "timestamp", "method", "duration_s",
                    "status", "error_code", "task_id",
                ])

    def record(self, method, duration, status, error_code="", task_id=""):
        """Record a solve attempt."""
        with self.lock:
            # Update in-memory stats
            stats = self.session_stats[method]
            stats["count"] += 1
            stats["total_time"] += duration
            if status == "success":
                stats["success"] += 1
            elif status == "timeout":
                stats["timeout"] += 1
            else:
                stats["error"] += 1

            # Write to CSV
            with open(self.log_file, "a", newline="") as f:
                writer = csv.writer(f)
                writer.writerow([
                    datetime.datetime.utcnow().isoformat(),
                    method, f"{duration:.2f}",
                    status, error_code, task_id,
                ])

    def get_session_summary(self):
        """Get current session statistics."""
        summary = {}
        for method, stats in self.session_stats.items():
            avg_time = (
                stats["total_time"] / stats["count"]
                if stats["count"] > 0 else 0
            )
            success_rate = (
                stats["success"] / stats["count"] * 100
                if stats["count"] > 0 else 0
            )
            summary[method] = {
                "total": stats["count"],
                "success": stats["success"],
                "errors": stats["error"],
                "timeouts": stats["timeout"],
                "success_rate": f"{success_rate:.1f}%",
                "avg_time": f"{avg_time:.1f}s",
            }
        return summary

Zwei Details tragen die ganze Auswertung: Die Datei wird beim ersten Start mit Kopfzeile angelegt, und jeder Versuch wird geschrieben, auch der fehlgeschlagene. Ein Log, das nur Erfolge kennt, zeigt später eine makellose Erfolgsquote – und ist wertlos.

Schritt 2: Den Solver instrumentieren

Damit keine Lösung am Log vorbeiläuft, kapseln Sie Übermittlung und Polling in einer Klasse. Der finally-Block ist das Herzstück: Er protokolliert auch dann, wenn ein Timeout eintrat oder die API einen Fehlercode zurückgab.

import requests
import time


class MonitoredSolver:
    """Solver with automatic metric collection."""

    def __init__(self, api_key, metrics=None):
        self.api_key = api_key
        self.base = "https://ocr.captchaai.com"
        self.metrics = metrics or MetricsCollector()

    def solve(self, method, **params):
        start = time.time()
        task_id = ""
        status = "error"
        error_code = ""

        try:
            # Submit
            data = {"key": self.api_key, "method": method, "json": 1}
            data.update(params)
            resp = requests.post(
                f"{self.base}/in.php", data=data, timeout=30,
            )
            result = resp.json()

            if result.get("status") != 1:
                error_code = result.get("request", "UNKNOWN")
                raise RuntimeError(f"Submit error: {error_code}")

            task_id = result["request"]

            # Poll
            token = self._poll(task_id)
            status = "success"
            return token

        except TimeoutError:
            status = "timeout"
            raise
        except Exception as e:
            error_code = str(e)[:50]
            raise
        finally:
            duration = time.time() - start
            self.metrics.record(method, duration, status, error_code, task_id)

    def _poll(self, task_id, timeout=120):
        start = time.time()
        while time.time() - start < timeout:
            time.sleep(5)
            resp = requests.get(f"{self.base}/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)
            data = resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                if data.get("status") == 1:
                    return data["request"]
                raise RuntimeError(f"Solve error: {data['request']}")
        raise TimeoutError("Poll timeout")

    def print_summary(self):
        """Print current session metrics."""
        summary = self.metrics.get_session_summary()
        print("\n=== CaptchaAI Usage Summary ===")
        for method, stats in summary.items():
            print(f"\n{method}:")
            for key, value in stats.items():
                print(f"  {key}: {value}")


# Usage
metrics = MetricsCollector()
solver = MonitoredSolver("YOUR_API_KEY", metrics)

# Solve some CAPTCHAs
for i in range(10):
    try:
        token = solver.solve(
            "userrecaptcha",
            googlekey="SITE_KEY",
            pageurl="https://example.com",
        )
    except Exception as e:
        print(f"Failed: {e}")

# Print results
solver.print_summary()

Die Methode ist bewusst generisch: method und die typspezifischen Parameter reichen Sie durch – für reCAPTCHA v2 googlekey und pageurl, für Cloudflare Turnstile sitekey und pageurl. So landen alle Typen in derselben Auswertung.

Schritt 3: Berichte aus den Rohdaten erzeugen

Die Auswertung liest dieselbe CSV und aggregiert sie in drei Blickrichtungen: pro Tag, pro Methode und pro Fehlercode. Damit ist die Trendfrage („wird es schlechter?“) sauber von der Ursachenfrage („woran liegt es?“) getrennt.

import csv
import datetime
from collections import defaultdict


class UsageReport:
    """Generate usage reports from metrics CSV."""

    def __init__(self, log_file="captchaai_metrics.csv"):
        self.log_file = log_file

    def _load_data(self, days=None):
        """Load metrics, optionally filtered by date range."""
        cutoff = None
        if days:
            cutoff = datetime.datetime.utcnow() - datetime.timedelta(days=days)

        records = []
        with open(self.log_file, "r") as f:
            reader = csv.DictReader(f)
            for row in reader:
                ts = datetime.datetime.fromisoformat(row["timestamp"])
                if cutoff and ts < cutoff:
                    continue
                row["_ts"] = ts
                row["_duration"] = float(row["duration_s"])
                records.append(row)
        return records

    def daily_summary(self, days=7):
        """Summarize usage per day."""
        records = self._load_data(days=days)
        by_day = defaultdict(lambda: {"count": 0, "success": 0, "total_time": 0})

        for rec in records:
            day = rec["_ts"].date().isoformat()
            by_day[day]["count"] += 1
            if rec["status"] == "success":
                by_day[day]["success"] += 1
            by_day[day]["total_time"] += rec["_duration"]

        print(f"=== Daily Summary (last {days} days) ===")
        print(f"{'Date':<12} {'Total':>6} {'Success':>8} {'Rate':>7} {'Avg Time':>9}")
        for day in sorted(by_day.keys()):
            stats = by_day[day]
            rate = stats["success"] / stats["count"] * 100 if stats["count"] > 0 else 0
            avg = stats["total_time"] / stats["count"] if stats["count"] > 0 else 0
            print(f"{day:<12} {stats['count']:>6} {stats['success']:>8} {rate:>6.1f}% {avg:>8.1f}s")

    def method_breakdown(self, days=30):
        """Summarize usage by CAPTCHA type."""
        records = self._load_data(days=days)
        by_method = defaultdict(lambda: {"count": 0, "success": 0, "total_time": 0})

        for rec in records:
            method = rec["method"]
            by_method[method]["count"] += 1
            if rec["status"] == "success":
                by_method[method]["success"] += 1
            by_method[method]["total_time"] += rec["_duration"]

        print(f"\n=== Method Breakdown (last {days} days) ===")
        print(f"{'Method':<25} {'Total':>6} {'Success':>8} {'Rate':>7} {'Avg Time':>9}")
        for method in sorted(by_method.keys()):
            stats = by_method[method]
            rate = stats["success"] / stats["count"] * 100
            avg = stats["total_time"] / stats["count"]
            print(f"{method:<25} {stats['count']:>6} {stats['success']:>8} {rate:>6.1f}% {avg:>8.1f}s")

    def error_breakdown(self, days=7):
        """Show error distribution."""
        records = self._load_data(days=days)
        errors = defaultdict(int)

        for rec in records:
            if rec["status"] != "success" and rec["error_code"]:
                errors[rec["error_code"]] += 1

        if errors:
            print(f"\n=== Error Breakdown (last {days} days) ===")
            for error, count in sorted(errors.items(), key=lambda x: -x[1]):
                print(f"  {error}: {count}")


# Usage
report = UsageReport()
report.daily_summary(days=7)
report.method_breakdown(days=30)
report.error_breakdown(days=7)

Lesen Sie die Ausgabe in dieser Reihenfolge: Tagesbericht (bricht die Erfolgsquote an einem Tag ein?), Methodenaufschlüsselung (alle Typen oder nur einer?), Fehlerverteilung. Häufen sich Parameterfehler, etwa durch einen veralteten Sitekey, liegt es an Ihrer Integration; häufen sich Timeouts bei nur einem Typ, sind eher Zielseite oder Polling-Intervall die Ursache.

Schritt 4: Guthabenverlauf und Ausgaben verfolgen

Das Guthaben fragen Sie über res.php mit action=getbalance ab. Ein Job, der den Wert stündlich in eine zweite CSV schreibt, ergibt nach wenigen Tagen eine Verbrauchskurve – damit erkennen Sie Lastspitzen und setzen einen Alarm, bevor das Konto leer ist.

import requests
import time
import csv
import datetime


class BalanceDashboard:
    """Track balance over time for spending analysis."""

    def __init__(self, api_key, log_file="balance_history.csv"):
        self.api_key = api_key
        self.log_file = log_file

    def record(self):
        resp = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        })
        balance = float(resp.json()["request"])

        with open(self.log_file, "a", newline="") as f:
            writer = csv.writer(f)
            writer.writerow([
                datetime.datetime.utcnow().isoformat(),
                f"{balance:.4f}",
            ])
        return balance

    def get_spending(self, hours=24):
        """Calculate spending over time period."""
        cutoff = datetime.datetime.utcnow() - datetime.timedelta(hours=hours)
        balances = []

        try:
            with open(self.log_file, "r") as f:
                reader = csv.reader(f)
                for row in reader:
                    ts = datetime.datetime.fromisoformat(row[0])
                    if ts > cutoff:
                        balances.append(float(row[1]))
        except FileNotFoundError:
            return 0

        if len(balances) < 2:
            return 0
        return balances[0] - balances[-1]

Rufen Sie record() zeitgesteuert auf, nicht bei jeder Lösung: Stündlich reicht, und die Datei bleibt klein genug für Monate.

Schwellenwerte, die einen Alarm rechtfertigen

Ein Dashboard ohne Schwellenwerte sieht sich nach zwei Wochen niemand mehr an. Als Ausgangspunkt dienen die dokumentierten Richtwerte je Typ: unter 60 s für reCAPTCHA v2, unter 10 s für Cloudflare Turnstile, unter 12 s für GeeTest v3 und unter 0,5 s für Bild-CAPTCHAs. Liegt Ihr Median dauerhaft darüber, prüfen Sie zuerst Polling-Intervall und Thread-Auslastung.

Beginnen Sie mit zwei Alarmen statt mit zehn: Erfolgsquote einer Methode 30 Minuten unter Ihrem Normalwert, oder Guthaben für rechnerisch weniger als 48 Stunden. Alles Weitere ergänzen Sie, sobald ein Vorfall ein fehlendes Signal offenlegt.

Monitoring skalieren: von der CSV zu Prometheus und Grafana

Solange ein einzelner Worker läuft, ist die CSV die richtige Antwort: kein zusätzlicher Dienst, jederzeit in der Tabellenkalkulation zu öffnen. Arbeiten mehrere Worker auf verschiedenen Hosts, exportieren Sie dieselben Werte mit der prometheus_client-Bibliothek als Counter und Histogram und lassen Grafana zeichnen – record() schreibt dann zusätzlich in die Registry.

Viele DACH-Teams wählen einen pragmatischen Zwischenschritt: Der Tagesbericht läuft als geplanter GitLab-CI-Job und landet als Artefakt – kein neuer Dienst, aber eine nachvollziehbare Historie.

Aufbewahrung und Datenschutz

Die Metrik-CSV enthält keine gelösten Tokens, aber Zeitstempel, Methoden und – je nach Erweiterung – Ziel-URLs. Wer Felder wie pageurl oder die IP-Adressen der eingesetzten Proxys mitschreibt, verarbeitet unter Umständen personenbezogene Daten im Sinne der DSGVO. Prüfen Sie Zweck, Rechtsgrundlage und Löschfristen, bevor Sie das Log um solche Felder erweitern.

Bewährte Praxis: Rohdaten täglich rotieren, Tagesaggregate länger vorhalten, alles Weitere löschen statt archivieren.

Fehlerbehebung

FAQ

Wie erkenne ich, ob meine Threads ausgelastet sind?

An der Zahl gleichzeitig offener Lösungen. Erweitern Sie den Sammler um einen Zähler, der beim Start erhöht und im finally-Block verringert wird, und schreiben Sie das Maximum pro Minute mit. Liegt es dauerhaft an der Thread-Grenze Ihres Plans, warten Jobs bereits.

Welche Kennzahl zeigt, dass ein größerer Plan sinnvoll ist?

Die Wartezeit vor der Übermittlung, nicht die Lösungszeit. Bleibt die Lösungszeit stabil, während die Warteschlange wächst, sind die Threads das Nadelöhr. Der Schritt von STANDARD (30 $/Monat, 15 Threads) auf ADVANCE (90 $/Monat, 50 Threads) hebt die Gleichzeitigkeit von 15 auf 50.

Wie lange sollte ich die Rohdaten aufbewahren?

30 Tage detailliert, danach nur noch Tagesaggregate. Legen Sie die Frist einmal fest und automatisieren Sie das Löschen.

Warum weichen meine Zahlen vom Verlauf im CaptchaAI-Konto ab?

Fast immer, weil Fehlversuche fehlen: Wird nur bei Erfolg protokolliert, zählt Ihr Log zu wenige Versuche. Zweithäufigste Ursache: mehrere Prozesse schreiben ohne Sperre in dieselbe Datei.

Lässt sich die Überwachung in eine CI-Pipeline einbauen?

Ja. Ein geplanter Job in GitLab CI oder GitHub Actions ruft den Tagesbericht auf, legt ihn als Artefakt ab und schlägt fehl, sobald die Erfolgsquote unter Ihren Schwellenwert rutscht.

Verwandte Leitfäden

• Guthaben prüfen und automatisch auffüllen
• Antwortformate der CaptchaAI-API verstehen

Zahlen statt Bauchgefühl – starten Sie die Überwachung Ihrer CaptchaAI-Nutzung.