Captchas mit Python asyncio parallel lösen

Wenn Sie Captchas aus Python heraus parallel lösen möchten, verwenden Sie AsyncCapSkip mit asyncio.gather. Python ist das eine CapSkip-SDK, in dem der asynchrone Client eine echte Async-Implementierung und kein Alias ist. Ein Stapel von zehn reCAPTCHAs braucht daher etwa so lange wie das langsamste davon statt die Summe aller zehn. Dieser Leitfaden zeigt den lauffähigen Code, wie Sie die Nebenläufigkeit begrenzen, damit Sie den Solver nicht überfluten, und wie ein einzelner Fehlschlag nicht den ganzen Stapel mitreißt.

Warum ausgerechnet der Python-Client interessant ist

Alle vier SDKs exportieren etwas namens AsyncCapSkip. Nur bei einem davon ist es eine eigene Implementierung.

SDK	Was AsyncCapSkip ist	Wie Sie Arbeit nebenläufig ausführen
Python	Ein echter asynchroner Client	await asyncio.gather(...)
Node.js	Ein Alias für CapSkip	await Promise.all([...])
.NET	Ein Alias für CapSkipClient	await Task.WhenAll(...)
PHP	Nur ein Alias, aus Gründen der Quellcode-Parität	Synchron, keine Nebenläufigkeit

Node und .NET sind ohnehin nicht blockierend, dort kostet der Alias nichts. PHP ist synchron, und der Alias bringt dort überhaupt nichts. In Python macht der Unterschied sehr wohl etwas aus: Der einfache CapSkip Client blockiert die Event-Loop, während er pollt. Ihn in eine Coroutine zu stecken liefert Ihnen also Nebenläufigkeit auf dem Papier und sequenzielle Laufzeiten in der Praxis.

Was Sie brauchen

• Python 3.10 oder neuer
• pip install capskip
• CapSkip läuft mit aktiviertem API-Server und lauscht auf 127.0.0.1:8080
• Eine Liste von Sitekeys und Seiten-URLs, die abgearbeitet werden sollen

Nichts verlässt Ihren Rechner, es gibt also kein Rate-Limit auszuhandeln und keinen Zähler für Kosten pro Lösung, der beim Experimentieren mitläuft. Die vollständigen Methodensignaturen für jeden Typ finden Sie auf der Seite SDK zum Captcha-Lösen .

Die sequenzielle Variante und was sie kostet

So fangen die meisten an. Der Code ist korrekt – und langsam.

# pip install capskip
from capskip import CapSkip

solver = CapSkip(host="127.0.0.1", port=8080)

targets = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

# Each call blocks until that one CAPTCHA comes back.
for sitekey, url in targets:
    result = solver.recaptcha(sitekey=sitekey, url=url)
    print(result["code"][:40])   # the token, truncated for the log

Eine reCAPTCHA-Lösung besteht größtenteils aus Warten. Ihr Prozess liegt brach, während der Solver arbeitet, geht dann zum nächsten über und liegt wieder brach. Drei Lösungen brauchen die Zeit von drei Lösungen. Dreißig eben die von dreißig.

Mehrere Captchas gleichzeitig mit asyncio.gather lösen

Tauschen Sie den Client, awaiten Sie die Aufrufe und übergeben Sie alle an gather. Die Typen dürfen gemischt werden: reCAPTCHA, Turnstile und GeeTest im selben Stapel sind völlig in Ordnung.

# pip install capskip
import asyncio
from capskip import AsyncCapSkip

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)

    # gather starts all three now and waits for the slowest.
    results = await asyncio.gather(
        solver.recaptcha(sitekey="SITEKEY_A", url="https://example.com/page-a"),
        solver.turnstile(sitekey="SITEKEY_B", url="https://example.com/page-b"),
        solver.normal("captcha.png"),
    )

    for r in results:
        print(r["code"][:40])   # token for widgets, text for images

asyncio.run(main())

Jede Methode gibt ein dict mit denselben Kernfeldern zurück: captchaId und code. Turnstile liefert zusätzlich userAgent, das Sie zusammen mit dem Token zurücksenden müssen, wenn Sie die Lösung einer Challenge-Seite einreichen. GeeTest ergänzt challenge, validate und seccodeund legt das rohe JSON in code.

Begrenzen Sie die Nebenläufigkeit mit einem Semaphore

Feuern Sie nicht zweihundert Lösungen auf einen lokalen Daemon ab und hoffen Sie das Beste. Das Lösen ist CPU-Arbeit auf Ihrem eigenen Rechner. Ab einer gewissen Breite stellen Sie sich also nur selbst in die Warteschlange, und jede einzelne Lösung wird langsamer. Ein Semaphore hält eine feste Anzahl gleichzeitig in Arbeit.

import asyncio
from capskip import AsyncCapSkip

# Start at 4 or 5, then measure. More is not automatically faster.
sem = asyncio.Semaphore(5)

