AsyncWebCrawler 中的 Hooks 和 Auth

Crawl4AI 的钩子可让您在管道中的特定点自定义爬虫：

1.on_browser_created – 创建浏览器后。2.on_page_context_created – 创建新上下文和页面后。3.before_goto – 导航至某个页面之前。4.after_goto – 导航完成后立即。5.on_user_agent_updated – 每当用户代理发生变化时。6.on_execution_started – 一旦自定义 JavaScript 开始执行。7.before_retrieve_html – 就在爬虫程序检索最终 HTML 之前。8.before_return_html – 在返回 HTML 内容之前。

重要提示：避免在on_browser_created因为你还没有页面上下文。如果你需要登录，请在on_page_context_created。

注意“重要的钩子使用警告” 避免误用钩子：不要在错误的钩子中或在错误的时间操作页面对象，因为这可能会导致管道崩溃或产生错误的结果。一个常见的错误是尝试过早处理身份验证——例如在on_browser_created. 使用正确的 Hook 进行身份验证：如果您需要登录或设置令牌，请使用on_page_context_created。这可确保您拥有可用的有效页面/上下文，而不会中断主抓取流程。基于身份的抓取：为了实现可靠的身份验证，请考虑使用基于身份的抓取（或传递会话 ID）来保存状态。在单独的、定义明确的流程中运行初始登录步骤，然后将该会话提供给主抓取，而不是将复杂的身份验证强行塞入早期的钩子中。查看基于身份的抓取了解更多详情。注意：在错误的钩子中覆盖或删除元素可能会影响最终的抓取。让钩子专注于较小的任务（例如路由过滤器、自定义标头），并让您的主要逻辑（抓取、数据提取）正常进行。

下面是一个示例演示。

示例：在 AsyncWebCrawler 中使用 Hooks

import asyncio
import json
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
from playwright.async_api import Page, BrowserContext

async def main():
    print("🔗 Hooks Example: Demonstrating recommended usage")

    # 1) Configure the browser
    browser_config = BrowserConfig(
        headless=True,
        verbose=True
    )

    # 2) Configure the crawler run
    crawler_run_config = CrawlerRunConfig(
        js_code="window.scrollTo(0, document.body.scrollHeight);",
        wait_for="body",
        cache_mode=CacheMode.BYPASS
    )

    # 3) Create the crawler instance
    crawler = AsyncWebCrawler(config=browser_config)

    #
    # Define Hook Functions
    #

    async def on_browser_created(browser, **kwargs):
        # Called once the browser instance is created (but no pages or contexts yet)
        print("[HOOK] on_browser_created - Browser created successfully!")
        # Typically, do minimal setup here if needed
        return browser

    async def on_page_context_created(page: Page, context: BrowserContext, **kwargs):
        # Called right after a new page + context are created (ideal for auth or route config).
        print("[HOOK] on_page_context_created - Setting up page & context.")

        # Example 1: Route filtering (e.g., block images)
        async def route_filter(route):
            if route.request.resource_type == "image":
                print(f"[HOOK] Blocking image request: {route.request.url}")
                await route.abort()
            else:
                await route.continue_()

        await context.route("**", route_filter)

        # Example 2: (Optional) Simulate a login scenario
        # (We do NOT create or close pages here, just do quick steps if needed)
        # e.g., await page.goto("https://example.com/login")
        # e.g., await page.fill("input[name='username']", "testuser")
        # e.g., await page.fill("input[name='password']", "password123")
        # e.g., await page.click("button[type='submit']")
        # e.g., await page.wait_for_selector("#welcome")
        # e.g., await context.add_cookies([...])
        # Then continue

        # Example 3: Adjust the viewport
        await page.set_viewport_size({"width": 1080, "height": 600})
        return page

    async def before_goto(
        page: Page, context: BrowserContext, url: str, **kwargs
    ):
        # Called before navigating to each URL.
        print(f"[HOOK] before_goto - About to navigate: {url}")
        # e.g., inject custom headers
        await page.set_extra_http_headers({
            "Custom-Header": "my-value"
        })
        return page

    async def after_goto(
        page: Page, context: BrowserContext, 
        url: str, response, **kwargs
    ):
        # Called after navigation completes.
        print(f"[HOOK] after_goto - Successfully loaded: {url}")
        # e.g., wait for a certain element if we want to verify
        try:
            await page.wait_for_selector('.content', timeout=1000)
            print("[HOOK] Found .content element!")
        except:
            print("[HOOK] .content not found, continuing anyway.")
        return page

    async def on_user_agent_updated(
        page: Page, context: BrowserContext, 
        user_agent: str, **kwargs
    ):
        # Called whenever the user agent updates.
        print(f"[HOOK] on_user_agent_updated - New user agent: {user_agent}")
        return page

    async def on_execution_started(page: Page, context: BrowserContext, **kwargs):
        # Called after custom JavaScript execution begins.
        print("[HOOK] on_execution_started - JS code is running!")
        return page

    async def before_retrieve_html(page: Page, context: BrowserContext, **kwargs):
        # Called before final HTML retrieval.
        print("[HOOK] before_retrieve_html - We can do final actions")
        # Example: Scroll again
        await page.evaluate("window.scrollTo(0, document.body.scrollHeight);")
        return page

    async def before_return_html(
        page: Page, context: BrowserContext, html: str, **kwargs
    ):
        # Called just before returning the HTML in the result.
        print(f"[HOOK] before_return_html - HTML length: {len(html)}")
        return page

    #
    # Attach Hooks
    #

    crawler.crawler_strategy.set_hook("on_browser_created", on_browser_created)
    crawler.crawler_strategy.set_hook(
        "on_page_context_created", on_page_context_created
    )
    crawler.crawler_strategy.set_hook("before_goto", before_goto)
    crawler.crawler_strategy.set_hook("after_goto", after_goto)
    crawler.crawler_strategy.set_hook(
        "on_user_agent_updated", on_user_agent_updated
    )
    crawler.crawler_strategy.set_hook(
        "on_execution_started", on_execution_started
    )
    crawler.crawler_strategy.set_hook(
        "before_retrieve_html", before_retrieve_html
    )
    crawler.crawler_strategy.set_hook(
        "before_return_html", before_return_html
    )

    await crawler.start()

    # 4) Run the crawler on an example page
    url = "https://example.com"
    result = await crawler.arun(url, config=crawler_run_config)

    if result.success:
        print("\nCrawled URL:", result.url)
        print("HTML length:", len(result.html))
    else:
        print("Error:", result.error_message)

    await crawler.close()

