Captchas in APScheduler lösen, ohne dass Durchläufe übersprungen werden

Ein APScheduler-Captcha-Job ist eine gewöhnliche Funktion, die CapSkip aufruft und mit einem Intervall-Trigger eingeplant wird. Der Lösevorgang selbst ist eine Zeile. Das Problem sind die Standardwerte drumherum. APScheduler führt immer nur eine Instanz eines Jobs gleichzeitig aus. Der synchrone Client des Python-SDK von CapSkip setzt für seine HTTP-Aufrufe kein Timeout. Zusammen bedeutet das: Ein einziger Lösevorgang, der nie zurückkehrt, blockiert jeden späteren Durchlauf, und das einzige Anzeichen ist eine Warnung alle zehn Minuten. Ein Durchlauf, der mehr als eine Sekunde zu spät startet, wird verworfen. Und unter Windows mit Python 3.13 oder älter bewirkt Ctrl+C nichts, bis der nächste Durchlauf fällig ist. Dieser Leitfaden behebt alle drei Probleme in einem Job, der sich alle zehn Minuten bei einer Website hinter reCAPTCHA anmeldet.

Was Sie brauchen

• CapSkip auf einem Windows-Rechner, mit eingeschaltetem reCAPTCHA-Löser.
• Python 3.10 oder neuer. Der Code unten wurde unter Python 3.13.7 und 3.14.0 ausgeführt.
• APScheduler 3.11, die aktuelle stabile Versionslinie, dazu die Pakete capskip und requests. Pinnen Sie APScheduler auf eine Version unter 4: Die Vorabversionen von 4.0 haben eine andere API, und die Importe unten scheitern daran.
• Eine Adresse für den Solver. Im Local-Modus antwortet CapSkip auf 127.0.0.1 nur für dieses Gerät. Im Server-Modus lauscht es auf Ihrer Netzwerkadresse oder öffentlichen IP, damit ein Scheduler auf einem anderen Rechner es über die API aufrufen kann. Eingestellt werden beide Modi unter Verbindungseinstellungen.

Schritt 1: Ein Job, der das Captcha löst

Ein APScheduler-Job ist eine beliebige Funktion. Dieser hier fordert alle zehn Minuten einen reCAPTCHA-Token bei CapSkip an. Schreiben Sie 127.0.0.1 statt localhost: Unter Windows versucht localhost zuerst die IPv6-Adresse, und in unseren Durchläufen verdoppelte das die Zeit, bis ein nicht laufender Solver gemeldet wurde, von 2 auf 4 Sekunden.

# pip install "apscheduler<4" capskip
from apscheduler.schedulers.blocking import BlockingScheduler
from capskip import CapSkip

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


def solve_login_captcha():
    result = solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")
    print("token:", result["code"][:20], "...")


scheduler = BlockingScheduler()
scheduler.add_job(solve_login_captcha, "interval", minutes=10)
scheduler.start()

BlockingScheduler läuft im Haupt-Thread und schickt jeden Durchlauf an einen Pool aus 10 Worker-Threads. Der erste Durchlauf findet ein Intervall nach dem Start statt, nicht beim Start. Das ist in Ordnung, bis sich ein Lösevorgang danebenbenimmt, und genau darum geht es im Rest dieses Leitfadens. Die kurzen Beispiele verwenden BlockingScheduler, damit sie kurz bleiben. Unter Windows mit Python 3.13 oder älter verwenden Sie stattdessen die Schleife aus dem vollständigen Skript unten.

Warum ein langsamer Lösevorgang den nächsten Durchlauf überspringt

Bei jedem Job ist max_instances standardmäßig auf 1 gesetzt. Wenn ein Durchlauf fällig ist und der vorherige noch läuft, startet APScheduler keinen zweiten. Stattdessen loggt es eine Warnung, auf Level WARNING unter dem Logger apscheduler.scheduler:

Execution of job "solve_login_captcha (trigger: interval[0:10:00], next run at: 2026-10-07 13:10:00 +08)" skipped: maximum number of running instances reached (1)

