Captcha in einem Prefect-Flow mit Wiederholungen lösen

Ein Captcha-Schritt in Prefect ist ein einzelner Task mit Wiederholungen, und der ganze Flow umfasst etwa zwanzig Zeilen. Das Detail, das Umsteiger von anderen gehosteten Plattformen überrascht, ist ein angenehmes: Prefect Cloud führt Ihren Code nie aus. Es plant Arbeit ein, und ein Worker auf Ihrer eigenen Maschine holt sie sich über eine ausgehende Verbindung ab. Der Löser kann also auf 127.0.0.1 sitzen, und der Flow erreicht ihn, was bei Zapier oder Make.com nicht der Fall ist. Die zwei Dinge, die tatsächlich schiefgehen, sind das Wiederholungsverhalten und das Caching, und beide sind je ein Argument.

Was Sie brauchen

• Prefect 3 und Python 3.10 oder neuer, dazu das CapSkip Python SDK.
• Ein Prefect-Cloud-Workspace oder ein selbst gehosteter Prefect-Server. Beides funktioniert hier auf dieselbe Weise.
• Die Seiten-URL des geschützten Formulars und dessen sitekey.
• CapSkip läuft im Local-Modus, wenn Worker und Löser sich eine Maschine teilen, oder im Server-Modus, wenn das nicht so ist. Beide sind beschrieben unter Verbindungseinstellungen.

# pip install prefect
pip install -U prefect capskip

# Point the CLI at your workspace, then start a worker
# on the machine that should run the flows.
prefect cloud login

Wo Ihr Flow tatsächlich läuft

Das ist die Frage, die Ihr gesamtes Netzwerk-Setup entscheidet, beantworten Sie sie also zuerst. Die Voreinstellung von Prefect ist ein hybrides Modell: Die Orchestrierungsebene ist gehostet, die Ausführungsebene gehört Ihnen. Prefect Cloud speichert Metadaten und koordiniert Läufe, es führt Ihren Code nicht aus, und es braucht keinen eingehenden Zugriff auf Ihr Netzwerk. Ein Worker, den Sie in Ihrer eigenen Infrastruktur starten, fragt nach außen nach Arbeit und startet die Läufe lokal.

Die praktische Folge sollte man klar aussprechen. Lassen Sie einen Process-Worker auf derselben Windows-Maschine wie CapSkip laufen, und der Flow ruft 127.0.0.1:8080 genau so auf, wie es ein Skript auf Ihrem Schreibtisch täte. Kein Tunnel, keine öffentliche Adresse, kein Zertifikat. Das ist das Gegenteil der Lage auf einer reinen Cloud-Automatisierungsplattform, und es ist der Hauptgrund dafür, dass ein Orchestrator ein bequemer Ort für Lösungsarbeit ist.

Es gibt eine Ausnahme, und Sie sollten sie kennen, bevor Sie einen Work Pool wählen. Prefect Managed Work Pools führen Ihren Flow auf der Infrastruktur von Prefect aus statt auf Ihrer, was bequem ist und die Loopback-Option vollständig beseitigt. In dieser Konfiguration braucht der Löser den Server-Modus und eine erreichbare Adresse. Prefect veröffentlicht sechs statische ausgehende Adressen, die Managed-Läufe verwenden, Sie können also genau diese durch die Firewall zum Port des Lösers zulassen und alles andere verwerfen. Managed-Läufe müssen außerdem ein offizielles Prefect-Image verwenden und sind auf 24 Stunden begrenzt, ein weiterer Grund, warum ein Process- oder Docker-Work-Pool hier meist besser passt.

Schritt 1: den Schlüssel in einen Secret block legen

Schreiben Sie den API-Schlüssel nicht fest in die Flow-Datei. Prefect bringt einen Secret block mit, die Werte werden im Backend verschlüsselt gespeichert, und das Laden ist zwei Zeilen lang. Speichern Sie ihn einmal aus einer Python-Shell.

# pip install prefect
from prefect.blocks.system import Secret

secret = Secret(value="YOUR_API_KEY")
secret.save("capskip-api-key")

# Rotating it later needs overwrite, or the save is refused.
# secret.save("capskip-api-key", overwrite=True)

Schritt 2: den Solve-Task schreiben

Ein Task, eine Lösung. Geben Sie ihm Wiederholungen, denn ein Löser-Aufruf ist ein Netzwerkaufruf, und Netzwerkaufrufe schlagen fehl. Prefect nimmt eine feste Verzögerung, eine Liste von Verzögerungen oder einen Helper für exponentielles Backoff entgegen, und ein Jitter-Faktor zieht die Wiederholungen auseinander, damit eine Reihe von Fehlschlägen nicht im Gleichschritt zurückkommt.

