So betreiben Sie einen FastAPI-Captcha-Solver-Dienst unter Windows

Um einen FastAPI-Captcha-Solver-Dienst zu betreiben, erstellen Sie beim Start der App einen einzigen AsyncCapSkip-Client. Stellen Sie einen POST-Endpunkt bereit, der einen sitekey und eine Seiten-URL entgegennimmt. Umhüllen Sie jeden Lösevorgang mit asyncio.timeout und einer Semaphore und ordnen Sie die Exceptions des SDK HTTP-Statuscodes zu. Scraper, Worker und Skripte in jeder Sprache rufen dann eine einzige interne Adresse auf und sehen nie den Port oder den Schlüssel von CapSkip. CapSkip läuft auf Ihrem eigenen Windows-Rechner, daher kommt beim Dienst keine Gebühr pro Lösung hinzu. Dieser Leitfaden baut ihn in etwa 100 Zeilen und zeigt, was jeder Teil tut, wenn CapSkip langsam, geschlossen oder stumm ist.

Was Sie brauchen

• CapSkip läuft auf einem Windows-PC, mit eingeschaltetem reCAPTCHA-Löser. Die Einrichtungsanleitung führt durch den ersten Start.
• Python 3.11 oder neuer, weil der Code asyncio.timeout verwendet. Alles unten lief unter Windows 11 mit Python 3.14.0, capskip 1.3.0, fastapi 0.150.0 und uvicorn 0.54.0.
• Eine PowerShell-Eingabeaufforderung. Die Befehle verwenden PowerShell-Syntax.
• Eine Seite mit einer reCAPTCHA-v2-Checkbox. Der Dienst braucht nur ihren sitekey und ihre Adresse.
• Eine Adresse für CapSkip. Im Local-Modus antwortet es auf 127.0.0.1 nur für dieses Gerät. Im Server-Modus lauscht es auf Ihrer Netzwerkadresse oder öffentlichen IP, damit ein Dienst auf einem anderen Rechner es über die API aufrufen kann. Eingestellt werden beide Modi unter Verbindungseinstellungen.

Die Zahlen in diesem Leitfaden stammen von einem Mock-CapSkip (einer nachgebauten Testattrappe) auf demselben PC. FastAPI, uvicorn und das SDK sind die echten. Der Mock hält jeden Lösevorgang eine festgelegte Zeit, meist zwei Sekunden, daher weichen Ihre echten Lösevorgänge ab. Nichts hier wurde gegen ein echtes CapSkip oder eine echte Website ausgeführt.

Warum ein Dienst vor CapSkip sinnvoll ist

Löst nur ein Python-Skript Captchas, rufen Sie das SDK direkt aus diesem Skript auf. Die Python-Captcha-Solver-Seite behandelt das, und die Anleitung zum parallelen Lösen behandelt viele Lösungen in einem Prozess. Ein Dienst lohnt sich, wenn mehrere Aufrufer ein CapSkip teilen. Dann gehören vier Aufgaben an einen Ort.

• Eine Frist. Das Python-SDK wartet standardmäßig bis zu 300 Sekunden. Sein einfacher Client setzt überhaupt kein HTTP-Timeout. Aufrufer sollten nicht jeder einen eigenen Timer mit sich führen.
• Ein Limit. CapSkip führt eine feste Zahl von Lösevorgängen gleichzeitig aus, und weitere Aufgaben warten darin. Eine Warteschlange, die Sie selbst kontrollieren, lässt sich leichter durchschauen.
• Ein Schlüssel. Aufrufer erhalten einen Dienstschlüssel. Adresse und API-Schlüssel von CapSkip bleiben in den Einstellungen des Dienstes.
• Jede Sprache. Ein C#-Worker, eine PHP-Website und eine PowerShell-Aufgabe sprechen alle HTTP und JSON.

Schritt 1: Die Pakete installieren

Legen Sie zuerst einen Ordner und eine virtuelle Umgebung an, wenn Sie pro Projekt eine führen. Das Paket capskip zieht httpx nach, das der asynchrone Client verwendet.

pip install capskip fastapi uvicorn

Schritt 2: Die Einstellungen als Umgebungsvariablen setzen

