如何在 Crawl4AI 中识别验证码而不丢失会话

# pip install crawl4ai
pip install -U crawl4ai capskip

# Downloads the browser Crawl4AI drives, once per machine.
crawl4ai-setup

# pip install crawl4ai
import re
from crawl4ai import CrawlerRunConfig

PAGE_URL = "https://example.com/signup"
SESSION = "signup"
# The widget element, in any attribute order, with or without other classes.
WIDGET = r'<[^>]*class="(?:[^"]*\s)?g-recaptcha(?:\s[^"]*)?"[^>]*>'

async def read_sitekey(crawler):
    # session_id keeps this tab open for the next arun call.
    config = CrawlerRunConfig(session_id=SESSION)
    first = await crawler.arun(PAGE_URL, config=config)
    # Do not stop on first.success: a CAPTCHA page can be
    # flagged as blocked while its HTML is complete.
    widget = re.search(WIDGET, first.html)
    match = widget and re.search(r'data-sitekey="([^"]+)"', widget.group(0))
    return match.group(1) if match else None

# pip install capskip
from capskip import AsyncCapSkip

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

async def solve(sitekey):
    # Invisible v2 takes invisible=1, Enterprise takes enterprise=1.
    result = await solver.recaptcha(sitekey=sitekey, url=PAGE_URL)
    return result["code"]   # the g-recaptcha-response value

import json

async def submit(crawler, token):
    # Crawl4AI wraps this in an async function, so statements work.
    inject = (
        "document.getElementById('g-recaptcha-response').value = "
        f"{json.dumps(token)};"
        "document.querySelector('form').submit();"
    )
    config = CrawlerRunConfig(
        session_id=SESSION,
        js_only=True,                 # same tab, no reload
        js_code_before_wait=inject,   # runs BEFORE wait_for
        wait_for="css:.signup-complete",
    )
    return await crawler.arun(PAGE_URL, config=config)

from capskip import CapSkipError

async def after_goto(page, context, url, response, **kwargs):
    holder = await page.query_selector("div.g-recaptcha[data-sitekey]")
    if holder is None:
        return page   # no widget, nothing to do
    sitekey = await holder.get_attribute("data-sitekey")
    try:
        result = await solver.recaptcha(sitekey=sitekey, url=page.url)
    except CapSkipError:
        return page   # keep the page HTML if the solve fails
    async with page.expect_navigation():
        await page.evaluate(
            "t => { document.getElementById('g-recaptcha-response').value = t;"
            " document.querySelector('form').submit(); }", result["code"])
    return page

crawler.crawler_strategy.set_hook("after_goto", after_goto)

# pip install crawl4ai capskip
import asyncio
import json
import os
import re
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from capskip import AsyncCapSkip

PAGE_URL = "https://example.com/signup"
SESSION = "signup"
WIDGET = r'<[^>]*class="(?:[^"]*\s)?g-recaptcha(?:\s[^"]*)?"[^>]*>'

solver = AsyncCapSkip(
    apiKey=os.getenv("CAPSKIP_API_KEY", "capskip"),
    host=os.getenv("CAPSKIP_HOST", "127.0.0.1"),
    port=int(os.getenv("CAPSKIP_PORT", "8080")),
)

async def main():
    async with AsyncWebCrawler() as crawler:
        first = await crawler.arun(
            PAGE_URL, config=CrawlerRunConfig(session_id=SESSION))
        widget = re.search(WIDGET, first.html)
        match = widget and re.search(r'data-sitekey="([^"]+)"', widget.group(0))
        if not match:
            raise RuntimeError(f"no sitekey found: {first.error_message}")

        result = await solver.recaptcha(sitekey=match.group(1), url=PAGE_URL)

        inject = (
            "document.getElementById('g-recaptcha-response').value = "
            f"{json.dumps(result['code'])};"
            "document.querySelector('form').submit();"
        )
        done = await crawler.arun(PAGE_URL, config=CrawlerRunConfig(
            session_id=SESSION,
            js_only=True,
            js_code_before_wait=inject,
            wait_for="css:.signup-complete",
            wait_for_timeout=30000,
        ))
        print(done.success)   # True once the next page has loaded

asyncio.run(main())