Das wichtige Argument ist das andere. Ein Captcha-Token ist zur einmaligen Verwendung und läuft innerhalb weniger Minuten ab, es darf also niemals aus einem Cache kommen. Prefect 3 lässt das Caching aus, solange die Ergebnispersistenz nicht eingeschaltet ist, die meisten sind damit zufällig auf der sicheren Seite. Wenn Ihr Team die Persistenz global aktiviert hat, und das haben viele, kann ein wiederholter Task das Token zurückgeben, das er beim ersten Mal erzeugt hat, und das Formular weist es ab. Setzen Sie die Policy explizit und denken Sie nicht mehr darüber nach.

# pip install capskip
from prefect import task
from prefect.tasks import exponential_backoff
from prefect.cache_policies import NO_CACHE
from prefect.blocks.system import Secret
from capskip import CapSkip

# NO_CACHE matters: a token is valid once and expires fast.
@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=5),
    retry_jitter_factor=0.5,
    cache_policy=NO_CACHE,
)
def solve_recaptcha(sitekey: str, page_url: str) -> str:
    key = Secret.load("capskip-api-key").get()
    solver = CapSkip(host="127.0.0.1", port=8080, apiKey=key)
    return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

Eine Methode deckt reCAPTCHA v2, Invisible, Enterprise und v3 ab. Die Varianten sind Optionen und keine getrennten Aufrufe: invisible=1, enterprise=1 oder version="v3" mit einer Action. Turnstile und GeeTest haben eigene Methoden und dieselbe Form. Die vollständigen Parameterlisten stehen in der CapSkip-API-Dokumentation.

Schritt 3: die Fehler nicht wiederholen, die nie durchgehen

Blindes Wiederholen verschwendet Zeit auf Fehlschläge, die deterministisch sind. Ein fehlerhaft geformtes Argument scheitert bei jedem Versuch identisch, und ein sitekey, der nicht zur Seite gehört, ebenso. Eine Retry-Condition-Funktion bekommt den State und entscheidet, und ein zurückgegebenes False beendet den Task sofort mit der ursprünglichen Exception.

# Retry the transient ones. Fail fast on the rest.
from capskip import ValidationException, ApiException

def worth_retrying(task, task_run, state) -> bool:
    try:
        state.result()
    except (ValidationException, ApiException):
        return False   # bad arguments or a bad sitekey
    except Exception:
        return True    # solver down, or a timeout
    return True

Übergeben Sie sie als retry_condition_fn am Task. NetworkException bedeutet, dass CapSkip nicht läuft oder der Host falsch ist, und TimeoutException bedeutet, dass die Lösung länger gedauert hat als recaptchaTimeout, dessen Standardwert bei 300 Sekunden liegt. Beide sind wirklich einen weiteren Versuch wert. Diese zwei leiten sich zusammen mit ValidationException und ApiException von einer gemeinsamen Basis ab, ein Abfangen von CapSkipError funktioniert also, wenn Sie Fehler lieber an einer Stelle behandeln.

Vollständiges lauffähiges Beispiel

Der gesamte Flow. Die Seite abrufen, den sitekey daraus ziehen, lösen, dann das Token mit dem Formular zurücksenden. Jeder Schritt ist ein Task, jeder bekommt also eigene Wiederholungen, eigene Logs und einen eigenen Eintrag im Ausführungsgraphen.

# pip install prefect capskip httpx
import re
import httpx
from prefect import flow, task
from prefect.tasks import exponential_backoff
from prefect.cache_policies import NO_CACHE
from prefect.blocks.system import Secret
from capskip import CapSkip

PAGE_URL = "https://example.com/page-with-recaptcha"

@task(retries=2, retry_delay_seconds=5)
def read_sitekey(page_url: str) -> str:
    html = httpx.get(page_url, timeout=30).text
    match = re.search(r'data-sitekey=["\']([^"\']+)', html)
    if not match:
        raise RuntimeError("No data-sitekey on the page.")
    return match.group(1)

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=5),
    cache_policy=NO_CACHE,
)
def solve_recaptcha(sitekey: str, page_url: str) -> str:
    key = Secret.load("capskip-api-key").get()
    solver = CapSkip(host="127.0.0.1", port=8080, apiKey=key)
    return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

@task(retries=2, cache_policy=NO_CACHE)
def submit_form(page_url: str, token: str) -> int:
    reply = httpx.post(
        page_url,
        data={"g-recaptcha-response": token},
        timeout=30,
    )
    return reply.status_code

@flow(name="captcha-protected-submit")
def run():
    sitekey = read_sitekey(PAGE_URL)
    token = solve_recaptcha(sitekey, PAGE_URL)
    return submit_form(PAGE_URL, token)

