Batch-Bild-CAPTCHA-Lösung: Verarbeitung von mehr als 1000 Bildern

Ein Stapel mit 1.000 Bild-CAPTCHAs scheitert selten an der API, sondern an der Schleife davor. Bild-CAPTCHAs gehören zu den schnellsten Typen im CaptchaAI-Katalog – die Lösungszeit liegt unter 0,5 Sekunden. Nacheinander abgearbeitet, mit je fünf Sekunden Wartezeit pro Ergebnis, dauert derselbe Stapel über eine Stunde. Zwei getrennte Phasen – erst alle Bilder einreichen, dann alle Task-IDs parallel abfragen – drücken denselben Lauf rechnerisch auf wenige Minuten. Dieser Leitfaden zeigt die Pipeline in Python und Node.js, samt Rate-Limiting, Fortschrittsanzeige und CSV-Report.

Warum die sequenzielle Schleife bei 1.000 Bild-CAPTCHAs kippt

Rechnen Sie den naiven Ablauf einmal durch. Pro Datei fallen drei Wartezeiten an: das Übertragen der Base64-Payload, das Intervall bis zur ersten Statusabfrage und die Antwort selbst. Das Abfrageintervall dominiert alles andere – bei POLL_INTERVAL = 5 sind das fünf Sekunden pro Bild, unabhängig davon, wie schnell die Lösung vorliegt. 1.000 Dateien mal fünf Sekunden ergeben rund 83 Minuten reine Wartezeit, in denen genau ein CAPTCHA in Bearbeitung ist.

Der eigentliche Engpass ist damit nicht das Intervall selbst, sondern die Kopplung von Einreichung und Abfrage im selben Schleifendurchlauf – jede Wartezeit blockiert die übrigen 999 Dateien.

Pipeline-Architektur: einreichen, dann abfragen

Die Zielarchitektur hat vier Stufen: eine Warteschlange mit den Dateipfaden, Submit-Worker gegen in.php, Poll-Worker gegen res.php und einen Ergebnisspeicher, der CSV oder JSON schreibt.

[Image Queue] → [Submit Workers] → [Poll Workers] → [Results Store]
     ↓                ↓                  ↓                ↓
  1000 images    20 concurrent      Adaptive poll     CSV/JSON output
                   submits           intervals

Beide Phasen werden getrennt begrenzt: Einreichungen sind kurz und rechenlastig (Base64-Encoding), Abfragen sind lang und warten fast nur. Im Python-Beispiel stehen deshalb zwei Semaphoren mit unterschiedlichen Werten.

Threads dimensionieren: wie viel Parallelität Ihr Plan zulässt

CaptchaAI rechnet Thread-basiert ab – Sie zahlen pro gleichzeitig laufendem CAPTCHA, nicht pro gelöster Aufgabe. Ein Thread ist ein CAPTCHA in Bearbeitung; ist es fertig, nimmt derselbe Thread das nächste. Die Zahl der Lösungen im Abrechnungsmonat ist nicht gedeckelt: Ein Lauf mit 1.000 Bildern kostet nicht mehr als einer mit 100, er belegt die Threads nur länger. MAX_CONCURRENT_SUBMITS = 20 ergibt deshalb nur Sinn, wenn genügend Threads bereitstehen.

Darüber liegen ENTERPRISE sowie VIP-1 bis VIP-3; alle Preise in US-Dollar. Setzen Sie die Parallelität nie höher als Ihre Thread-Zahl.

Python: asynchroner Stapelprozessor mit asyncio und aiohttp

Das Skript arbeitet strikt zweiphasig. submit_image() kodiert jede Datei nach Base64 und schickt sie mit method=base64 an in.php; zurück kommt eine Task-ID. poll_result() fragt res.php ab und behandelt CAPCHA_NOT_READY als Zwischenstand – jede andere Antwort beendet die Aufgabe, erfolgreich oder mit Fehler. Zwei Semaphoren deckeln die Phasen getrennt, asyncio.gather() sammelt ein, der Report landet als CSV auf der Platte.

import asyncio
import aiohttp
import base64
import json
import time
import csv
from pathlib import Path

API_KEY = "YOUR_API_KEY"
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
MAX_CONCURRENT_SUBMITS = 20
MAX_CONCURRENT_POLLS = 30
POLL_INTERVAL = 5


