CaptchaAI-Guthabenprüfung und automatische Nachfüllintegration

Ihr aktueller CaptchaAI-Guthabenstand steht hinter genau einem GET-Request: res.php mit action=getbalance. Interessant wird erst, was danach kommt: wann Sie abfragen, ab welchem Wert Sie warnen und wie Sie verhindern, dass ein Nachtlauf um 3 Uhr stehen bleibt, weil das Konto leer gelaufen ist.

Dieser Leitfaden geht den Weg komplett durch: Abfrage, Prüfung vor dem Pipeline-Start, Dauermonitor mit Slack-Benachrichtigung, Verbrauchsprotokoll und Einbindung in den Lösungs-Workflow. Alle Beispiele sind Python und brauchen kein SDK – requests genügt.

Guthaben per API abfragen

Der Endpunkt ist https://ocr.captchaai.com/res.php, die Aktion heißt getbalance, und mit json=1 bekommen Sie eine auswertbare Antwort statt reinem Text:

import requests

API_KEY = "YOUR_API_KEY"

resp = requests.get("https://ocr.captchaai.com/res.php", params={
    "key": API_KEY,
    "action": "getbalance",
    "json": 1,
})

data = resp.json()
balance = float(data["request"])
print(f"Balance: ${balance:.2f}")

Antwortformat:

{"status": 1, "request": "12.345"}

status: 1 bedeutet nur, dass die Abfrage funktioniert hat – nicht, dass genug Guthaben vorhanden ist. Der Wert steckt als Zeichenkette in request und muss vor jedem Vergleich in float umgewandelt werden. Ohne json=1 kommt derselbe Wert als reiner Text zurück; in Python ist die JSON-Variante robuster.

Was der Wert bei Thread-basierter Abrechnung bedeutet

CaptchaAI rechnet pro gleichzeitigem Thread ab, nicht pro gelöstem CAPTCHA. Ein Plan wie BASIC (15 $/Monat, 5 Threads) oder ADVANCE (90 $/Monat, 50 Threads) enthält unbegrenzte Lösungen pro Thread; der Durchsatz hängt an der Zahl paralleler Threads und an der Lösungszeit des CAPTCHA-Typs. Daraus folgen zwei Punkte fürs Monitoring.

Klären Sie die Einheit, bevor Sie Grenzen festlegen. Die API-Dokumentation beschreibt den Rückgabewert als Gesamtzahl der Plan-Threads (Beispielwert 600); die Beispiele hier formatieren ihn wie 2Captcha-kompatible Clients als Dollarbetrag. Prüfen Sie den Rohwert Ihres Kontos einmal und rechnen Sie Ihre Alarmgrenzen auf diese Einheit um.

Leeres Konto und ausgelastete Threads erzeugen denselben Fehler. in.php antwortet in beiden Fällen mit ERROR_ZERO_BALANCE. Laut API-Dokumentation liefert action=threadsinfo zusätzlich threads und working_threads – die Auslastungsseite derselben Medaille.

Guthabenprüfung vor dem Pipeline-Start

Der günstigste Fehlschlag passiert vor dem ersten Task. Die folgende Funktion prüft den Stand, gibt bei Unterschreitung None zurück und beendet den Prozess mit Exit-Code 1 – so schlägt ein CI-Job sauber fehl, statt Hunderte Tasks ins Leere zu schicken:

import requests
import sys


def check_balance(api_key, min_required=1.0):
    """Check balance and abort if too low."""
    resp = requests.get("https://ocr.captchaai.com/res.php", params={
        "key": api_key,
        "action": "getbalance",
        "json": 1,
    })
    data = resp.json()

    if data.get("status") != 1:
        print(f"Balance check failed: {data.get('request')}")
        return None

    balance = float(data["request"])
    print(f"Current balance: ${balance:.2f}")

    if balance < min_required:
        print(f"WARNING: Balance ${balance:.2f} below minimum ${min_required:.2f}")
        return None

    return balance


# Usage
API_KEY = "YOUR_API_KEY"
balance = check_balance(API_KEY, min_required=5.0)

if balance is None:
    print("Insufficient balance. Add funds before running pipeline.")
    sys.exit(1)

