如何在 Huey 任务中识别验证码：在 Windows 上使用 SQLite

# Windows, in PowerShell or cmd: a virtual environment in your project folder
python -m venv .venv
.venv\Scripts\python -m pip install huey capskip requests

# tasks.py
import os
from pathlib import Path

from capskip import CapSkip
from huey import SqliteHuey

# The whole queue is one SQLite file. Keep it next to this module, so the
# consumer and every script that imports tasks use the same file, whatever
# folder they are started from.
huey = SqliteHuey(
    "captcha",
    filename=str(Path(__file__).with_name("huey.db")),
    store_intermediate_errors=False,  # get() waits for the last retry
)

# Local mode: CapSkip on this PC. Server mode: set CAPSKIP_HOST (and
# CAPSKIP_API_KEY if key validation is on) to reach it on another machine.
solver = CapSkip(
    apiKey=os.environ.get("CAPSKIP_API_KEY", "capskip"),
    host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.environ.get("CAPSKIP_PORT", "8080")),
)

from capskip import ApiException, ValidationException
from huey import CancelExecution


# Any exception uses one of the two retries, 15 s apart: NetworkException
# when CapSkip is not running, TimeoutException, an unsolvable task.
@huey.task(retries=2, retry_delay=15)
def solve_recaptcha(sitekey, page_url):
    try:
        return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]
    except (ApiException, ValidationException) as exc:
        if "ERROR_CAPTCHA_UNSOLVABLE" in str(exc):
            raise  # this attempt failed; a fresh one may not
        # ERROR_GOOGLEKEY, ERROR_PAGEURL, ERROR_BAD_PARAMETERS and
        # ERROR_KEY_DOES_NOT_EXIST fail the same way on every attempt.
        # CancelExecution's first argument is retry: False ends the task now.
        raise CancelExecution(False, f"{type(exc).__name__}: {exc}") from exc

# In the folder that holds tasks.py, in PowerShell or cmd
.venv\Scripts\huey_consumer tasks.huey -w 4

# enqueue.py, next to tasks.py: .venv\Scripts\python enqueue.py
from huey.exceptions import ResultTimeout, TaskException

from tasks import huey, solve_recaptcha

result = solve_recaptcha("YOUR_SITEKEY", "https://example.com/login")
try:
    # Three attempts of up to about 5.5 minutes each, plus two 15 s delays.
    token = result.get(blocking=True, timeout=1100)
except TaskException as exc:
    print("solve failed:", exc)
except ResultTimeout:
    print("no result yet: is the consumer running?")
else:
    print("token:", token[:30], "...")
    # Reading the result deleted it from huey.db. This handle keeps its
    # copy, but a new handle for the same task id finds nothing.
    print("read again:", huey.result(result.id))  # None

import functools
import types

import requests
import capskip.api

# capskip 1.3.0 calls requests.post/get with no timeout, and Huey cannot stop
# a thread worker that is stuck in one. Give every SDK call 5 s to connect
# and 30 s to answer. Module level in tasks.py, so it runs before any solve.
capskip.api.requests = types.SimpleNamespace(
    RequestException=requests.RequestException,
    post=functools.partial(requests.post, timeout=(5, 30)),
    get=functools.partial(requests.get, timeout=(5, 30)),
)

from huey import crontab


# Every 15 minutes. The lock goes under periodic_task: when a run is due
# while the previous one is still solving, another worker skips it instead
# of starting a second solve for the same login.
@huey.periodic_task(crontab(minute="*/15"))
@huey.lock_task("login")
def refresh_login():
    result = solver.recaptcha(sitekey="YOUR_SITEKEY", url="https://example.com/login")
    print("new token:", result["code"][:30], "...")

# tasks.py: pip install huey capskip requests
# Start: .venv\Scripts\huey_consumer tasks.huey -w 4 -f
import functools
import logging
import os
import re
import types
from pathlib import Path

import requests
import capskip.api
from capskip import ApiException, CapSkip, ValidationException
from huey import CancelExecution, SqliteHuey, crontab

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

huey = SqliteHuey(
    "captcha",
    filename=str(Path(__file__).with_name("huey.db")),  # same file from any folder
    store_intermediate_errors=False,  # get() waits for the last retry
)
log = logging.getLogger("huey.captcha")  # printed by the consumer

# The SDK's HTTP calls have no timeout; give them one so a stuck CapSkip
# cannot hold a worker thread forever.
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(
    apiKey=os.environ.get("CAPSKIP_API_KEY", "capskip"),
    host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.environ.get("CAPSKIP_PORT", "8080")),
)


def solve(sitekey, page_url):
    try:
        return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]
    except (ApiException, ValidationException) as exc:
        if "ERROR_CAPTCHA_UNSOLVABLE" in str(exc):
            raise  # worth another attempt
        # A refused sitekey, page URL or API key fails the same way every time.
        log.error("not retrying: %s: %s", type(exc).__name__, exc)
        raise CancelExecution(False, f"{type(exc).__name__}: {exc}") from exc


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="([^"]+)"')
        csrf = field(page.text, r'name="csrfmiddlewaretoken" value="([^"]+)"')
        token = solve(sitekey, LOGIN_URL)  # solve right before the post
        reply = session.post(LOGIN_URL, timeout=30, headers={"Referer": LOGIN_URL}, data={
            "csrfmiddlewaretoken": csrf,
            "username": USERNAME,
            "password": PASSWORD,
            "g-recaptcha-response": token,
        })
        reply.raise_for_status()  # a refused login fails the attempt
        log.info("logged in (HTTP %s)", reply.status_code)
        return reply.status_code


# Both tasks take the same lock, so an on-demand login never runs while a
# scheduled one is solving, and the other way round.

# On demand: from tasks import log_in; log_in().get(blocking=True, timeout=1100)
@huey.task(retries=2, retry_delay=15)  # a locked attempt uses a retry
@huey.lock_task("login")
def log_in():
    return login()


# On a schedule; a run that comes due while the lock is held is skipped.
@huey.periodic_task(crontab(minute="*/15"))
@huey.lock_task("login")
def scheduled_login():
    login()

# PowerShell, before starting the consumer: CapSkip runs in Server mode
# on another Windows machine on your network
$env:CAPSKIP_HOST = "192.168.1.50"
$env:CAPSKIP_API_KEY = "YOUR_API_KEY"
# $env:CAPSKIP_PORT = "8080"   only if you changed the port
.venv\Scripts\huey_consumer tasks.huey -w 4 -f