如何在 APScheduler 中识别验证码，不再跳过运行

# 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()

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)

# 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)

# 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()

# 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()

# 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()

# 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()

# 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())

# 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)