Crawl4ai

Crawl4ai

概述

Crawl4ai 是一个由 AI 驱动的网络爬虫框架，专为高效地从网站中提取结构化数据而设计。它将传统的 HTML 解析技术与 AI 能力相结合，可处理动态内容、智能提取文本，并对来自复杂网页的原始数据进行清洗与结构化。

何时使用该能力

当 Codex 需要执行以下操作时，请使用本技能：

• 从网页中提取结构化数据（如商品信息、文章、表单、表格等）
• 爬取含有动态内容或复杂 JavaScript 的网站
• 对来自多种 HTML 结构的提取结果进行清洗与标准化
• 与返回 HTML 的 API 或 Web 服务协同工作
• 通过直接爬取绕过 CORS 限制
• 在大规模场景下可靠地处理网页内容

触发短语示例：

• “从该网站提取数据”
• “为此页面爬取 [具体数据]”
• “解析此 HTML”
• “从 [URL] 获取数据”
• “从 [网站] 提取结构化信息”
• “为 [数据类型] 爬取 [网站]”
• “Web scrape [URL]”

快速入门

基础用法

from crawl4ai import AsyncWebCrawler, BrowserMode

async def scrape_page(url):
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url=url,
            browser_mode=BrowserMode.LATEST,
            headless=True
        )
        return result.markdown, result.clean_html

提取结构化数据

from crawl4ai import AsyncWebCrawler, JsonModeScreener
import json

async def extract_products(url):
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url=url,
            screenshot=True,
            javascript=True,
            bypass_cache=True
        )
        # 提取商品数据
        products = []
        for item in result.extracted_content:
            if item['type'] == 'product':
                products.append({
                    'name': item['name'],
                    'price': item['price'],
                    'url': item['url']
                })
        return products

常见任务

网络爬虫基础

场景： 用户希望爬取某网站上的全部文章标题。

from crawl4ai import AsyncWebCrawler

async def scrape_articles(url):
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url=url,
            javascript=True,
            verbose=True
        )
        # 从 HTML 中提取文章标题
        articles = result.extracted_content if result.extracted_content else []
        titles = [item.get('name', item.get('text', '')) for item in articles]
        return titles

触发短语： “为此网站爬取文章标题” 或 “从 [URL] 获取所有标题”

动态内容处理

场景： 网站通过 JavaScript 加载数据。

from crawl4ai import AsyncWebCrawler

async def scrape_dynamic_site(url):
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url=url,
            javascript=True,  # 等待 JS 执行完成
            wait_for="body",   # 等待特定元素出现
            delay=1.5,         # 页面加载后额外等待时间
            headless=True
        )
        return result.markdown

触发短语： “爬取这个动态网站” 或 “该页面需通过 JavaScript 加载数据”

结构化数据提取

场景： 提取价格、描述等特定字段。

from crawl4ai import AsyncWebCrawler

async def extract_product_details(url):
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url=url,
            screenshot=True,
            js_code="""
                const products = document.querySelectorAll('.product');
                return Array.from(products).map(p => ({
                    name: p.querySelector('.name')?.textContent,
                    price: p.querySelector('.price')?.textContent,
                    url: p.querySelector('a')?.href
                }));
            """
        )
        return result.extracted_content

触发短语： “从此页面提取商品详情” 或 “从 [URL] 获取价格和名称”

HTML 清洗与解析

场景： 清理杂乱 HTML 并提取干净文本。

from crawl4ai import AsyncWebCrawler

async def clean_and_parse(url):
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url=url,
            remove_tags=['script', 'style', 'nav', 'footer', 'header'],
            only_main_content=True
        )
        # 清洗并返回 Markdown 格式内容
        clean_text = result.clean_html
        return clean_text

触发短语： “清洗此 HTML” 或 “从此页面提取主体内容”

高级功能

自定义 JavaScript 注入

async def custom_scrape(url, custom_js):
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url=url,
            js_code=custom_js,
            js_only=True  # 仅执行 JS，不下载其他资源
        )
        return result.extracted_content

会话管理

from crawl4ai import AsyncWebCrawler

async def multi_page_scrape(base_url, urls):
    async with AsyncWebCrawler() as crawler:
        results = []
        for url in urls:
            result = await crawler.arun(
                url=url,
                session_id=f"session_{url}",
                bypass_cache=True
            )
            results.append({
                'url': url,
                'content': result.markdown,
                'status': result.success
            })
        return results

最佳实践

1. 始终检查目标网站是否允许爬取 —— 尊重 robots.txt 及服务条款
2. 设置合理的请求间隔 —— 添加延迟以避免对服务器造成过大压力
3. 优雅地处理错误 —— 实现重试机制与异常捕获逻辑
4. 有选择性地提取数据 —— 仅获取所需内容，避免整页抓取
5. 可靠地存储数据 —— 将提取结果保存为结构化格式（如 JSON、CSV）
6. 规范化 URL —— 正确处理重定向及格式错误的 URL

错误处理

async def robust_scrape(url):
    try:
        async with AsyncWebCrawler() as crawler:
            result = await crawler.arun(
                url=url,
                timeout=30000  # 设置 30 秒超时
            )
            if result.success:
                return result.markdown, result.extracted_content
            else:
                print(f"爬取失败：{result.error_message}")
                return None, None
    except Exception as e:
        print(f"爬取异常：{str(e)}")
        return None, None

输出格式

Crawl4ai 支持多种输出格式：

• Markdown：简洁、易读的纯文本（result.markdown）
• Clean HTML：结构清晰、已清洗的 HTML（result.clean_html）
• Extracted Content：结构化的 JSON 数据（result.extracted_content）
• Screenshot：页面截图（result.screenshot）
• Links：页面中发现的所有链接（result.links）

相关资源

scripts/

常用爬虫操作的 Python 脚本：

• scrape_single_page.py — 基础爬取工具
• scrape_multiple_pages.py — 支持分页的批量爬取
• extract_from_html.py — HTML 解析辅助脚本
• clean_html.py — HTML 清洗工具

references/

文档与示例：

• api_reference.md — 完整 API 文档
• examples.md — 常见用例与模式
• error_handling.md — 故障排查指南