Für einen Captcha-Job ist dieser Standard richtig. Sie wollen nie zwei Lösevorgänge für denselben Login gleichzeitig. Ein Lösevorgang, der etwas länger dauert als das Intervall, kostet einen übersprungenen Durchlauf, und der Job macht zum nächsten geplanten Zeitpunkt weiter. In einem Test mit einem Intervall von 5 Sekunden und einem Lösevorgang von 12,8 Sekunden wurden pro Lösevorgang zwei Durchläufe übersprungen, und der nächste Durchlauf startete zum darauffolgenden geplanten Zeitpunkt, nicht direkt nach dem Lösevorgang.

Der Ärger beginnt, wenn ein Lösevorgang nie fertig wird. Dann bleibt die eine erlaubte Instanz für immer belegt, und jeder spätere Durchlauf wird mit derselben Warnung übersprungen. Der Job sieht eingeplant aus, das Log sieht lebendig aus, und nichts passiert.

Schritt 2: Dem SDK ein HTTP-Timeout geben

Das Python-SDK von CapSkip ruft requests.post und requests.get ohne Timeout auf. Wenn CapSkip eine Verbindung annimmt und nie antwortet, zum Beispiel weil die App eingefroren ist, wartet der Aufruf für immer. Das recaptchaTimeout des SDK, standardmäßig 300 Sekunden, hilft nicht, weil es nur zwischen zwei Abfragen geprüft wird, und eine hängende Abfrage endet nie. In unserem Test blieb ein so festgefahrener Job auch dann hängen, als sich der Solver wieder erholt hatte, und loggte nur noch Warnungen über übersprungene Durchläufe.

Zwei naheliegende Lösungen funktionieren nicht. socket.setdefaulttimeout hat keine Wirkung, weil requests immer seinen eigenen Timeout-Wert weiterreicht, selbst wenn dieser Wert None ist. Und das SDK nimmt kein Session-Objekt an, das Sie konfigurieren könnten. Was funktioniert: Geben Sie dem Modul des SDK einen kleinen Ersatz für das Paket requests, dessen Aufrufe post und get immer ein Timeout übergeben.

# pip install capskip requests
import functools
import types

import requests
import capskip.api
from capskip import CapSkip


# The SDK calls requests.post/get with no timeout. Hand its module a
# stand-in whose calls always pass one: 5 s to connect, 30 s to answer.
capskip.api.requests = types.SimpleNamespace(
    RequestException=requests.RequestException,
    post=functools.partial(requests.post, timeout=(5, 30)),
    get=functools.partial(requests.get, timeout=(5, 30)),
)  # do this before the first solve
solver = CapSkip(host="127.0.0.1", port=8080)

Das ersetzt den Namen requests innerhalb von capskip.api, dem Modul, das in capskip 1.3.0 die Anfragen des SDK an CapSkip sendet. Prüfen Sie das nach einem Upgrade des SDK erneut. Führen Sie es einmal aus, vor dem ersten Lösevorgang, dann bekommt jeder Aufruf 5 Sekunden für den Verbindungsaufbau und 30 Sekunden für die Antwort.

Die beiden Arten von Hängern enden dann unterschiedlich. Hört CapSkip auf zu antworten, während die Aufgabe übermittelt wird, scheitert der Job nach 30 Sekunden mit einer NetworkException mit der Meldung read timed out. Hört es auf zu antworten, während das SDK das Ergebnis abfragt, behandelt das SDK das Timeout als noch nicht bereit und pollt weiter, sodass der Job bei recaptchaTimeout mit einer TimeoutException endet. Mit den Standardwerten dauerte das in unserem Test 333 Sekunden. Der schlimmste Fall liegt also bei etwa fünfeinhalb Minuten, was in ein Intervall von zehn Minuten passt, und der nächste Durchlauf löst normal.

Schritt 3: Festlegen, was mit verspäteten Durchläufen passiert