Das SDK liest selbst keine Umgebungsvariablen, daher liest der Dienst sie und reicht jeden Wert weiter. Setzen Sie sie in dem PowerShell-Fenster, aus dem Sie den Dienst starten. Ersetzen Sie beide Platzhalterschlüssel. YOUR_CAPSKIP_KEY ist der API-Schlüssel aus den Einstellungen von CapSkip, und solange die Schlüsselvalidierung in CapSkip aus ist, funktioniert beliebiger Text. YOUR_API_KEY ist ein Schlüssel, den Sie für Ihre Aufrufer erfinden.

# Where CapSkip listens and which key it expects.
$env:CAPSKIP_HOST = "127.0.0.1"
$env:CAPSKIP_PORT = "8080"
$env:CAPSKIP_API_KEY = "YOUR_CAPSKIP_KEY"
# Match Max. Threads in the CapSkip settings.
$env:CAPSKIP_MAX_THREADS = "10"
# The key your callers must send to this service.
$env:SOLVER_API_KEY = "YOUR_API_KEY"
# Seconds a caller may wait, queue time included.
$env:SOLVE_DEADLINE = "120"

Verwenden Sie 127.0.0.1 als Host. Mit diesem Client dauerte eine verweigerte Verbindung auf 127.0.0.1 etwa 2,4 Sekunden und auf localhost etwa 2,7 Sekunden. Der einfache requests-Client brauchte in einem früheren Test auf localhost etwa 4 Sekunden.

Schritt 3: Die Einstellungen lesen

Erstellen Sie main.py. Die Blöcke in den Schritten 3 bis 10 sind aufeinanderfolgende Teile dieser einen Datei. Die vollständige Datei am Ende führt ihre Importe oben zusammen.

import asyncio
import os
from contextlib import asynccontextmanager

from capskip import AsyncCapSkip
from fastapi import FastAPI