print(f"Balance OK (${balance:.2f}). Starting pipeline...")

Zwei Details lohnen den Blick. Der status-Check fängt einen ungültigen API-Schlüssel ab, bevor float() über einen Fehlerstring stolpert. Und min_required gehört in die Konfiguration: In einer GitLab-CI-Pipeline setzen Sie den Wert als Variable pro Umgebung, damit der nächtliche Massenlauf eine höhere Untergrenze verlangt als ein Smoke-Test.

Welcher Schwellenwert ist sinnvoll?

Ein brauchbarer Schwellenwert deckt den Zeitraum ab, den Sie realistisch zum Nachladen brauchen. Drei Fragen führen zur Zahl:

1. Wie hoch ist der Verbrauch in der Spitzenstunde? Der Tracker weiter unten liefert den Wert nach zwei bis drei Tagen Betrieb.
2. Wie lange dauert Ihr Nachlade-Prozess? Eine Kartenzahlung ist in Minuten durch. Läuft die Aufladung über die Buchhaltung mit Freigabe und Rechnungslauf, sind zwei Werktage realistisch.
3. Was kostet ein Stillstand? Ein Crawler, der eine Nacht aussetzt, ist ärgerlich; ein Terminportal, das am Freigabetag nicht geprüft wird, ist teuer.

Als Startpunkt bewährt hat sich der Verbrauch von 48 Stunden, ergänzt um eine zweite, deutlich niedrigere Warnstufe.

Dauermonitor mit Warnung bei niedrigem Stand

Die folgende Klasse fragt in festen Intervallen ab, hält eine kurze Historie im Speicher, meldet einmal pro Unterschreitung statt im Minutentakt und schätzt die verbleibende Laufzeit:

import requests
import time
import smtplib
from email.message import EmailMessage


class BalanceMonitor:
    """Monitor CaptchaAI balance and send alerts."""

    def __init__(self, api_key, alert_threshold=5.0, check_interval=300):
        self.api_key = api_key
        self.alert_threshold = alert_threshold
        self.check_interval = check_interval  # seconds
        self.base_url = "https://ocr.captchaai.com"
        self.history = []
        self.alerted = False

    def get_balance(self):
        resp = requests.get(f"{self.base_url}/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        }, timeout=10)
        data = resp.json()
        return float(data["request"])

    def check_and_alert(self):
        balance = self.get_balance()
        self.history.append({
            "time": time.time(),
            "balance": balance,
        })

        print(f"Balance: ${balance:.2f}")

        if balance < self.alert_threshold and not self.alerted:
            self.send_alert(balance)
            self.alerted = True
        elif balance >= self.alert_threshold:
            self.alerted = False

        return balance

    def send_alert(self, balance):
        """Send low-balance alert. Override for your notification system."""
        print(f"ALERT: Balance low! ${balance:.2f} < ${self.alert_threshold:.2f}")
        # Add your notification logic:
        # - Email, Slack webhook, SMS, etc.

    def get_spending_rate(self, hours=1):
        """Calculate spending rate over the last N hours."""
        cutoff = time.time() - (hours * 3600)
        recent = [h for h in self.history if h["time"] > cutoff]

        if len(recent) < 2:
            return 0.0

        spent = recent[0]["balance"] - recent[-1]["balance"]
        return max(0.0, spent)

    def estimate_remaining_hours(self):
        """Estimate how many hours until balance runs out."""
        rate = self.get_spending_rate(hours=1)
        if rate <= 0:
            return float("inf")

        balance = self.history[-1]["balance"] if self.history else 0
        return balance / rate

    def run(self):
        """Run continuous monitoring."""
        print(f"Monitoring balance (alert at ${self.alert_threshold:.2f})")
        while True:
            try:
                self.check_and_alert()
                rate = self.get_spending_rate()
                remaining = self.estimate_remaining_hours()
                print(f"  Spending: ${rate:.2f}/hr, ~{remaining:.1f}hrs remaining")
            except Exception as e:
                print(f"Monitor error: {e}")
            time.sleep(self.check_interval)


# Usage
monitor = BalanceMonitor(
    api_key="YOUR_API_KEY",
    alert_threshold=5.0,
    check_interval=300,  # Check every 5 minutes
)
monitor.run()

Das alerted-Flag ist der wichtigste Teil dieser Klasse. Ohne das Zurücksetzen bei erholtem Guthaben erzeugt jeder Durchlauf eine neue Meldung – und nach zwei Stunden liest im Team niemand mehr den Kanal. check_interval=300, also alle 5 Minuten, ist ein guter Kompromiss zwischen Reaktionszeit und Rauschen.

Im Dauerbetrieb gehört der Monitor in einen eigenen Prozess. Ein systemd-Service auf demselben Hetzner- oder netcup-Server, auf dem Ihre Worker laufen, ist unkomplizierter als ein Cronjob: Mit Restart=always kommt er von selbst wieder hoch.

Warnung an Slack weiterreichen

send_alert ist im Monitor bewusst leer gelassen. In der Praxis hängt dort ein Incoming Webhook – Slack, Mattermost und Microsoft Teams reagieren auf denselben POST mit JSON-Payload:

import requests


def send_slack_alert(webhook_url, balance, threshold):
    """Send balance alert to Slack channel."""
    payload = {
        "text": f":warning: CaptchaAI balance low!",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        f"*CaptchaAI Balance Alert*\n"
                        f"Current balance: *${balance:.2f}*\n"
                        f"Alert threshold: ${threshold:.2f}\n"
                        f"Action: Add funds at captchaai.com"
                    ),
                },
            },
        ],
    }
    requests.post(webhook_url, json=payload)