Der zweite Standardwert ist misfire_grace_time, nämlich 1 Sekunde. Ein Durchlauf, der mehr als eine Sekunde nach seinem geplanten Zeitpunkt startet, wird verworfen, und APScheduler loggt eine Warnung, dass der Ausführungszeitpunkt verpasst wurde, samt Verzögerung. Durchläufe starten öfter zu spät, als man denkt. Ein Pool, dessen Threads alle mit Lösen beschäftigt sind, oder ein Laptop, der einen Durchlauf verschlafen hat, reicht schon.

# pip install "apscheduler<4" capskip
from datetime import datetime

from apscheduler.schedulers.blocking import BlockingScheduler
from capskip import CapSkip

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


def solve_login_captcha():
    result = solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")
    print("token:", result["code"][:20])


scheduler = BlockingScheduler()
scheduler.add_job(
    solve_login_captcha, "interval", minutes=10,
    next_run_time=datetime.now(),  # first run now, not ten minutes from now
    jitter=30,                     # add 0 to 30 s to each run
    max_instances=1,               # the default: a slow solve skips the next run
    coalesce=True,                 # the default: after a stall, run once
    misfire_grace_time=None,       # default is 1 s; None runs late instead of dropping
)
scheduler.start()

Setzen Sie misfire_grace_time auf None, dann findet ein verspäteter Durchlauf trotzdem statt, eben verspätet. Lassen Sie die Option coalesce auf ihrem Standardwert True, damit nach einem langen Stillstand mehrere verpasste Durchläufe zu einem einzigen zusammenfallen, statt nacheinander zu laufen. Wir haben einen Scheduler bei einem Intervall von 4 Sekunden für 14 Sekunden angehalten, um einen Computer im Energiesparmodus zu simulieren. Mit den Standardwerten wurde der Nachholdurchlauf verworfen. Mit misfire_grace_time auf None lief er einmal. Mit zusätzlich auf False gesetzter Option coalesce lief er dreimal hintereinander: drei Captcha-Lösevorgänge für einen einzigen Login.

Übergeben Sie next_run_time, um den Job beim Start sofort auszuführen. Die Option jitter fügt jedem Durchlauf eine zufällige Verzögerung von bis zu so vielen Sekunden hinzu, was die Anfragen verteilt. Jede Verzögerung wird auf den vorherigen Ausführungszeitpunkt aufgeschlagen, der bereits eine Verzögerung enthält, daher beträgt der durchschnittliche Abstand das Intervall plus die Hälfte des Jitters.

Schritt 4: Fehlschläge sehen, sobald sie passieren

Eine Exception, die in einem Job ausgelöst wird, stoppt den Scheduler nicht. APScheduler loggt sie auf Level ERROR, mit einer Meldung, dass der Job eine Exception ausgelöst hat, und dem vollständigen Traceback, und der Job behält seinen Zeitplan. Ein Listener erhält dieselben Events als Objekte, sodass Sie sie zählen, bei ihnen Alarm auslösen oder sie dorthin schreiben können, wo Sie sie haben wollen.

# pip install "apscheduler<4" capskip
import logging

from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MAX_INSTANCES, EVENT_JOB_MISSED
from apscheduler.schedulers.blocking import BlockingScheduler
from capskip import CapSkip

logging.basicConfig(level=logging.INFO)
solver = CapSkip(host="127.0.0.1", port=8080)


def solve_login_captcha():
    solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")


def on_event(event):
    if event.code == EVENT_JOB_ERROR:
        logging.error("job %s failed: %r", event.job_id, event.exception)
    elif event.code == EVENT_JOB_MAX_INSTANCES:
        logging.warning("job %s still solving, this run skipped", event.job_id)
    else:
        logging.warning("job %s started too late, this run dropped", event.job_id)


scheduler = BlockingScheduler()
scheduler.add_listener(on_event, EVENT_JOB_ERROR | EVENT_JOB_MAX_INSTANCES | EVENT_JOB_MISSED)
scheduler.add_job(solve_login_captcha, "interval", minutes=10, id="login")
scheduler.start()