# The SDK reads no env vars, so pass each setting in by hand.
HOST = os.getenv("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.getenv("CAPSKIP_PORT", "8080"))
KEY = os.getenv("CAPSKIP_API_KEY", "capskip")
MAX_THREADS = int(os.getenv("CAPSKIP_MAX_THREADS", "10"))
DEADLINE = float(os.getenv("SOLVE_DEADLINE", "120"))

Schritt 4: Beim Start der App einen Client und eine Schleuse bauen

Eine lifespan-Funktion läuft einmal beim Start. Sie erstellt den Client und eine Semaphore und legt beide im App-Zustand ab, sodass jede Anfrage sie teilt. Geben Sie der Semaphore einen Slot pro CapSkip-Thread, also die Einstellung Max. Threads in CapSkip (Standardwert 10).

AsyncCapSkip ist ein echter asynchroner Client in Python. Er baut auf httpx auf und pollt mit asyncio.sleep, sodass das Warten zwischen den Abfragen die Event-Loop nicht blockiert. Die Aufrufe selbst kosten etwas CPU-Zeit, wie der Abschnitt Ein versteckter Preis erklärt. Der einfache CapSkip-Client ruft requests ohne Timeout auf, daher würde ein stummes CapSkip einen Thread für immer halten. Setzt man recaptchaTimeout auf die Frist, bleibt die eigene Uhr des SDK gleich unserer. Bei Gleichstand löst asyncio.timeout zuerst aus.

@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client and one gate, shared by every request.
    app.state.cap = AsyncCapSkip(
        apiKey=KEY, host=HOST, port=PORT,
        recaptchaTimeout=DEADLINE,
    )
    app.state.slots = asyncio.Semaphore(MAX_THREADS)
    yield


app = FastAPI(lifespan=lifespan)

Schritt 5: Den Dienstschlüssel prüfen

Ein ungeschützter Endpunkt zum Lösen auf einer Netzwerkadresse ist ein kostenloser Solver für jeden, der den Port erreichen kann. Diese Dependency vergleicht den Header X-API-Key mit secrets.compare_digest, das dieselbe Zeit braucht, egal wo sich die Bytes unterscheiden. Ein fehlender, leerer, falscher, längerer und nicht-ASCII-Schlüssel bekamen auf dem Mock alle 401. Der richtige Schlüssel bekam 200.

Die Prüfung am Anfang verhindert, dass der Dienst mit einem leeren Schlüssel startet. Ohne sie würde eine Anfrage ohne Header standardmäßig einen leeren String bekommen und passen. Die Dependency läuft, bevor der Body geprüft wird, daher bekommt eine Anfrage ohne Schlüssel und mit einem Body, der die Validierung nicht besteht, 401, nicht 422. Ein Body, der kein gültiges JSON ist, bekommt weiterhin 422, weil FastAPI das JSON zuerst dekodiert. Aufrufer ohne Schlüssel erfahren wenig über Ihr Schema.

import secrets

from fastapi import Depends, Header, HTTPException

SERVICE_KEY = os.environ["SOLVER_API_KEY"]
if not SERVICE_KEY:
    raise SystemExit("SOLVER_API_KEY must not be empty")


def require_key(x_api_key: str = Header(default="")):
    good = secrets.compare_digest(
        x_api_key.encode(), SERVICE_KEY.encode())
    if not good:
        raise HTTPException(
            401, "missing or wrong X-API-Key",
            headers={"WWW-Authenticate": "APIKey"})

Schritt 6: Die Anfrage validieren und die Frist setzen

Das Anfragemodell weist einen leeren sitekey oder eine URL zurück, die nicht mit http beginnt. FastAPI beantwortet einen fehlerhaften Body mit seiner eigenen 422 und listet jedes Problem auf.

Die Frist ist der wichtige Teil. asyncio.timeout umschließt das Warten auf einen Slot und den Lösevorgang gemeinsam. Ein gering ausgelasteter Dienst antwortet insgesamt innerhalb von DEADLINE Sekunden. Würde das Timeout nur den Lösevorgang umschließen, könnte eine Anfrage minutenlang in der Warteschlange stehen, bevor ihre Uhr überhaupt startet. Unter hoher Last kann die Antwort später kommen, wie der Abschnitt Ein versteckter Preis erklärt.

from fastapi import Request
from pydantic import BaseModel, Field


class SolveRequest(BaseModel):
    sitekey: str = Field(min_length=1)
    url: str = Field(pattern=r"^https?://")


async def solve_v2(state, sitekey: str, url: str) -> str:
    # The deadline also covers the wait for a free slot.
    async with asyncio.timeout(DEADLINE):
        async with state.slots:
            result = await state.cap.recaptcha(sitekey, url)
    return result["code"]

Schritt 7: Den Endpunkt zum Lösen hinzufügen

Der Endpunkt besteht aus drei Zeilen. Zuerst läuft die Schlüssel-Dependency, dann der Lösevorgang, dann geht der Token als JSON zurück.

@app.post("/solve", dependencies=[Depends(require_key)])
async def solve(body: SolveRequest, request: Request):
    token = await solve_v2(request.app.state, body.sitekey, body.url)
    return {"token": token}

Schritt 8: Die Exceptions des SDK Statuscodes zuordnen

Das SDK löst vier Exception-Klassen aus, alle unter CapSkipError und alle aus capskip importierbar. Ein einziger Handler auf der Basisklasse macht aus jeder einen Statuscode und einen kurzen Fehlernamen. Die Details gehen ins Log und nie an den Aufrufer, sodass die Meldungen von CapSkip privat bleiben.

import logging

from capskip import ApiException, CapSkipError, NetworkException
from capskip import TimeoutException, ValidationException
from fastapi.responses import JSONResponse

STATUS = {
    NetworkException: (503, "capskip_unreachable"),
    TimeoutException: (504, "capskip_timeout"),
    ValidationException: (422, "bad_parameters"),
    ApiException: (502, "capskip_error"),
}

@app.exception_handler(CapSkipError)
async def capskip_error(request: Request, exc: CapSkipError):
    # Log the detail here, keep it away from the caller.
    logging.getLogger("uvicorn.error").warning("%r", exc)
    status, code = STATUS.get(type(exc), (502, "capskip_error"))
    if "UNSOLVABLE" in str(exc):
        status, code = 422, "unsolvable"
    return JSONResponse({"error": code}, status)

Jede Zeile unten wurde gegen den Mock gezielt ausgelöst.

Die 502 ist Absicht. Ein abgelehnter CapSkip-Schlüssel ist der Fehler des Dienstes, nicht des Aufrufers. Die Zeile mit 5,2 Sekunden ist das Standard-Timeout von httpx, 5 Sekunden, das der asynchrone Client erbt. Eine stumme Antwort auf eine Abfrage ist anders. Das SDK behandelt eine fehlgeschlagene Abfrage als noch nicht bereit und versucht es erneut, daher beendet nur die Frist sie. Eine ValidationException kann mit nur einem sitekey und einer URL nicht auftreten. Sie braucht zusätzliche Optionen wie eine Action bei einem v2-Lösevorgang, die dieser Dienst nie sendet.

Schritt 9: Die Frist gesondert abfangen

asyncio.timeout löst Pythons eingebautes TimeoutError aus. Die TimeoutException des SDK ist eine andere Klasse, und das Abfangen der eingebauten fängt sie nicht ab. Der Handler aus Schritt 8 ordnet die TimeoutException des SDK bereits einer 504 namens capskip_timeout zu. Dieser zweite Handler deckt die Frist ab, und sein Fehlername sagt dem Aufrufer, welche Uhr abgelaufen ist.

@app.exception_handler(TimeoutError)
async def deadline_hit(request: Request, exc: TimeoutError):
    # asyncio.timeout raises the built-in TimeoutError. The SDK's
    # TimeoutException is a different class, so it needs its own entry.
    return JSONResponse({"error": "deadline_exceeded"}, 504)

Schritt 10: Einen Health-Check hinzufügen

CapSkip hat keinen Ping-Aufruf, daher öffnet der Check eine TCP-Verbindung zu seinem Port und gibt nach drei Sekunden auf. Bei geschlossenem CapSkip antwortete er nach etwa 2 Sekunden mit 503, und bei laufendem Mock antwortete er sofort mit 200. Er sagt einem Überwachungsskript, dass der Dienst läuft und CapSkip nicht. Er beweist nicht, dass ein Lösevorgang funktioniert.

@app.get("/health")
async def health():
    # CapSkip has no ping call, so check that its port accepts a
    # connection.
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(HOST, PORT), timeout=3)
    except (OSError, TimeoutError):
        return JSONResponse({"ok": False}, 503)
    writer.close()
    return {"ok": True}