# Add to BalanceMonitor.send_alert():
# send_slack_alert(SLACK_WEBHOOK, balance, self.alert_threshold)

Die Webhook-URL gehört in eine Umgebungsvariable oder einen Secret-Store – im Repository ist sie ein Zugang, den jeder mit Lesezugriff nutzen kann. Formulieren Sie die Meldung so, dass der Bereitschaftsdienst ohne Rückfrage handeln kann: Stand, Schwellenwert, geschätzte Restlaufzeit, nächste Handlung.

Verbrauch protokollieren statt schätzen

Eine Warnung sagt Ihnen, dass es eng wird. Ein Protokoll sagt Ihnen, warum. Der folgende Tracker schreibt jeden gemessenen Stand mit Zeitstempel in eine CSV-Datei und leitet daraus den Tagesverbrauch ab:

import csv
import datetime


class SpendingTracker:
    """Track CaptchaAI spending over time."""

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

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

    def record_balance(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_daily_spending(self):
        """Calculate today's spending from log."""
        today = datetime.date.today().isoformat()
        balances = []

        with open(self.log_file, "r") as f:
            reader = csv.DictReader(f)
            for row in reader:
                if row["timestamp"].startswith(today):
                    balances.append(float(row["balance"]))

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

    def summary(self):
        """Print spending summary."""
        balance = self.record_balance()
        daily = self.get_daily_spending()
        print(f"Current balance: ${balance:.2f}")
        print(f"Spent today: ${daily:.2f}")
        if daily > 0:
            print(f"Daily rate: ${daily:.2f}/day")
            print(f"Days remaining: {balance / daily:.1f}")


# Usage
tracker = SpendingTracker("YOUR_API_KEY")
tracker.summary()

Eine CSV-Datei trägt erstaunlich weit: Sie lässt sich in jeder Tabellenkalkulation öffnen, per Cronjob fortschreiben und später in eine Zeitreihendatenbank überführen. Aufschlussreich sind die Ausreißer – ein Tag mit doppeltem Verbrauch bedeutet erfahrungsgemäß nicht doppelten Traffic, sondern eine Wiederholungslogik, die dieselbe Seite immer wieder neu löst.

Ein Eintrag alle 15 Minuten reicht: Häufigere Messungen erhöhen die Auflösung kaum, weil einzelne Lösungen den Stand nur in kleinen Schritten verändern.

Guthabenprüfung in den Lösungs-Workflow einbauen

Am Ende soll die Prüfung kein separates Skript bleiben, sondern im Solver selbst sitzen. Der folgende Wrapper prüft nach Zeit oder nach Anzahl – alle 5 Minuten oder alle 50 Lösungen – und bricht mit einer klaren Ausnahme ab, statt in eine Fehlerschleife zu laufen:

import requests
import time


class BalanceAwareSolver:
    """Solver that checks balance before solving."""

    def __init__(self, api_key, min_balance=1.0):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.min_balance = min_balance
        self.last_balance_check = 0
        self.cached_balance = None
        self.solves_since_check = 0

    def solve(self, method, **params):
        """Solve with balance pre-check."""
        # Check balance every 50 solves or every 5 minutes
        if self._should_check_balance():
            balance = self._get_balance()
            if balance < self.min_balance:
                raise RuntimeError(
                    f"Balance too low: ${balance:.2f} "
                    f"(minimum: ${self.min_balance:.2f})"
                )

        return self._do_solve(method, **params)

    def _should_check_balance(self):
        elapsed = time.time() - self.last_balance_check
        return elapsed > 300 or self.solves_since_check >= 50

    def _get_balance(self):
        resp = requests.get(f"{self.base_url}/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        })
        self.cached_balance = float(resp.json()["request"])
        self.last_balance_check = time.time()
        self.solves_since_check = 0
        return self.cached_balance

    def _do_solve(self, method, **params):
        data = {"key": self.api_key, "method": method, "json": 1}
        data.update(params)
        resp = requests.post(f"{self.base_url}/in.php", data=data)
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            })
            data = result.json()
            if data["request"] != "CAPCHA_NOT_READY":
                self.solves_since_check += 1
                return data["request"]

        raise TimeoutError("Solve timeout")