if __name__ == "__main__":
    print(run())

Lösen Sie unmittelbar vor dem Absenden, niemals in einem früher eingeplanten Schritt. Ein Token, das zehn Minuten in einem Result Store liegt, während ein vorgelagerter Task fertig wird, ist tot, sobald das Formular es sieht.

Einen Batch lösen, ohne den Löser zu überfluten

Prefect führt Tasks standardmäßig nebenläufig über einen Thread-Pool aus, hundert Lösungen sind also eine List Comprehension über submit und überhaupt keine Task-Runner-Konfiguration. Das ist mehr Parallelität, als Sie vermutlich auf eine einzige Maschine richten wollen.

Die Kontrolle dafür ist ein globales Concurrency Limit. Legen Sie das Limit einmal per CLI an, belegen Sie dann innerhalb des Tasks einen Slot, und jeder Lauf über der Obergrenze wartet, statt sich obendrauf zu stapeln.

# Create the limit once. Six solves in flight at a time.
prefect gcl create capskip --limit 6

# The limit is enforced across every flow run, not per flow.
from prefect import flow, task
from prefect.cache_policies import NO_CACHE
from prefect.concurrency.sync import concurrency
from prefect.futures import wait
from capskip import CapSkip

@task(retries=3, cache_policy=NO_CACHE)
def solve_one(sitekey: str, page_url: str) -> str:
    with concurrency("capskip", occupy=1):
        solver = CapSkip(host="127.0.0.1", port=8080)
        return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

@flow
def solve_many(sitekey: str, urls):
    # submit, not map: map would iterate the sitekey string.
    futures = [solve_one.submit(sitekey, u) for u in urls]
    wait(futures)

Das Limit gilt über jeden Flow-Lauf im Workspace hinweg, und genau das wollen Sie, wenn drei Zeitpläne auf denselben Löser zeigen. Wenn Sie den Fan-out lieber innerhalb eines einzelnen Prozesses erledigen: AsyncCapSkip aus dem Python SDK ist ein echter asyncio-Client, und dieser Ansatz wird behandelt in der Anleitung zum parallelen Lösen von Captchas.

Den Löser auf einer anderen Maschine betreiben

Worker wandern. Aus einem Process-Worker auf Ihrem Schreibtisch wird ein Docker-Worker auf einem Server, dann ein Kubernetes-Work-Pool, und irgendwann läuft der Flow nicht mehr auf der Maschine, auf der der Löser läuft. Im Code ändert sich nichts außer dem Host.

CapSkip hat zwei Verbindungsmodi. Local bindet an 127.0.0.1 und antwortet nur diesem Gerät. Server bindet an Ihre Netzwerk- oder öffentliche IP, sodass ein Worker auf einer VM, ein Container-Host oder ein Managed Work Pool dieselbe Windows-Maschine über die API aufruft. Eine statische öffentliche IP hält diese Adresse stabil. Es bleibt so oder so Ihre eigene Hardware und bleibt ohne Verbrauchsabrechnung, die Kosten eines arbeitsreichen Tages ändern sich also nicht mit dem Modus.

# Same SDK, same call. Only the host moves.
solver = CapSkip(host="10.0.0.12", port=8080, apiKey=key)

Schalten Sie die Schlüsselvalidierung ein, sobald der Löser auf einer Netzwerkadresse lauscht, und geben Sie jedem Worker einen eigenen Schlüssel, damit einer widerrufen werden kann, ohne die anderen anzufassen. Beide Modi werden durchgegangen in der CapSkip-Einrichtungsanleitung.

Häufige Fehler und was sie bedeuten

FAQ

Die Kurzfassung

Legen Sie den Schlüssel in einen Secret block, verpacken Sie die Lösung in einen Task mit Wiederholungen und NO_CACHE, und rufen Sie ihn im Schritt direkt vor dem Absenden auf. Lassen Sie einen Worker auf der Maschine laufen, auf der CapSkip liegt, und der Host bleibt 127.0.0.1. Fügen Sie ein globales Concurrency Limit hinzu, bevor Sie einen Batch auffächern. Die Python-Seite von all dem finden Sie auf der Python-Captcha-Solver-Seite. Dieselben drei Aufrufe gibt es auch in Node.js, PHP und C#. Aufgeführt sind sie hier: Die Seite zu den SDKs fürs Captcha-Lösen.

Eines sollten Sie wissen, bevor Sie das stündlich einplanen. CapSkip erledigt die Captcha-Umgehung auf Hardware, die Ihnen bereits gehört, und deshalb kostet ein Flow, der zehntausend am Tag löst, genauso viel wie einer, der zehn löst.