Unter Windows ausführen

Starten Sie uvicorn aus dem Ordner, der main.py enthält. An 127.0.0.1 gebunden, antwortet der Dienst nur auf diesem PC.

python -m uvicorn main:app --host 127.0.0.1 --port 8000

Für Aufrufer auf anderen Rechnern ändern Sie den Host auf 0.0.0.0, was jeden Netzwerkadapter bedeutet. Auf dem Test-PC antwortete das unter der LAN-Adresse des PCs, und eine Bindung an 127.0.0.1 verweigerte sie. Setzen Sie zuerst einen echten Dienstschlüssel und prüfen Sie, welche Firewall-Regel Windows für den Port anlegt. Der Dienstschlüssel wandert über reines HTTP im Klartext, verwenden Sie das also nur in einem Netzwerk, dem Sie vertrauen. Die Firewall-Rückfrage wurde hier nicht beobachtet, und der Zugriff von einem anderen Rechner wurde nicht getestet.

Führen Sie einen einzigen Worker aus. Mit der Option workers auf 2 und einem Limit von 3 je Worker sah der Mock bis zu 6 gleichzeitige Lösevorgänge. Jeder Worker ist ein eigener Prozess mit seiner eigenen Semaphore, das Limit vervielfacht sich also. Die Option reload kann beim Bearbeiten gefahrlos verwendet werden. Das Berühren der Datei mitten im Lösevorgang ließ die laufende Anfrage nicht fallen, und eine neue Anfrage 1,5 Sekunden später war ebenfalls erfolgreich.

Die Dokumentationsseiten sind standardmäßig offen. Lauscht der Dienst auf einer Netzwerkadresse, erstellen Sie die App mit docs_url und openapi_url auf None, damit das Schema nicht öffentlich ist.

Betrieb des Dienstes getrennt von CapSkip