Drei Events decken einen Captcha-Job ab. EVENT_JOB_ERROR trägt die Exception selbst, etwa eine ApiException mit der Meldung ERROR_CAPTCHA_UNSOLVABLE. EVENT_JOB_MAX_INSTANCES feuert einmal für jeden übersprungenen Durchlauf, wodurch ein festgefahrener Lösevorgang als steigender Zähler sichtbar wird. EVENT_JOB_MISSED feuert für jeden verworfenen verspäteten Durchlauf.

Der vollständige APScheduler-Captcha-Job

Dieses Skript ruft die Login-Seite mit einer requests-Session ab, liest den sitekey und das CSRF-Feld aus, löst das reCAPTCHA mit CapSkip und sendet das Formular mit dem Token ab. Es fängt die Fehler von CapSkip innerhalb des Jobs ab, sodass ein gescheiterter Lösevorgang nur eine Warnung ist und der nächste Durchlauf es erneut versucht.

# pip install "apscheduler<4" capskip requests
import functools
import logging
import re
import time
import types
from datetime import datetime

import requests
import capskip.api
from apscheduler.schedulers.background import BackgroundScheduler
from capskip import CapSkip, CapSkipError

LOGIN_URL = "https://example.com/login"
USERNAME, PASSWORD = "YOUR_USERNAME", "YOUR_PASSWORD"

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
log = logging.getLogger("login-job")


# The SDK's HTTP calls have no timeout; give them one so a stuck CapSkip
# cannot hang the job.
capskip.api.requests = types.SimpleNamespace(
    RequestException=requests.RequestException,
    post=functools.partial(requests.post, timeout=(5, 30)),
    get=functools.partial(requests.get, timeout=(5, 30)),
)
solver = CapSkip(host="127.0.0.1", port=8080)


def field(html, pattern):
    match = re.search(pattern, html)
    if not match:
        raise ValueError(f"not found on the login page: {pattern}")
    return match.group(1)


def login():
    with requests.Session() as session:
        page = session.get(LOGIN_URL, timeout=30)
        sitekey = field(page.text, r'data-sitekey="([^"]+)"')
        # The form's hidden CSRF field (Django names it csrfmiddlewaretoken).
        csrf = field(page.text, r'name="csrfmiddlewaretoken" value="([^"]+)"')
        try:
            token = solver.recaptcha(sitekey=sitekey, url=LOGIN_URL)["code"]
        except CapSkipError as exc:
            log.warning("CapSkip could not solve: %s: %s", type(exc).__name__, exc)
            return
        reply = session.post(LOGIN_URL, timeout=30, headers={"Referer": LOGIN_URL}, data={
            "csrfmiddlewaretoken": csrf,
            "username": USERNAME,
            "password": PASSWORD,
            "g-recaptcha-response": token,
        })
        # Many sites answer a failed login with 200 and the form again;
        # check for something only a logged-in page shows.
        if reply.ok:
            log.info("logged in (HTTP %s)", reply.status_code)
        else:
            log.warning("login refused (HTTP %s): %s", reply.status_code, reply.text[:200])


scheduler = BackgroundScheduler()
scheduler.add_job(
    login, "interval", minutes=10,
    next_run_time=datetime.now(),  # first run now, not ten minutes from now
    max_instances=1,               # never two solves for the same account at once
    coalesce=True,                 # after a stall, run once, not once per missed slot
    misfire_grace_time=None,       # a late run still runs instead of being dropped
)
scheduler.start()
try:
    while True:
        time.sleep(1)  # Ctrl+C lands here at once, on any Python version
except KeyboardInterrupt:
    log.info("stopping; a solve in flight finishes first")
    scheduler.shutdown()