async def submit_image(session, sem, image_path):
    """Submit a single image CAPTCHA."""
    async with sem:
        with open(image_path, "rb") as f:
            img_b64 = base64.b64encode(f.read()).decode()

        data = {
            "key": API_KEY,
            "method": "base64",
            "body": img_b64,
            "json": "1",
        }

        async with session.post(SUBMIT_URL, data=data) as resp:
            result = await resp.json()

        if result["status"] != 1:
            return {"file": str(image_path), "error": result["request"]}

        return {
            "file": str(image_path),
            "task_id": result["request"],
            "submitted_at": time.time(),
        }


async def poll_result(session, sem, task):
    """Poll for a single task result."""
    async with sem:
        for attempt in range(24):
            await asyncio.sleep(POLL_INTERVAL)

            params = {
                "key": API_KEY,
                "action": "get",
                "id": task["task_id"],
                "json": "1",
            }

            async with session.get(RESULT_URL, params=params) as resp:
                result = await resp.json()

            if result["status"] == 1:
                return {
                    "file": task["file"],
                    "task_id": task["task_id"],
                    "answer": result["request"],
                    "solve_time": time.time() - task["submitted_at"],
                }
            if result["request"] != "CAPCHA_NOT_READY":
                return {
                    "file": task["file"],
                    "task_id": task["task_id"],
                    "error": result["request"],
                }

        return {
            "file": task["file"],
            "task_id": task["task_id"],
            "error": "TIMEOUT",
        }


async def process_batch(image_dir, output_file="results.csv"):
    """Process all images in a directory."""
    image_paths = sorted(Path(image_dir).glob("*.png")) + \
                  sorted(Path(image_dir).glob("*.jpg"))

    print(f"Found {len(image_paths)} images")

    submit_sem = asyncio.Semaphore(MAX_CONCURRENT_SUBMITS)
    poll_sem = asyncio.Semaphore(MAX_CONCURRENT_POLLS)

    async with aiohttp.ClientSession() as session:
        # Phase 1: Submit all images
        print("Submitting...")
        submit_tasks = [
            submit_image(session, submit_sem, path)
            for path in image_paths
        ]
        submissions = await asyncio.gather(*submit_tasks)

        # Separate successes and errors
        pending = [s for s in submissions if "task_id" in s]
        errors = [s for s in submissions if "error" in s]
        print(f"Submitted: {len(pending)}, Errors: {len(errors)}")

        # Phase 2: Poll all pending tasks
        print("Polling for results...")
        poll_tasks = [
            poll_result(session, poll_sem, task)
            for task in pending
        ]
        results = await asyncio.gather(*poll_tasks)

    # Combine results
    all_results = results + errors

    # Write to CSV
    with open(output_file, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=[
            "file", "task_id", "answer", "solve_time", "error"
        ])
        writer.writeheader()
        for r in all_results:
            writer.writerow({
                "file": r.get("file", ""),
                "task_id": r.get("task_id", ""),
                "answer": r.get("answer", ""),
                "solve_time": round(r.get("solve_time", 0), 2),
                "error": r.get("error", ""),
            })

    solved = sum(1 for r in results if "answer" in r)
    failed = sum(1 for r in results if "error" in r)
    print(f"Done: {solved} solved, {failed} failed, {len(errors)} submit errors")
    print(f"Results saved to {output_file}")


# Run
asyncio.run(process_batch("./captcha_images"))

Erwartete Ausgabe:

Found 1000 images
Submitting...
Submitted: 997, Errors: 3
Polling for results...
Done: 985 solved, 12 failed, 3 submit errors
Results saved to results.csv

Gelöste Aufgaben, fehlgeschlagene Abfragen und Einreichungsfehler werden getrennt gezählt. Einreichungsfehler deuten meist auf defekte oder zu große Bilder hin, nicht auf die API. Liegt der Median in solve_time deutlich unter fünf Sekunden, senken Sie POLL_INTERVAL.

Node.js: derselbe Ablauf als Worker-Pool

Die JavaScript-Variante verarbeitet statt zweier globaler Phasen Blöcke von 20 Dateien und wartet mit Promise.all(), bis der Block komplett ist. Das liest sich einfacher und liefert sofort Feedback pro Datei, kostet aber Durchsatz – der langsamste Eintrag hält die übrigen 19 auf. Für Stapel bis wenige Tausend Bilder ist das ein guter Kompromiss.

const axios = require('axios');
const fs = require('fs');
const path = require('path');
const { createObjectCsvWriter } = require('csv-writer');

const API_KEY = 'YOUR_API_KEY';
const SUBMIT_URL = 'https://ocr.captchaai.com/in.php';
const RESULT_URL = 'https://ocr.captchaai.com/res.php';
const MAX_CONCURRENT = 20;
const POLL_INTERVAL_MS = 5000;