Der Dienst muss nicht auf demselben PC laufen. Schalten Sie CapSkip auf einem Windows-Rechner, den Sie kontrollieren, in den Server-Modus, möglichst mit statischer öffentlicher IP. Setzen Sie CAPSKIP_HOST und CAPSKIP_PORT auf diese Adresse, bevor Sie uvicorn starten. Es gelten drei Warnungen. Aktivieren Sie in CapSkip die Validierung des API-Schlüssels und verwenden Sie denselben Schlüssel in CAPSKIP_API_KEY. Das SDK spricht reines HTTP, daher wandert dieser Schlüssel im Klartext durchs Netzwerk. CapSkip legt außerdem eine Regel an, die für seinen Port alles zulässt, fügen Sie also für jede Adresse, die ihn nicht erreichen soll, eine Blockierregel hinzu. Der Abschnitt Verbindungseinstellungen der Einrichtungsanleitung listet beide Modi auf.

Von PowerShell aufrufen

Der Aufrufer sendet den sitekey, die Seiten-URL und den Dienstschlüssel. Die Antwort ist ein JSON-Objekt mit einem Feld, token. Verwenden Sie den Token innerhalb von etwa zwei Minuten und nur einmal. Die Anleitung zur Gültigkeit von Token erklärt diese Frist. Die Adresse unten ist ein Platzhalter für Ihren Dienst.

Invoke-RestMethod -Method Post -Uri http://192.168.1.50:8000/solve `
  -Headers @{ "X-API-Key" = "YOUR_API_KEY" } `
  -ContentType "application/json" `
  -Body '{"sitekey":"YOUR_SITEKEY","url":"https://example.com/page-with-recaptcha"}'

Ein POST an /solve/ mit einem abschließenden Schrägstrich bekommt eine 307-Weiterleitung. Invoke-RestMethod in Windows PowerShell 5.1 folgte ihr und behielt Body und Header bei. Wenn Sie curl.exe bevorzugen, übergeben Sie das JSON über stdin. Das übliche JSON-Argument in einfachen Anführungszeichen scheitert in PowerShell 5.1, weil die inneren doppelten Anführungszeichen entfernt werden, und der Dienst antwortete mit einem JSON-Fehler.

# curl.exe reads the JSON from stdin, so no quote escaping is needed.
$body = '{"sitekey":"YOUR_SITEKEY","url":"https://example.com/page-with-recaptcha"}'
$body | curl.exe -s http://192.168.1.50:8000/solve `
  -H "X-API-Key: YOUR_API_KEY" `
  -H "Content-Type: application/json" `
  --data-binary "@-"

Was die Schleuse leistet und was nicht

Das Thread-Limit des Mocks war 3, und zehn Anfragen kamen gleichzeitig an, jede mit einem Lösevorgang von 2 Sekunden. Die Schleuse hielt den Mock bei drei gleichzeitigen Lösevorgängen, und alle zehn kamen mit 200 zurück. Ohne funktionierende Schleuse erreichten alle zehn Aufgaben CapSkip und stauten sich dort. Im Mock gaben vier davon auf, nachdem sie über sein Warte-Limit von drei Sekunden hinaus gewartet hatten, und kamen als unlösbar zurück. Diese Antwort stammt aus dem eigenen Modell des Mocks und wurde auf einem echten CapSkip nicht bestätigt.

Das Warten auf einen Slot liegt aus gutem Grund innerhalb der Frist. Bei einer Frist von 5 Sekunden, einem Lösevorgang von 1 Sekunde und zehn gleichzeitigen Anfragen kamen sechs mit 200 und vier mit 504 zurück. Eine Anfrage erreichte CapSkip nie, und drei der neun, die es erreichten, wurden mitten im Lösevorgang aufgegeben.