Es verwendet BackgroundScheduler und eine Sleep-Schleife statt BlockingScheduler, aus einem Grund, der nur unter Windows auftritt. Mit BlockingScheduler unter Python 3.13 oder älter wartet der Haupt-Thread auf einen Lock, den Ctrl+C nicht unterbrechen kann, sodass der KeyboardInterrupt erst beim nächsten geplanten Aufwachen ankommt. Bei einem Intervall von zehn Minuten können das zehn Minuten sein. Python 3.14 hat das unter Windows behoben, aber die Schleife funktioniert mit jeder Version: Ctrl+C unterbricht time.sleep sofort.

Das Stoppen wartet trotzdem auf einen laufenden Lösevorgang. Bevor der Prozess endet, wartet Python, bis die Worker-Threads des Pools fertig sind, ein laufender Lösevorgang wird also zuerst abgeschlossen. Das passiert unabhängig davon, ob Sie scheduler.shutdown() aufrufen, und mit dem Timeout aus Schritt 2 dauert das Warten höchstens etwa fünfeinhalb Minuten. Ein zweites Ctrl+C verkürzt es nicht.

Die Session sendet das Formular mit einem Referer-Header ab, weil requests von sich aus keinen sendet und Frameworks wie Django einen HTTPS-Formular-POST ohne diesen Header ablehnen. Die Session behält außerdem das Cookie, das zum CSRF-Feld passt. Lösen Sie direkt vor dem Absenden: Ein reCAPTCHA-Token läuft zwei Minuten nach seiner Ausgabe ab. Unsichtbare Widgets, Enterprise-Schlüssel und die anderen Varianten behandelt der Leitfaden zum Lösen von reCAPTCHA v2 in Python.

Viele Konten mit einem Zeitplan

Ein Job pro Konto funktioniert gut, mit einem gemeinsamen CapSkip-Client, der sich gefahrlos aus mehreren Threads verwenden lässt. Machen Sie den Thread-Pool so groß wie die Zahl der Konten.

# pip install "apscheduler<4" capskip
import logging
from datetime import datetime

from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.schedulers.blocking import BlockingScheduler
from capskip import CapSkip

logging.basicConfig(level=logging.INFO)
solver = CapSkip(host="127.0.0.1", port=8080)  # one client is safe to share across threads
ACCOUNTS = [f"user{i:02d}" for i in range(20)]


def login(account):
    token = solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")["code"]
    logging.info("%s got a token %s", account, token[:12])


scheduler = BlockingScheduler(executors={"default": ThreadPoolExecutor(max_workers=20)})
for account in ACCOUNTS:
    scheduler.add_job(login, "interval", minutes=10, args=[account], id=account,
                      next_run_time=datetime.now(), misfire_grace_time=None)
scheduler.start()

Der Standard-Pool hat 10 Threads. Mit 20 gleichzeitig fälligen Jobs warteten 10 Durchläufe auf einen Thread, starteten 7,8 Sekunden zu spät und wurden bei der standardmäßigen Toleranz von 1 Sekunde verworfen. Die Hälfte der Konten hat sich nie angemeldet, und nur die Warnungen über verpasste Durchläufe haben es verraten. Ein Pool mit 20 Threads oder misfire_grace_time auf None behebt das.

CapSkip hat ein eigenes Limit. Seine reCAPTCHA-Einstellungen erlauben standardmäßig 10 gleichzeitige Lösevorgänge, unter Max. Threads, und spätere Aufgaben warten auf einen freien Thread. In unserem Test mit 20 Threads dauerten 10 Lösevorgänge 7,8 Sekunden und die anderen 10 dauerten 12,8 Sekunden, weil sie 5 Sekunden auf CapSkip warteten. Erhöhen Sie Max. Threads in CapSkip, wenn Ihr Rechner die Kapazität hat, und verteilen Sie die Konten mit jitter, wenn nicht.

Mit asyncio statt Threads

Wenn der Rest Ihres Codes asynchron ist, verwenden Sie AsyncIOScheduler mit AsyncCapSkip, dem nativen asynchronen Client des SDK. Coroutine-Jobs laufen in der Event-Loop, und max_instances funktioniert genau wie vorher.