class BatchProcessor {
  constructor(concurrency = MAX_CONCURRENT) {
    this.concurrency = concurrency;
    this.results = [];
    this.processed = 0;
    this.total = 0;
  }

  async submitImage(imagePath) {
    const imgBase64 = fs.readFileSync(imagePath, { encoding: 'base64' });
    const resp = await axios.post(SUBMIT_URL, null, {
      params: {
        key: API_KEY,
        method: 'base64',
        body: imgBase64,
        json: 1,
      },
    });

    if (resp.data.status !== 1) {
      throw new Error(resp.data.request);
    }
    return resp.data.request;
  }

  async pollResult(taskId) {
    for (let i = 0; i < 24; i++) {
      await new Promise(r => setTimeout(r, POLL_INTERVAL_MS));
      const resp = await axios.get(RESULT_URL, {
        params: { key: API_KEY, action: 'get', id: taskId, json: 1 },
      });

      if (resp.data.status === 1) return resp.data.request;
      if (resp.data.request !== 'CAPCHA_NOT_READY') {
        throw new Error(resp.data.request);
      }
    }
    throw new Error('TIMEOUT');
  }

  async processOne(imagePath) {
    const startTime = Date.now();
    try {
      const taskId = await this.submitImage(imagePath);
      const answer = await this.pollResult(taskId);
      this.processed++;
      const elapsed = ((Date.now() - startTime) / 1000).toFixed(1);
      console.log(`[${this.processed}/${this.total}] ${path.basename(imagePath)}: ${answer} (${elapsed}s)`);
      return { file: imagePath, answer, solveTime: elapsed, error: '' };
    } catch (err) {
      this.processed++;
      return { file: imagePath, answer: '', solveTime: 0, error: err.message };
    }
  }

  async run(imageDir, outputFile = 'results.csv') {
    const files = fs.readdirSync(imageDir)
      .filter(f => /\.(png|jpg|jpeg|gif)$/i.test(f))
      .map(f => path.join(imageDir, f));

    this.total = files.length;
    console.log(`Processing ${this.total} images with ${this.concurrency} workers`);

    // Process in chunks
    for (let i = 0; i < files.length; i += this.concurrency) {
      const chunk = files.slice(i, i + this.concurrency);
      const chunkResults = await Promise.all(
        chunk.map(f => this.processOne(f))
      );
      this.results.push(...chunkResults);
    }

    // Write CSV
    const csvWriter = createObjectCsvWriter({
      path: outputFile,
      header: [
        { id: 'file', title: 'File' },
        { id: 'answer', title: 'Answer' },
        { id: 'solveTime', title: 'Solve Time (s)' },
        { id: 'error', title: 'Error' },
      ],
    });
    await csvWriter.writeRecords(this.results);

    const solved = this.results.filter(r => r.answer).length;
    console.log(`Done: ${solved}/${this.total} solved. Results: ${outputFile}`);
  }
}

const processor = new BatchProcessor(20);
processor.run('./captcha_images');

Rate-Limiting: 429-Antworten vermeiden statt abfangen

Eine Semaphore begrenzt, wie viele Anfragen gleichzeitig offen sind – nicht, wie viele pro Sekunde starten. Genau daran scheitern viele Stapelskripte: Beim Start feuern alle Coroutinen zugleich los und erzeugen eine Spitze, die je nach Auslastung mit 429-Antworten quittiert wird. Ein Rate-Limiter mit gleitendem Ein-Sekunden-Fenster glättet den Anlauf.

class RateLimiter:
    def __init__(self, max_per_second=10):
        self.max_per_second = max_per_second
        self.timestamps = []

    async def acquire(self):
        now = time.time()
        self.timestamps = [t for t in self.timestamps if now - t < 1.0]

        if len(self.timestamps) >= self.max_per_second:
            wait = 1.0 - (now - self.timestamps[0])
            if wait > 0:
                await asyncio.sleep(wait)

        self.timestamps.append(time.time())

# Use in submit loop
rate_limiter = RateLimiter(max_per_second=10)

async def submit_with_rate_limit(session, image_path):
    await rate_limiter.acquire()
    # ... submit as before

Zehn Einreichungen pro Sekunde sind ein konservativer Startwert; tasten Sie sich anhand der Fehlerquote nach oben.

Fortschritt, Durchsatz und ETA sichtbar machen

Ein Lauf über 1.000 Bilder ohne Ausgabe fühlt sich an wie ein Absturz. Der folgende Tracker schreibt eine einzeilige Statusanzeige mit erledigten Aufgaben, Erfolgen, Fehlern, Durchsatz und Restzeit – im Terminal wie im CI-Log, denn die Rate verrät sofort, ob die Parallelität greift.