Ein Slot wird etwas länger gehalten als der Lösevorgang selbst. Das SDK pollt sofort, wartet dann 0,25; 0,5; 1; 2 und 4 Sekunden und danach 5 Sekunden zwischen den Abfragen. Das legt die Abfragen auf 0; 0,25; 0,75; 1,75; 3,75; 7,75 und 12,75 Sekunden, und der Slot bleibt belegt, bis eine Abfrage die Antwort liest. Ein Ergebnis, das kurz nach einer Abfrage erscheint, wartet auf die nächste. Die Abfrage bei 7,75 s verfehlt einen Lösevorgang von 8 Sekunden um eine Viertelsekunde. Auf dem Test-PC schoben die langsamen Aufrufe aus dem nächsten Abschnitt diese Abfrage über 8 s hinaus, sodass die Antwort nach etwa 8,4 bis 8,9 s gelesen wurde. Ohne diese Kosten wartete derselbe Lösevorgang auf die Abfrage bei 12,75 s und kam mit dem Standard-pollingInterval von 5 nach 12,9 s zurück, mit einem pollingInterval von 1 dagegen nach 8,9 s. Sind die Lösevorgänge lang und die Warteschlange lang, gibt ein niedrigeres pollingInterval Slots früher frei.

Zwei Grenzen bleiben. Ein Aufrufer, der die Verbindung beendet, bricht seinen Handler nicht ab, sodass der verwaiste Lösevorgang seinen Slot behält, bis er endet. Mit einem Slot und einem Lösevorgang von 4 Sekunden brauchte die nächste Anfrage insgesamt 8,3 Sekunden, gegenüber 4,7 Sekunden mit einem freien Slot. Der Mock arbeitet außerdem nach Ablauf der Frist an einer aufgegebenen Aufgabe weiter, und das SDK hat keinen Abbruch-Aufruf. Die Zahl der Slots kann der Zahl der echten Threads von CapSkip daher einige Sekunden vorauslaufen. Das ist im Mock nachgebildet. Die echte App wurde nicht getestet.

Ein versteckter Preis

AsyncCapSkip baut für jede Anfrage, die es sendet, einen neuen httpx-Client, jede Abfrage eingeschlossen. Auf dem Test-PC kostete das jedes Mal etwa 141 ms blockierende CPU-Zeit, hauptsächlich fürs Laden des Zertifikatspakets, das eine einfache HTTP-Adresse nie braucht. Zehn gleichzeitige Lösevorgänge von 2 Sekunden waren nach etwa 6 s fertig, und eine während der Welle gesendete Anfrage an /openapi.json wartete bis zu etwa 1,9 s. Dieselben zehn über einen synchronen Endpunkt brauchten 3,8 s mit einem schlechtesten Fall von wenigen Millisekunden, aber dieser Weg kann einen hängenden Aufruf nicht beenden. Eine Frist um einen Thread lässt den Thread festhängen. Nach 41 hängenden Anfragen hielt der Dienst 41 festhängende Threads, und sie bleiben bis zum Neustart des Prozesses.

Die blockierte Event-Loop lässt unter Last auch die Uhren abweichen. Gegen ein stummes CapSkip mit einer Frist von 2 Sekunden ergaben 12 gleichzeitige Anfragen 8 Antworten nach 2,6 bis 2,9 s und 4 Antworten mit dem Fehlernamen capskip_timeout nach 8,5 s. Bei 41 Anfragen brauchte die langsamste 13,9 s. Eine einzelne Anfrage hielt ihre Frist ein. Die Ursache der Antworten nach 8,5 s wurde nicht isoliert, aber die blockierte Event-Loop gehört dazu, und das SDK allein zeigte dieselbe Abweichung. Für eine Handvoll Aufrufer ist der Stillstand harmlos. Er wächst mit der Zahl der laufenden Lösevorgänge, halten Sie die Zahl der Slots also bei dem, was CapSkip wirklich ausführen kann.

Die vollständige main.py

Das sind die Schritte 3 bis 10, zusammengefügt, mit den Importen gesammelt oben. Speichern Sie sie als main.py.

import asyncio
import logging
import os
import secrets
from contextlib import asynccontextmanager

from capskip import ApiException, AsyncCapSkip, CapSkipError
from capskip import NetworkException, TimeoutException, ValidationException
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