# pip install "apscheduler<4" capskip
import asyncio

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from capskip import AsyncCapSkip

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


async def solve_login_captcha():
    result = await solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")
    print("token:", result["code"][:20], "...")


async def main():
    scheduler = AsyncIOScheduler()
    scheduler.add_job(solve_login_captcha, "interval", minutes=10)
    scheduler.start()
    await asyncio.Event().wait()  # keep the event loop running


asyncio.run(main())

Zwei Dinge sind anders. AsyncCapSkip basiert auf httpx, das ein Standard-Timeout von 5 Sekunden hat, daher endet ein hängendes CapSkip auch ohne den Patch aus Schritt 2 mit einer Exception: nach 5 Sekunden, wenn das Einreichen hängt, und bei recaptchaTimeout, wenn eine Abfrage hängt. Eine NetworkException daraus kann eine leere Meldung haben, loggen Sie also auch den Typ der Exception. Und das Stoppen bricht einen laufenden Lösevorgang ab, statt auf ihn zu warten: Der Job loggt dann einen CancelledError als seine Exception. Der Traceback zeigt darüber außerdem eine NetworkException, die nur das Signal des SDK für eine noch nicht bereite Antwort ist, kein Netzwerkfehler. Ctrl+C beendet das Programm in Sekundenbruchteilen.

CapSkip auf einem anderen Rechner betreiben

Der Scheduler und der Solver müssen sich keinen PC teilen. Schalten Sie CapSkip in den Verbindungseinstellungen in den Server-Modus, dann lauscht es auf Ihrer Netzwerkadresse oder öffentlichen IP. Richten Sie den Client anschließend auf diese Adresse.

# Replace the solver line with this when CapSkip runs in Server mode
# on another Windows machine on your network.
solver = CapSkip(apiKey="YOUR_API_KEY", host="192.168.1.50", port=8080)

Verwenden Sie eine statische öffentliche IP, wenn der Weg über das Internet führt, und aktivieren Sie die Validierung des API-Schlüssels, mit demselben Schlüssel im Client. CapSkip öffnet seinen Port beim Start des Server-Modus für jede Adresse. Um den Zugriff auf die Rechner zu beschränken, die Sie erwarten, fügen Sie in der Windows Firewall eine Blockierregel für alle anderen hinzu. Der Solver bleibt Ihr eigener Windows-Rechner, und abgerechnet wird pro Lösung weiterhin nichts.

Häufige Captcha-Fehler mit APScheduler

FAQ

Die Kurzfassung

Ein APScheduler-Captcha-Job braucht mehr als den Lösevorgang. Geben Sie den HTTP-Aufrufen des SDK ein Timeout, denn mit max_instances auf 1 überspringt ein einziger festgefahrener Lösevorgang jeden späteren Durchlauf. Setzen Sie misfire_grace_time auf None und lassen Sie coalesce eingeschaltet, damit verspätete Durchläufe einmal stattfinden, statt verworfen zu werden. Dimensionieren Sie den Thread-Pool nach Ihren Jobs, fangen Sie CapSkipError innerhalb des Jobs ab und verwenden Sie BackgroundScheduler mit einer Sleep-Schleife, wenn Sie den Scheduler unter Windows mit Python 3.13 oder älter mit Ctrl+C stoppen. Das Benutzerhandbuch von APScheduler 3.x dokumentiert jede hier verwendete Option.

• Alle Captcha-Typen, die das Python-Paket verarbeitet, finden Sie auf der Python-Captcha-Solver-Seite.
• Wie das Checkbox-reCAPTCHA funktioniert und was der Solver abdeckt: die reCAPTCHA-v2-Solver-Seite.

Ein Job, der alle zehn Minuten läuft, kommt auf über viertausend Lösungen im Monat. Wenn Sie sie über einen Captcha-Löser auf Ihrem eigenen Windows-Rechner laufen lassen, wird aus dieser Zahl nie eine Rechnung.