import sys

class ProgressTracker:
    def __init__(self, total):
        self.total = total
        self.completed = 0
        self.solved = 0
        self.failed = 0
        self.start_time = time.time()

    def update(self, success=True):
        self.completed += 1
        if success:
            self.solved += 1
        else:
            self.failed += 1

        elapsed = time.time() - self.start_time
        rate = self.completed / elapsed if elapsed > 0 else 0
        eta = (self.total - self.completed) / rate if rate > 0 else 0

        sys.stdout.write(
            f"\r[{self.completed}/{self.total}] "
            f"Solved: {self.solved} | Failed: {self.failed} | "
            f"Rate: {rate:.1f}/s | ETA: {eta:.0f}s"
        )
        sys.stdout.flush()

Praxisbeispiel: nächtlicher Lauf auf einem eigenen VPS

Ein typisches Setup aus dem DACH-Raum: Ein Automatisierungsteam legt die Bilder tagsüber in einem Verzeichnis ab und lässt den Stapel nachts auf einem kleinen Hetzner- oder netcup-VPS laufen, angestoßen von einer geplanten GitLab-CI-Pipeline. Der API-Schlüssel liegt als maskierte CI-Variable vor, results.csv wird als Artefakt abgelegt, und ein Nachlauf-Schritt lässt den Job fehlschlagen, sobald die Fehlerquote einen Schwellwert überschreitet. Morgens verrät der Job-Status, ob der Lauf sauber war.

Zwei Punkte tauchen in deutschsprachigen Teams regelmäßig auf: Legen Sie fest, wie lange results.csv als Artefakt vorgehalten wird – enthalten Bilder oder gelöste Texte personenbezogene Daten, brauchen Sie dafür eine DSGVO-konforme Grundlage. Dokumentieren Sie zudem Herkunft und Rechtsgrundlage der Bilder; diese Sorgfaltspflicht liegt beim Betreiber der Pipeline.

Fehlerbehebung

Häufige Fragen

Wie lange dauert ein Lauf mit 1.000 Bildern?

Bei 20 gleichzeitigen Aufgaben und fünf Sekunden Abfrageintervall liegt die rechnerische Laufzeit im einstelligen Minutenbereich. Nicht die Lösungszeit bremst den Lauf: Für Bild-CAPTCHAs nennt CaptchaAI unter 0,5 Sekunden. Den Rest bestimmen Bildgröße, Netzwerklatenz und Ihr Abfrageintervall.

Was passiert mit Bildern, die fehlschlagen?

Sie landen mit Fehlercode in derselben CSV-Datei und halten den Lauf nicht auf. Trennen Sie Einreichungsfehler – die Aufgabe wurde nie angelegt – von Abfragefehlern, bei denen die Aufgabe existiert, aber kein Ergebnis liefert. Nur die erste Gruppe lässt sich unverändert erneut einreichen.

Lässt sich ein abgebrochener Stapel fortsetzen?

Ja, sofern Sie die Zuordnung von Datei zu Task-ID direkt nach Phase 1 persistieren, etwa als JSON-Lines-Datei. Beim Neustart überspringen Sie alles mit vorhandener Antwort und fragen nur die offenen IDs erneut ab. Ohne diese Datei bleibt nur, den kompletten Stapel neu einzureichen.

Kann ich verschiedene CAPTCHA-Typen im selben Stapel mischen?

Technisch ja: Einreichen, Task-ID merken, Ergebnis abfragen funktioniert für alle Typen gleich, nur der method-Parameter und die Pflichtfelder unterscheiden sich. Neben Bild- und Rasterbild-CAPTCHAs deckt CaptchaAI unter anderem reCAPTCHA v2 und v3, Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3 und BLS ab; CaptchaFox, Friendly Captcha und Lemin laufen in der Beta. In der Praxis empfiehlt sich je Typ eine eigene Warteschlange, weil die Lösungszeiten weit auseinanderliegen.

Nächster Schritt: den ersten Stapel testen

Legen Sie ein Verzeichnis mit 20 Testbildern an, setzen Sie MAX_CONCURRENT_SUBMITS auf 5 und lassen Sie das Python-Skript einmal durchlaufen. Stimmen CSV-Ausgabe und Fehlerquote, skalieren Sie auf die Thread-Zahl Ihres Plans. Ihren API-Schlüssel erhalten Sie unter captchaai.com.

Verwandte Leitfäden

• Asynchron lösen mit Python und asyncio
• Parallel oder sequenziell: Performance im Vergleich
• Rate-Limits und Drosselung im Griff behalten