# Usage
solver = BalanceAwareSolver("YOUR_API_KEY", min_balance=2.0)

try:
    token = solver.solve("userrecaptcha", googlekey="KEY", pageurl="https://example.com")
except RuntimeError as e:
    print(f"Balance issue: {e}")

_should_check_balance() ist der Kern: Eine Prüfung vor jeder einzelnen Lösung verdoppelt die Anfragen ohne Erkenntnisgewinn, gar keine Prüfung führt zur Fehlerserie.

Wichtig ist, was Ihre Pipeline mit dem RuntimeError macht. Der Lauf sollte die Warteschlange erhalten: Tasks, die wegen eines leeren Kontos nie abgeschickt wurden, gehören zurück in die Queue, damit die Verarbeitung nach dem Aufladen dort weitermacht.

Fehlerbehebung

Häufige Fragen

Kann CaptchaAI mein Guthaben automatisch aufladen?

Nein – die API kennt keine Zahlungsaktion. Automatisieren lässt sich die Warnung, nicht die Zahlung: Der Monitor meldet die Unterschreitung, das Aufladen erfolgt im Konto oder über Ihren Beschaffungsprozess. Planen Sie den Puffer entsprechend großzügig.

Was genau liefert getbalance zurück?

Einen einzelnen Zahlenwert – mit json=1 als Zeichenkette im Feld request, sonst als reinen Text. Die API-Dokumentation beschreibt ihn als Gesamtzahl der Plan-Threads (Beispielwert 600), die Codebeispiele hier formatieren ihn wie 2Captcha-kompatible Clients als Dollarbetrag. Gleichen Sie den Rohwert einmal mit Ihrem Konto ab.

Was passiert, wenn das Guthaben mitten im Lauf aufgebraucht ist?

in.php nimmt keine neuen Tasks mehr an und antwortet mit ERROR_ZERO_BALANCE; bereits laufende Abfragen an res.php liefern ihr Ergebnis noch. Behandeln Sie den Fehler als Signal zum Anhalten der Warteschlange, nicht als Grund, Tasks zu verwerfen.

Wie überwache ich mehrere API-Schlüssel gleichzeitig?

Legen Sie pro Schlüssel eine Monitor-Instanz an und geben Sie jeder Meldung eine eindeutige Kennung – Projektname oder Umgebung. Sonst sieht ein Team mit getrennten Konten für Staging und Produktion nur eine Warnung ohne Zuordnung.

Weiterführende Artikel

• Antwortformate und Fehlercodes der API im Überblick
• Bereitschaftsalarme über PagerDuty verteilen

Legen Sie den Schwellenwert fest, bevor der nächste Nachtlauf startet – Guthaben mit CaptchaAI im Blick behalten.