if __name__ == "__main__":
    asyncio.run(main())

钩子生命周期总结

1.on_browser_created : - 浏览器已启动，但尚无页面或上下文。- 仅进行简单设置 - 请勿尝试在此处打开或关闭页面（这属于on_page_context_created）。

2.on_page_context_created ：- 非常适合高级身份验证或路由阻止。- 您已准备好页面 + 上下文，但尚未导航到目标 URL。

3.before_goto : - 导航之前。通常用于设置自定义标头或记录目标 URL。

4.after_goto : - 页面导航完成后。适合验证内容或等待必要元素。

5.on_user_agent_updated : - 每当用户代理发生变化时（对于隐身或不同的 UA 模式）。

6.on_execution_started : - 如果你设置js_code或者运行自定义脚本，它会在你的 JS 即将启动时运行。

7.before_retrieve_html : - 在截取最终 HTML 快照之前。通常，您会在这里执行最终滚动或延迟加载触发器。

8.before_return_html : - 将 HTML 返回到CrawlResult. 适合记录 HTML 长度或微小修改。

何时处理身份验证

建议：使用on_page_context_created如果你需要：

导航到登录页面或填写表格
设置 cookies 或 localStorage 令牌
阻止资源路由以避免广告

这确保新创建的上下文在arun()导航至主 URL。

其他注意事项

会话管理：如果您想要多个arun()调用以重用单个会话，传递session_id=在你的CrawlerRunConfig。钩子保持不变。
性能：如果 Hooks 执行繁重的任务，可能会降低抓取速度。请保持其简洁。
错误处理：如果钩子失败，整个爬取过程可能会失败。请捕获异常或进行优雅的处理。
并发：如果你运行arun_many()每个 URL 都会并行触发这些钩子。请确保您的钩子是线程/异步安全的。

结论

钩子提供对以下内容的细粒度控制：

浏览器创建（仅限轻量级任务）
页面和上下文创建（授权、路由阻止）
导航阶段
最终 HTML 检索

遵循建议的使用方法： - 登录或高级任务on_page_context_created- 自定义标题或登录before_goto/after_goto - 滚动或最终检查before_retrieve_html/before_return_html