# The SDK reads no env vars, so pass each setting in by hand.
HOST = os.getenv("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.getenv("CAPSKIP_PORT", "8080"))
KEY = os.getenv("CAPSKIP_API_KEY", "capskip")
MAX_THREADS = int(os.getenv("CAPSKIP_MAX_THREADS", "10"))
DEADLINE = float(os.getenv("SOLVE_DEADLINE", "120"))


@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client and one gate, shared by every request.
    app.state.cap = AsyncCapSkip(
        apiKey=KEY, host=HOST, port=PORT,
        recaptchaTimeout=DEADLINE,
    )
    app.state.slots = asyncio.Semaphore(MAX_THREADS)
    yield


app = FastAPI(lifespan=lifespan)

SERVICE_KEY = os.environ["SOLVER_API_KEY"]
if not SERVICE_KEY:
    raise SystemExit("SOLVER_API_KEY must not be empty")


def require_key(x_api_key: str = Header(default="")):
    good = secrets.compare_digest(
        x_api_key.encode(), SERVICE_KEY.encode())
    if not good:
        raise HTTPException(
            401, "missing or wrong X-API-Key",
            headers={"WWW-Authenticate": "APIKey"})


class SolveRequest(BaseModel):
    sitekey: str = Field(min_length=1)
    url: str = Field(pattern=r"^https?://")


async def solve_v2(state, sitekey: str, url: str) -> str:
    # The deadline also covers the wait for a free slot.
    async with asyncio.timeout(DEADLINE):
        async with state.slots:
            result = await state.cap.recaptcha(sitekey, url)
    return result["code"]


@app.post("/solve", dependencies=[Depends(require_key)])
async def solve(body: SolveRequest, request: Request):
    token = await solve_v2(request.app.state, body.sitekey, body.url)
    return {"token": token}


STATUS = {
    NetworkException: (503, "capskip_unreachable"),
    TimeoutException: (504, "capskip_timeout"),
    ValidationException: (422, "bad_parameters"),
    ApiException: (502, "capskip_error"),
}


@app.exception_handler(CapSkipError)
async def capskip_error(request: Request, exc: CapSkipError):
    # Log the detail here, keep it away from the caller.
    logging.getLogger("uvicorn.error").warning("%r", exc)
    status, code = STATUS.get(type(exc), (502, "capskip_error"))
    if "UNSOLVABLE" in str(exc):
        status, code = 422, "unsolvable"
    return JSONResponse({"error": code}, status)


@app.exception_handler(TimeoutError)
async def deadline_hit(request: Request, exc: TimeoutError):
    # asyncio.timeout raises the built-in TimeoutError. The SDK's
    # TimeoutException is a different class, so it needs its own entry.
    return JSONResponse({"error": "deadline_exceeded"}, 504)


@app.get("/health")
async def health():
    # CapSkip has no ping call, so check that its port accepts a
    # connection.
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(HOST, PORT), timeout=3)
    except (OSError, TimeoutError):
        return JSONResponse({"ok": False}, 503)
    writer.close()
    return {"ok": True}

Häufige Fehler bei einem FastAPI-Captcha-Solver

FastAPIs eigene lifespan-Dokumentation behandelt den Startup-Hook, und jeder CapSkip-Fehlercode steht in der API-Dokumentation.

FAQ

Die Kurzfassung

Um einen FastAPI-Captcha-Solver-Dienst zu betreiben, verwenden Sie einen AsyncCapSkip, eine Semaphore und ein asyncio.timeout um das Warten und den Lösevorgang. Prüfen Sie zuerst den Dienstschlüssel. Fangen Sie die TimeoutException des SDK getrennt vom eingebauten TimeoutError ab. Führen Sie einen einzigen Worker aus. Setzen Sie eine Blockierregel vor CapSkip, wenn es im Server-Modus läuft.

• Alle Captcha-Typen, die das Python-Paket verarbeitet, finden Sie auf der Python-Captcha-Solver-Seite.
• Wie der Checkbox-Solver funktioniert: die reCAPTCHA-v2-Solver-Seite.
• Viele Lösungen in einem Skript: die Anleitung zum parallelen Lösen von Captchas in Python.

Ein gemeinsam genutzter Dienst kann ein ganzes Team bedienen, ohne dass jemand Lösungen zählt. CapSkip übernimmt die Captcha-Umgehung auf Ihrem eigenen Windows-Rechner, sodass jede Lösung hinter dem Dienst frei von Kosten pro Aufruf bleibt.