如何在 Prefect flow 中识别验证码并配置重试

# 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

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

# 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"]

# 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

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

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

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