Einen separaten Batch-Endpunkt gibt es bei CaptchaAI nicht: in.php nimmt genau eine Aufgabe pro Aufruf entgegen, parallelisiert wird auf Ihrer Seite. Genau daran entscheidet sich, ob ein Lauf Minuten oder Stunden dauert. Wer 200 Seiten crawlt, jede CAPTCHA-Abfrage einzeln einreicht und abwartet, addiert 200 Lösungszeiten hintereinander. Wer alle Aufgaben zuerst einreicht und die Ergebnisse anschließend gemeinsam abfragt, wartet im Wesentlichen nur einmal.
Dieser Artikel zeigt drei Muster dafür – Thread-Pool, asyncio und die strikte Trennung von Einreichung und Abfrage – und beantwortet die Frage, die in der Praxis vorher kommt: Wie viele Aufgaben dürfen überhaupt gleichzeitig laufen?
Wie viele Aufgaben parallel laufen dürfen: die Thread-Frage
CaptchaAI rechnet pro gleichzeitigem Thread ab, nicht pro Lösung. Ein Thread ist eine Aufgabe, die gerade in Bearbeitung ist; sobald sie fertig ist, nimmt derselbe Thread die nächste. Die Zahl der Lösungen innerhalb des Abrechnungsmonats ist je Thread unbegrenzt.
Für die Batch-Planung heißt das: Die gebuchte Thread-Zahl ist die Obergrenze Ihrer sinnvollen Parallelität – und kein Budget, das sich je Aufgabe verbraucht.
- BASIC (15 $/Monat, 5 Threads) – Entwicklung, Tests, kleine Nachtläufe
- ADVANCE (90 $/Monat, 50 Threads) – regelmäßiges Web-Scraping im Produktivbetrieb
- ENTERPRISE (300 $/Monat, 200 Threads) – verteilte Pipelines mit mehreren Workern
Preise in US-Dollar. Mehr max_workers zu setzen, als Threads gebucht sind, beschleunigt nichts: Die zusätzlichen Aufgaben warten dann lediglich an einer anderen Stelle.
Sequenziell oder parallel: der Zeitunterschied als Skizze
Der Unterschied ist kein Optimierungsdetail, sondern entscheidet darüber, ob ein Lauf noch in ein Wartungsfenster passt:
Sequential (slow):
Submit #1 → Poll → Result (15s)
Submit #2 → Poll → Result (15s)
Submit #3 → Poll → Result (15s)
Total: ~45s for 3 solves
Parallel (fast):
Submit #1 ─┐
Submit #2 ─┤→ Poll all → Results arrive
Submit #3 ─┘
Total: ~15s for 3 solves
Bei drei Aufgaben wirkt das unspektakulär. Bei 2.000 Cloudflare-Turnstile-Abfragen, die einzeln in unter 10 Sekunden gelöst sind, stehen gut fünf Stunden sequenziell gegen wenige Minuten mit 50 gleichzeitigen Aufgaben.
Variante 1: Thread-Pool für Batches bis rund 50 Aufgaben
Für die meisten Scraping-Jobs ist ThreadPoolExecutor die richtige Wahl: wenig Code, gut zu debuggen, und die Wartezeit auf die HTTP-Antwort ist ohnehin I/O-gebunden. Jede Aufgabe durchläuft ihren eigenen Zyklus aus Einreichen und Abfragen, as_completed liefert die Ergebnisse in der Reihenfolge, in der sie fertig werden.
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"
def submit_task(method, **params):
"""Submit a single CAPTCHA task."""
data = {"key": API_KEY, "method": method, "json": 1}
data.update(params)
resp = requests.post(f"{BASE_URL}/in.php", data=data, timeout=30)
result = resp.json()
if result.get("status") != 1:
raise RuntimeError(f"Submit error: {result.get('request')}")
return result["request"]
def poll_result(task_id, timeout=120):
"""Poll until result is ready."""
start = time.time()
while time.time() - start < timeout:
time.sleep(5)
resp = requests.get(f"{BASE_URL}/res.php", params={
"key": API_KEY, "action": "get",
"id": task_id, "json": 1,
}, timeout=15)
data = resp.json()
if data["request"] != "CAPCHA_NOT_READY":
return data["request"]
raise TimeoutError(f"Task {task_id} timeout")
def solve_one(sitekey, pageurl):
"""Submit and poll a single task."""
task_id = submit_task("userrecaptcha", googlekey=sitekey, pageurl=pageurl)
token = poll_result(task_id)
return {"url": pageurl, "token": token}
def batch_solve(tasks, max_workers=10):
"""Solve multiple CAPTCHAs in parallel."""
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(solve_one, t["sitekey"], t["url"]): t
for t in tasks
}
for future in as_completed(futures):
task = futures[future]
try:
result = future.result()
results.append(result)
print(f"Solved: {result['url']}")
except Exception as e:
print(f"Failed: {task['url']} - {e}")
results.append({"url": task["url"], "token": None, "error": str(e)})
return results
# Usage
tasks = [
{"sitekey": "SITE_KEY_1", "url": "https://example.com/page1"},
{"sitekey": "SITE_KEY_2", "url": "https://example.com/page2"},
{"sitekey": "SITE_KEY_3", "url": "https://example.com/page3"},
]
results = batch_solve(tasks, max_workers=5)
print(f"Solved {sum(1 for r in results if r.get('token'))}/{len(tasks)}")
Drei Details lohnen den zweiten Blick:
max_workersist Ihre tatsächliche Parallelität – stimmen Sie den Wert auf die gebuchten Threads ab.- Fehlgeschlagene Aufgaben landen mit
token: Noneim Ergebnis, statt den gesamten Lauf abzubrechen. - Jedes Ergebnis trägt seine URL mit; damit bleibt die Zuordnung eindeutig, auch wenn die Reihenfolge wechselt.
Variante 2: asyncio, wenn es Hunderte gleichzeitig werden
Oberhalb von etwa 100 gleichzeitigen Aufgaben wird ein Thread je Aufgabe teuer – jeder Thread belegt Speicher und erzwingt Kontextwechsel, obwohl er die meiste Zeit nur wartet. asyncio mit aiohttp erledigt dieselbe Arbeit in einem einzigen Prozess ohne zusätzliche Betriebssystem-Threads, und asyncio.Semaphore begrenzt die Parallelität exakt auf den Wert, den Ihr Thread-Kontingent hergibt.
import asyncio
import aiohttp
import time
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"
async def submit_task_async(session, method, **params):
data = {"key": API_KEY, "method": method, "json": 1}
data.update(params)
async with session.post(f"{BASE_URL}/in.php", data=data) as resp:
result = await resp.json()
if result.get("status") != 1:
raise RuntimeError(f"Submit error: {result.get('request')}")
return result["request"]
async def poll_result_async(session, task_id, timeout=120):
start = time.time()
while time.time() - start < timeout:
await asyncio.sleep(5)
params = {
"key": API_KEY, "action": "get",
"id": task_id, "json": 1,
}
async with session.get(f"{BASE_URL}/res.php", params=params) as resp:
data = await resp.json()
if data["request"] != "CAPCHA_NOT_READY":
return data["request"]
raise TimeoutError(f"Task {task_id} timeout")
async def solve_one_async(session, sitekey, pageurl):
task_id = await submit_task_async(
session, "userrecaptcha",
googlekey=sitekey, pageurl=pageurl,
)
token = await poll_result_async(session, task_id)
return {"url": pageurl, "token": token}
async def batch_solve_async(tasks, max_concurrent=20):
"""Solve many CAPTCHAs concurrently with asyncio."""
semaphore = asyncio.Semaphore(max_concurrent)
results = []
async def solve_with_limit(task):
async with semaphore:
try:
result = await solve_one_async(
session, task["sitekey"], task["url"],
)
return result
except Exception as e:
return {"url": task["url"], "token": None, "error": str(e)}
async with aiohttp.ClientSession() as session:
coros = [solve_with_limit(t) for t in tasks]
results = await asyncio.gather(*coros)
return results
# Usage
tasks = [
{"sitekey": "KEY", "url": f"https://example.com/page{i}"}
for i in range(50)
]
results = asyncio.run(batch_solve_async(tasks, max_concurrent=20))
solved = sum(1 for r in results if r.get("token"))
print(f"Solved: {solved}/{len(tasks)}")
Der Semaphore-Wert ist dabei die eigentliche Stellschraube: Er hält die Zahl der offenen Anfragen konstant, während asyncio.gather die Ergebnisse einsammelt.
Variante 3: erst alles einreichen, dann gesammelt abfragen
Für maximalen Durchsatz trennen Sie die beiden Phasen. Zuerst gehen alle Aufgaben hinaus – mit rund 100 ms Abstand, damit die API nicht in einem Schlag getroffen wird. Danach fragt eine einzige Schleife sämtliche offenen Task-IDs ab und nimmt jedes Ergebnis heraus, sobald es vorliegt. Das spart je Aufgabe eine komplette Wartezeit, weil die ersten Lösungen bereits laufen, während Sie noch einreichen.
import requests
import time
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"
def batch_submit(tasks):
"""Submit all tasks first, return task IDs."""
submitted = []
for task in tasks:
try:
data = {
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": task["sitekey"],
"pageurl": task["url"],
"json": 1,
}
resp = requests.post(f"{BASE_URL}/in.php", data=data, timeout=30)
result = resp.json()
if result.get("status") == 1:
submitted.append({
"task_id": result["request"],
"url": task["url"],
})
time.sleep(0.1) # Brief delay between submits
except Exception as e:
print(f"Submit failed for {task['url']}: {e}")
return submitted
def batch_poll(submitted, timeout=120):
"""Poll all submitted tasks until complete."""
pending = {s["task_id"]: s for s in submitted}
results = []
start = time.time()
while pending and time.time() - start < timeout:
time.sleep(5)
for task_id in list(pending.keys()):
try:
resp = requests.get(f"{BASE_URL}/res.php", params={
"key": API_KEY, "action": "get",
"id": task_id, "json": 1,
}, timeout=15)
data = resp.json()
if data["request"] != "CAPCHA_NOT_READY":
info = pending.pop(task_id)
results.append({
"url": info["url"],
"token": data["request"],
})
except Exception:
pass
# Mark remaining as failed
for task_id, info in pending.items():
results.append({"url": info["url"], "token": None, "error": "timeout"})
return results
# Usage
tasks = [
{"sitekey": "KEY", "url": f"https://example.com/page{i}"}
for i in range(20)
]
submitted = batch_submit(tasks)
print(f"Submitted {len(submitted)} tasks")
results = batch_poll(submitted)
solved = sum(1 for r in results if r.get("token"))
print(f"Solved: {solved}/{len(tasks)}")
Entscheidend ist der Umgang mit Nachzüglern: Was nach dem Timeout noch in pending steht, wird als Fehlschlag markiert und geht in den nächsten Lauf – nicht in eine Endlosschleife.
Praxisbeispiel: nächtlicher Katalogabgleich im DACH-Raum
Ein Team betreibt einen Shopware-Shop und gleicht jede Nacht rund 800 Produktseiten autorisierter Lieferantenportale ab. Der Job läuft als GitLab-CI-Pipeline auf einem Hetzner-Cloud-Server; etwa jede vierte Seite zeigt eine reCAPTCHA-v2-Abfrage, also rund 200 Lösungen pro Nacht.
Sequenziell wären das bei bis zu 60 Sekunden je Lösung mehrere Stunden. Mit 50 gleichzeitigen Aufgaben und dem Submit-then-Poll-Muster passt derselbe Lauf in ein Wartungsfenster von etwa einer Stunde – an den Kosten ändert sich dabei nichts, weil Threads und nicht einzelne Lösungen abgerechnet werden.
Zwei Punkte, die in Projekten im deutschsprachigen Raum regelmäßig aufkommen: Klären Sie vorab, ob Sie die Zielseiten abrufen dürfen – eigene Systeme, Vertragspartner oder öffentlich zugängliche Daten –, und behandeln Sie IP-Adressen in Ihren Logs als personenbezogene Daten im Sinne der DSGVO. Ein Scraping-Log ist davon nicht ausgenommen.
Durchsatz realistisch planen
| Gleichzeitige Aufgaben | Ungefährer Durchsatz | Typischer Einsatz |
|---|---|---|
| 1–5 | 3–5 Lösungen/min | Entwicklung, einzelne Testläufe |
| 5–20 | 15–60 Lösungen/min | laufendes Web-Scraping |
| 20–50 | 60–150 Lösungen/min | Pipelines mit hohem Volumen |
| 50–100 | 150–300 Lösungen/min | verteilte Worker, Unternehmensmaßstab |
Die Werte sind Richtwerte für gemischte Aufgaben; die tatsächliche Rate hängt stark vom CAPTCHA-Typ ab. Ein Bild-CAPTCHA ist in unter 0,5 Sekunden gelöst, Cloudflare Turnstile in unter 10 Sekunden, reCAPTCHA v2 in unter 60 Sekunden – ein Batch ist immer so schnell wie sein langsamster Typ.
Was in Batches typischerweise schiefgeht
| Symptom | Ursache | Gegenmaßnahme |
|---|---|---|
| HTTP 429 beim Einreichen | zu viele Anfragen pro Sekunde | 100 ms Pause zwischen den Übermittlungen einbauen |
| viele Timeouts | Abfrage-Timeout zu kurz gewählt | auf 120–180 Sekunden erhöhen |
| kaum Zuwachs oberhalb von 50 gleichzeitigen Aufgaben | Netzwerk- oder Thread-Engpass | auf asyncio (aiohttp) wechseln, Thread-Kontingent prüfen |
| Ergebnisse den falschen Seiten zugeordnet | Task-ID nicht mitgeführt | Ergebnisse in einem Dict mit task_id als Schlüssel halten |
| Token beim Absenden abgelehnt | Ergebnis zu lange liegen geblieben | Formular direkt nach Erhalt des Tokens absenden |
Wie Sie Wiederholungen sauber staffeln, zeigt der Leitfaden zur Wiederholungslogik und Fehlerbehandlung in Node.js; die Grenzwerte selbst behandelt der Beitrag zu Rate-Limits und Anfragedrosselung.
Häufige Fragen
Brauche ich für Batch-Verarbeitung einen anderen Endpunkt?
Nein. Sie verwenden weiterhin in.php zum Einreichen und res.php zum Abfragen, nur eben mehrfach gleichzeitig. „Batch“ beschreibt Ihr Client-Muster, nicht eine eigene API.
Kann ich verschiedene CAPTCHA-Typen in einem Batch mischen?
Ja. Jede Aufgabe trägt ihre eigene method: userrecaptcha für reCAPTCHA v2 und v3, turnstile für Cloudflare Turnstile, geetest für GeeTest v3, post für Bild- und Rasterbild-CAPTCHAs. Planen Sie nur ein, dass die Typen unterschiedlich lange brauchen.
Wie vermeide ich 429-Antworten bei hoher Parallelität?
Verteilen Sie die Einreichungen, nicht die Aufgaben. Eine Pause von rund 100 ms zwischen den POST-Aufrufen genügt in der Regel; ausschlaggebend ist die Rate der Übermittlungen, nicht die Zahl der offenen Aufgaben.
Was passiert mit Tokens, die im Batch zu lange liegen bleiben?
Sie verfallen. reCAPTCHA-Tokens sind rund zwei Minuten gültig – lösen Sie deshalb erst dann, wenn das Formular auch abgesendet werden kann, statt Ergebnisse auf Vorrat zu halten.
Wie viele Threads brauche ich für 1.000 Lösungen pro Stunde?
Rechnen Sie mit der Lösungszeit Ihres häufigsten Typs. Cloudflare Turnstile ist in unter 10 Sekunden gelöst; ein Thread schafft damit rund 360 Lösungen pro Stunde, für 1.000 genügen drei bis vier Threads plus etwas Reserve. Bei reCAPTCHA v2 mit bis zu 60 Sekunden je Lösung sind es eher 17 bis 20 Threads.
Planen Sie Ihren nächsten Lauf nach Threads statt nach Lösungen – CaptchaAI testen und den ersten Batch parallel einreichen.