async def solve_one(solver, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def run(targets):
    solver = AsyncCapSkip()
    tasks = [solve_one(solver, k, u) for k, u in targets]
    return await asyncio.gather(*tasks)

Ermitteln Sie die Zahl durch Messen, nicht durch Raten. Lassen Sie dieselben 20 Ziele mit 2, 5 und 10 laufen und behalten Sie den Wert, der auf Ihrer Hardware am schnellsten fertig wird. Die richtige Antwort hängt von Ihrer CPU ab, nicht vom SDK.

Ein Client, nicht einer pro Task

Erzeugen Sie einen einzigen AsyncCapSkip und teilen Sie ihn wie oben zwischen den Coroutinen. Einen pro Task zu erzeugen ist Verschwendung und bringt nichts: Der Client hält Konfiguration, keinen Zustand einzelner Lösungen.

Verhindern, dass ein Fehlschlag den ganzen Stapel mitreißt

Standardmäßig gather reicht die erste Ausnahme nach oben durch, und Sie verlieren die Ergebnisse von allem anderen, was gerade lief. Übergeben Sie return_exceptions=True , dann kommen die Ausnahmen als ganz normale Einträge in der Ergebnisliste an, und Sie können Treffer von Fehlschlägen trennen.

from capskip import (
    AsyncCapSkip, ApiException, NetworkException,
    TimeoutException, ValidationException,
)

results = await asyncio.gather(*tasks, return_exceptions=True)

for target, r in zip(targets, results):
    if isinstance(r, TimeoutException):
        print("timed out, worth retrying:", target)
    elif isinstance(r, ApiException):
        print("api rejected this one:", target, r)
    elif isinstance(r, NetworkException):
        print("solver unreachable, stop the run:", target)
    elif isinstance(r, Exception):
        raise r
    else:
        print("ok:", r["code"][:40])

Die vier Ausnahmetypen sind in jedem CapSkip-SDK gleich, und alle leiten sich von einer Basisklasse CapSkipError ab, falls Sie lieber nur einen Typ abfangen möchten. ValidationException bedeutet, dass Ihre Parameter falsch sind und ein erneuter Versuch identisch scheitern wird. NetworkException bedeutet in der Regel, dass die App nicht läuft – ein Problem des gesamten Laufs, nicht eines einzelnen Ziels.

Timeouts und Polling, die sich je nach Typ unterschiedlich verhalten

Es greifen zwei getrennte Timeouts, und ein Stapel mit gemischten Typen unterliegt beiden.

Option	Standard	Gilt für
defaultTimeout	120 Sekunden	Bild-Captchas
recaptchaTimeout	300 Sekunden	reCAPTCHA, Turnstile, GeeTest
pollingInterval	5 Sekunden	The maximaler Abstand zwischen zwei Abfragen

pollingInterval sollten Sie verstehen, bevor Sie daran drehen. Das SDK pollt nicht in einem festen Intervall. Es beginnt bei 250 ms und nähert sich schrittweise dem von Ihnen gesetzten Wert an. Genau deshalb liefert eine SDK-Lösung meist früher ein Ergebnis als eine selbst gebaute Schleife nach dem Rat der rohen API, „zu warten und dann alle fünf Sekunden abzufragen“. Ein höherer Wert lässt schnelle Lösungen später eintreffen. Ein niedrigerer erzeugt nur mehr Anfragen ohne jeden Gewinn.

GeeTest-Challenges laufen ab – bauen Sie den Stapel also nicht im Voraus

Das beißt gerade dann, wenn Sie auf Parallelbetrieb umstellen. Der GeeTest-Wert gt ist pro Website konstant, aber challenge ist einmalig und läuft nach etwa einer Minute ab. Wenn Sie erst fünfzig Challenges einsammeln und dann mit dem Lösen beginnen, sind die am Ende der Schlange tot, bevor sie eingereicht werden. Holen Sie jede Challenge unmittelbar vor der Lösung, die sie verwendet.

Vollständiges lauffähiges Beispiel

# pip install capskip
import asyncio
from capskip import AsyncCapSkip, ApiException, TimeoutException

TARGETS = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

async def solve_one(solver, sem, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)
    sem = asyncio.Semaphore(5)

    tasks = [solve_one(solver, sem, k, u) for k, u in TARGETS]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    tokens = {}
    for (sitekey, url), r in zip(TARGETS, results):
        if isinstance(r, (ApiException, TimeoutException)):
            print("failed:", url, r)
        else:
            tokens[url] = r["code"]

    print(len(tokens), "of", len(TARGETS), "solved")
    return tokens

asyncio.run(main())

Das ist das ganze Muster: ein gemeinsamer Client, ein Semaphore, return_exceptions=Trueund am Ende ein dict mit Tokens. Setzen Sie es in einen Scraper ein, und der Captcha-Schritt ist nicht länger der Engpass. Die Seite Captcha-Löser für Web Scraping beschreibt, wo das in eine größere Pipeline passt.

Dieselbe Idee in den anderen SDKs

Wenn Sie das portieren, ändert sich das Nebenläufigkeits-Primitiv, die Struktur aber nicht. Node ist ohnehin nicht blockierend, dort genügt der einfache Client.

// npm install capskip
const { CapSkip } = require('capskip');

const solver = new CapSkip({ host: '127.0.0.1', port: 8080 });

// AsyncCapSkip here is just an alias. Promise.all does the work.
const results = await Promise.all([
  solver.recaptcha('SITEKEY_A', 'https://example.com/page-a'),
  solver.turnstile('SITEKEY_B', 'https://example.com/page-b'),
]);

console.log(results.map(r => r.code));

.NET ist dieselbe Geschichte mit Task.WhenAll, und PHP hat überhaupt keine Nebenläufigkeit zu bieten. Wenn Sie parallel lösen müssen und die Sprache frei wählen können, ist Python die mit dem eigens dafür gebauten Client. Die Seite Python-Captcha-Löser beschreibt den Rest der Schnittstelle.

Häufige Fehler

Fehler	Was passiert	Beheben
Verwendung von CapSkip innerhalb von Coroutinen	Blockiert die Event-Loop, die Laufzeit bleibt also sequenziell	Verwenden AsyncCapSkip
Kein Semaphore	Jede Lösung wird langsamer, sobald die Warteschlange tief wird	Begrenzen Sie die gleichzeitige Arbeit, beginnen Sie bei etwa 5
Einfaches gather	Ein Fehlschlag verwirft alle übrigen Ergebnisse	return_exceptions=True
GeeTest-Challenges im Voraus holen	Die späteren laufen vor dem Einreichen ab	Holen Sie jede unmittelbar vor dem Lösen
Erhöhen von pollingInterval	Schnelle Lösungen kommen später zurück, nicht früher	Belassen Sie es beim Standardwert
Proxy bei einer Bildlösung gesetzt	Für Bild-Captchas nicht unterstützt	Proxys gelten nur für reCAPTCHA, Turnstile und GeeTest

Die rohe Anfrage und Antwort für jeden Typ finden Sie, falls Sie sehen möchten, was das SDK tatsächlich sendet, auf der Seite API-Dokumentation. Siehe die Python-Dokumentation: Referenz zu asyncio-Tasks deckt ab gather beschreibt die Semantik im Detail.

Häufig gestellte Fragen

Wie viele Captchas kann ich gleichzeitig lösen?

Es gibt kein Kontingent, an das Sie stoßen könnten – die Grenze ist Ihre eigene Hardware. Gelöst wird lokal, die Nebenläufigkeit ist also durch die CPU begrenzt und nicht durch eine Kontostufe. Beginnen Sie mit fünf gleichzeitig, stoppen Sie die Zeit für einen festen Stapel und passen Sie von dort an.

Kann ich Captcha-Typen in einem gather-Aufruf mischen?

Ja. recaptcha, turnstile, geetest und normal sind allesamt Coroutinen auf demselben Client und können gemeinsam awaited werden. Denken Sie daran, dass Bildlösungen das Timeout von 120 Sekunden nutzen, alle übrigen 300.

Sollte ich stattdessen Threads verwenden?

Nur wenn Ihr umgebender Code ohnehin schon threaded ist. Die Arbeit besteht aus I/O-gebundenem Warten, und genau dafür ist asyncio da; eine Event-Loop ist zudem günstiger als ein Thread-Pool. Wenn Sie in einer synchronen Codebasis feststecken, ist ein Thread-Pool um den einfachen CapSkip Client funktioniert ebenfalls.

Muss AsyncCapSkip geschlossen werden?

Das SDK dokumentiert weder eine close-Methode noch einen asynchronen Kontextmanager. Erzeugen Sie also einen Client, nutzen Sie ihn für den gesamten Lauf und lassen Sie ihn am Prozessende aus dem Gültigkeitsbereich fallen.

Zusammenfassung

Verwenden AsyncCapSkip, teilen Sie einen Client, begrenzen Sie die gleichzeitigen Lösungen mit einem Semaphore und übergeben Sie return_exceptions=True , damit ein einzelnes schlechtes Ziel nicht den Stapel verwirft. Damit wird aus einer Schlange von Captchas statt eines seriellen Engpasses ein einziges Warten.

Sie können die Nebenläufigkeit deshalb frei erhöhen, weil der Solver auf Ihrem eigenen Rechner läuft. Es gibt keine Rechnung pro Lösung und keine geteilte Warteschlange mit Fremden. Skalieren ist damit eine Frage Ihrer CPU statt des Rate-Limits eines anderen. Genau diesen praktischen Unterschied macht ein lokales Captcha-Umgehung Werkzeug, sobald Ihre Stapelgrößen nicht mehr klein sind.