Crawl4AI CLI工具与自动化脚本

2026-02-04 04:54:58作者：侯霆垣

Crawl4AI提供了一个功能强大且直观的命令行界面（CLI），让用户能够通过简单的命令执行复杂的网页抓取和数据提取任务。CLI工具基于Python的Click库构建，提供了丰富的命令选项和配置功能，包括网页抓取、配置文件管理、浏览器调试、全局配置管理等多种核心命令。

命令行界面功能详解

Crawl4AI提供了一个功能强大且直观的命令行界面（CLI），让用户能够通过简单的命令执行复杂的网页抓取和数据提取任务。CLI工具基于Python的Click库构建，提供了丰富的命令选项和配置功能。

CLI命令结构概览

Crawl4AI CLI采用多级命令结构，主要包含以下几个核心命令：

flowchart TD
    A[crwl CLI] --> B[crawl<br>网页抓取]
    A --> C[profiles<br>浏览器配置管理]
    A --> D[cdp<br>浏览器调试]
    A --> E[config<br>全局配置]
    A --> F[browser<br>内置浏览器管理]
    A --> G[examples<br>使用示例]
    
    B --> B1[基本抓取]
    B --> B2[配置文件]
    B --> B3[LLM提取]
    B --> B4[深度抓取]
    
    C --> C1[创建配置]
    C --> C2[列表配置]
    C --> C3[删除配置]
    
    D --> D1[启动浏览器]
    D --> D2[CDP调试]

核心命令详解

1. 网页抓取命令（crawl）

crwl crawl 是主要的抓取命令，支持丰富的参数选项：

# 基本用法
crwl crawl https://example.com

# 使用Markdown输出
crwl crawl https://example.com -o markdown

# 使用JSON输出
crwl crawl https://example.com -o json

# 使用配置文件
crwl crawl https://example.com -B browser.yml -C crawler.yml

# LLM结构化数据提取
crwl crawl https://example.com -j "提取产品信息"

# 深度抓取
crwl crawl https://example.com --deep-crawl bfs --max-pages 20

主要参数选项：

参数	缩写	描述	示例
`--browser-config`	`-B`	浏览器配置文件	`-B browser.yml`
`--crawler-config`	`-C`	抓取器配置文件	`-C crawler.yml`
`--json-extract`	`-j`	LLM结构化提取	`-j "提取文章"`
`--schema`	`-s`	JSON Schema文件	`-s schema.json`
`--output`	`-o`	输出格式	`-o markdown`
`--profile`	`-p`	浏览器配置	`-p my-profile`
`--deep-crawl`		深度抓取策略	`--deep-crawl bfs`
`--max-pages`		最大页面数	`--max-pages 50`
`--verbose`	`-v`	详细输出	`-v`

2. 配置文件管理

Crawl4AI支持通过YAML/JSON配置文件进行详细配置：

浏览器配置文件示例 (browser.yml):

browser_type: "chromium"
headless: true
viewport_width: 1280
viewport_height: 800
user_agent_mode: "random"
verbose: true
ignore_https_errors: true
java_script_enabled: true
extra_args:
  - "--disable-gpu"
  - "--no-sandbox"

抓取器配置文件示例 (crawler.yml):

cache_mode: "bypass"
wait_until: "networkidle"
page_timeout: 30000
delay_before_return_html: 0.5
word_count_threshold: 100
scan_full_page: true
scroll_delay: 0.3
process_iframes: false
remove_overlay_elements: true
magic: true
verbose: true

3. 浏览器配置管理

crwl profiles 命令提供交互式的浏览器配置管理：

# 启动配置管理器
crwl profiles

# 使用特定配置进行抓取
crwl crawl https://example.com -p my-profile

浏览器配置功能允许用户创建和管理多个浏览器实例，每个配置可以保存不同的认证状态、cookies和浏览器设置，非常适合需要登录的网站抓取。

4. CDP浏览器调试

crwl cdp 命令启动带有Chrome DevTools Protocol调试功能的浏览器：

# 启动CDP浏览器
crwl cdp

# 使用特定端口和配置
crwl cdp -P 9223 -p my-profile --headless

这个功能对于高级用户和开发者非常有用，可以与其他自动化工具（如Puppeteer、Playwright）集成使用。

5. 全局配置管理

crwl config 命令组用于管理全局配置设置：

# 列出所有配置
crwl config list

# 获取特定配置值
crwl config get DEFAULT_LLM_PROVIDER

# 设置配置值
crwl config set VERBOSE true
crwl config set BROWSER_HEADLESS false

可用的全局配置选项：

配置项	类型	默认值	描述
`DEFAULT_LLM_PROVIDER`	string	`"openai/gpt-4o"`	默认LLM提供商
`DEFAULT_LLM_PROVIDER_TOKEN`	string	`""`	LLM API令牌
`VERBOSE`	boolean	`false`	详细输出模式
`BROWSER_HEADLESS`	boolean	`true`	无头浏览器模式
`BROWSER_TYPE`	string	`"chromium"`	浏览器类型
`CACHE_MODE`	string	`"bypass"`	缓存模式
`USER_AGENT_MODE`	string	`"default"`	User Agent模式

高级功能特性

1. 键值对参数解析

CLI支持灵活的键值对参数格式，可以动态设置各种配置：

# 浏览器参数设置
crwl crawl https://example.com -b "headless=false,viewport_width=1920,user_agent_mode=random"

# 抓取器参数设置
crwl crawl https://example.com -c "css_selector=#main,delay_before_return_html=2,scan_full_page=true"

支持的数据类型包括：字符串、布尔值、整数、浮点数、数组和JSON对象。

2. LLM集成提取

CLI深度集成LLM功能，支持智能内容提取：

# 自动结构化数据提取
crwl crawl https://example.com -j

# 带指令的提取
crwl crawl https://example.com -j "提取所有产品信息包括名称、价格和描述"

# 使用Schema的精确提取
crwl crawl https://example.com -j -s product_schema.json

3. 深度抓取策略

支持多种深度抓取算法：

# 广度优先搜索
crwl crawl https://example.com --deep-crawl bfs --max-pages 30

# 深度优先搜索
crwl crawl https://example.com --deep-crawl dfs --max-pages 20

# 最佳优先搜索
crwl crawl https://example.com --deep-crawl best-first --max-pages 40

4. 内容问答功能

CLI内置内容问答功能，可以直接对抓取的内容提问：

crwl crawl https://example.com -q "这篇文章的主要观点是什么？"

输出格式选项

CLI支持多种输出格式，满足不同场景需求：

输出格式	命令选项	描述
完整JSON	`-o all`	包含所有抓取数据的完整JSON
提取内容JSON	`-o json`	仅提取的结构化数据
原始Markdown	`-o markdown`	原始Markdown格式内容
优化Markdown	`-o markdown-fit`	经过优化的精简Markdown

错误处理与调试

CLI提供详细的错误信息和调试功能：

详细输出模式：-v 参数启用详细日志
配置验证：自动验证配置文件和参数格式
错误报告：清晰的错误消息和解决方案建议
缓存控制：--bypass-cache 参数控制缓存行为

典型工作流程

sequenceDiagram
    participant User
    participant CLI
    participant Config
    participant Browser
    participant Crawler
    participant Output

    User->>CLI: crwl crawl URL -B config.yml
    CLI->>Config: 加载配置文件
    Config-->>CLI: 配置对象
    CLI->>Browser: 初始化浏览器
    Browser-->>CLI: 浏览器实例
    CLI->>Crawler: 执行抓取任务
    Crawler->>Browser: 访问页面
    Browser-->>Crawler: 页面内容
    Crawler-->>CLI: 抓取结果
    CLI->>Output: 格式化输出
    Output-->>User: 最终结果

Crawl4AI CLI工具的设计充分考虑了开发者和数据工程师的需求，提供了从简单到复杂的各种使用场景支持。无论是快速的单页抓取还是复杂的多页深度抓取，CLI都能提供高效、灵活的解决方案。

C4A脚本语言使用指南

C4A脚本语言是Crawl4AI框架中的一种声明式自动化脚本语言，专门为网页交互和自动化任务设计。它采用简洁的语法结构，让开发者能够以直观的方式描述复杂的浏览器操作流程，而无需编写复杂的JavaScript代码。

语言基础与语法结构

C4A脚本语言基于类似自然语言的命令式语法，每个命令占据一行，支持流程控制、变量操作和过程定义等高级特性。

基本命令语法

# 导航命令
GO https://example.com
RELOAD
BACK
FORWARD

# 等待命令
WAIT 2000  # 等待2秒
WAIT "#content" 5  # 等待选择器出现，最多5秒

# 点击操作
CLICK "button.submit"
DOUBLE_CLICK ".product-image"
RIGHT_CLICK ".context-menu"

# 坐标操作
MOVE 100 200  # 移动到坐标(100,200)
DRAG 100 200 300 400  # 从(100,200)拖拽到(300,400)
SCROLL DOWN 500  # 向下滚动500像素

变量与流程控制

C4A脚本支持变量操作和条件判断，使脚本更加灵活：

# 变量设置
SETVAR $username = "john_doe"
SETVAR $password = "secure123"

# 条件判断
IF (EXISTS ".login-form") THEN
    TYPE $username INTO "input#username"
    TYPE $password INTO "input#password"
    CLICK "button.login"
ELSE
    CLICK ".show-login"
ENDIF

# 循环操作
REPEAT (CLICK ".load-more", 3)  # 点击"加载更多"按钮3次

核心命令详解

导航与页面控制

flowchart TD
    A[GO URL] --> B[页面加载]
    B --> C{等待条件}
    C -->|选择器存在| D[执行操作]
    C -->|超时| E[继续执行]
    D --> F[完成导航]
    E --> F

命令	参数	描述	示例
`GO`	URL字符串	导航到指定URL	`GO https://example.com`
`RELOAD`	无	重新加载当前页面	`RELOAD`
`BACK`	无	返回上一页	`BACK`
`FORWARD`	无	前进到下一页	`FORWARD`

元素交互操作

# 文本输入
TYPE "search query" INTO "input.search"
CLEAR "input.search"  # 清空输入框
SET "input.quantity" TO "5"  # 设置输入框值

# 键盘操作
PRESS "Enter"  # 按下回车键
KEY_DOWN "Control"  # 按下Control键
KEY_UP "Control"  # 释放Control键

# JavaScript执行
EVAL "console.log('Hello from C4A')"
EVAL "return document.title"  # 返回值可用于后续操作

高级特性与过程定义

过程（Procedure）定义

C4A脚本支持过程定义，实现代码复用：

PROC login
    WAIT ".login-form" 10
    TYPE $username INTO "#username"
    TYPE $password INTO "#password"
    CLICK "#login-btn"
    WAIT 2000  # 等待登录完成
ENDPROC

PROC search_product $product_name
    TYPE $product_name INTO "#search-box"
    PRESS "Enter"
    WAIT ".search-results" 5
ENDPROC

# 调用过程
login
search_product "laptop"

条件判断与循环

flowchart LR
    A[IF条件] --> B{条件满足?}
    B -->|是| C[执行THEN分支]
    B -->|否| D{有ELSE?}
    D -->|是| E[执行ELSE分支]
    D -->|否| F[继续执行]
    C --> F
    E --> F

# 复杂条件判断
IF (EXISTS ".welcome-banner") THEN
    CLICK ".close-banner"
ENDIF

IF (NOT EXISTS ".logged-in") THEN
    login  # 调用登录过程
ELSE
    CLICK ".profile-icon"
ENDIF

# 基于JavaScript的条件
IF (`document.title.includes("Dashboard")`) THEN
    NAVIGATE_TO_DASHBOARD
ENDIF

实战示例

电商网站自动化

# 电商产品搜索自动化脚本
PROC search_and_filter $category $price_range
    GO "https://example-store.com"
    WAIT ".search-input" 5
    
    TYPE $category INTO ".search-input"
    PRESS "Enter"
    WAIT ".search-results" 10
    
    # 价格筛选
    IF (EXISTS ".price-filter") THEN
        SET ".min-price" TO $price_range.min
        SET ".max-price" TO $price_range.max
        CLICK ".apply-filter"
        WAIT 3000
    ENDIF
    
    # 加载更多结果
    REPEAT (CLICK ".load-more", 5)
ENDPROC

# 使用脚本
search_and_filter "laptop" {"min": "500", "max": "1500"}

社交媒体数据采集

# 社交媒体内容采集脚本
PROC collect_posts $hashtag $max_posts
    GO "https://social-media.com/explore/tags/$hashtag"
    WAIT ".post-container" 10
    
    SETVAR $collected = 0
    
    WHILE ($collected < $max_posts) DO
        SCROLL DOWN 1000
        WAIT 2000
        
        # 收集可见的帖子
        EVAL `
            const posts = document.querySelectorAll('.post');
            return Array.from(posts).map(post => ({
                content: post.querySelector('.content').textContent,
                likes: post.querySelector('.like-count').textContent,
                timestamp: post.querySelector('.time').textContent
            }));
        `
        
        SETVAR $collected = $collected + 10
        
        # 检查是否还有更多内容
        IF (NOT EXISTS ".load-more") THEN
            BREAK
        ENDIF
    ENDWHILE
ENDPROC

错误处理与调试

C4A脚本提供详细的错误信息和调试支持：

# 错误处理示例
PROC safe_click $selector $timeout
    IF (EXISTS $selector $timeout) THEN
        CLICK $selector
    ELSE
        LOG "Element $selector not found within $timeout seconds"
        # 备用方案
        CLICK ".alternative-button"
    ENDIF
ENDPROC

# 使用try-catch风格的错误处理
PROC robust_navigation $url
    GO $url
    WAIT 5000  # 等待页面加载
    
    IF (`document.readyState !== "complete"`) THEN
        RELOAD
        WAIT 5000
    ENDIF
ENDPROC

集成与使用方式

Python代码集成

from crawl4ai.script.c4a_compile import compile
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig

# 编译C4A脚本
script = """
GO https://example.com
WAIT "#main-content" 10
CLICK ".accept-cookies"
"""

compilation_result = compile(script)

if compilation_result.success:
    # 配置爬虫使用编译后的JavaScript
    config = CrawlerRunConfig(
        js_code=compilation_result.js_code,
        verbose=True
    )
    
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com",
            config=config
        )
        print("执行完成:", result.success)
else:
    print("编译错误:", compilation_result.first_error.message)

CLI命令行使用

# 使用C4A脚本文件
crwl https://example.com --c4a-script my_script.c4a

# 直接内联脚本
crwl https://example.com --c4a "GO https://example.com; WAIT 2000; CLICK '.button'"

# 结合其他配置使用
crwl https://example.com \
  --c4a-script login_flow.c4a \
  --browser-config "headless=false,viewport_width=1200" \
  --output json

最佳实践与性能优化

选择器优化: 使用具体的选择器而非通用选择器提高执行效率
超时设置: 为关键操作设置合理的超时时间
错误处理: 为可能失败的操作提供备用方案
模块化设计: 将常用操作封装为过程便于复用
日志记录: 在关键步骤添加日志输出便于调试

# 优化后的脚本示例
PROC optimized_login $credentials
    # 使用ID选择器而非类选择器
    WAIT "#username" 5
    TYPE $credentials.username INTO "#username"
    TYPE $credentials.password INTO "#password"
    
    # 明确的超时设置
    CLICK "#login-btn"
    WAIT "#dashboard" 10  # 等待登录成功
    
    # 验证登录状态
    IF (NOT EXISTS "#dashboard") THEN
        LOG "登录可能失败，尝试备用方案"
        CLICK ".forgot-password"
        # 处理密码重置流程
    ENDIF
ENDPROC

C4A脚本语言通过其简洁的语法和强大的功能，为网页自动化任务提供了高效的解决方案，特别适合需要复杂交互流程的爬虫场景。

自动化爬取工作流构建

Crawl4AI提供了强大的CLI工具和灵活的配置系统，使得构建自动化爬取工作流变得简单高效。通过配置文件、批量处理能力和系统集成，您可以轻松构建生产级的自动化数据采集流水线。

配置文件驱动的自动化

Crawl4AI支持YAML和JSON格式的配置文件，允许您将复杂的爬取配置参数化，便于版本控制和批量执行。

浏览器配置文件示例 (browser.yml)

browser_type: "chromium"
headless: true
viewport_width: 1280
viewport_height: 800
user_agent_mode: "random"
verbose: true
ignore_https_errors: true
java_script_enabled: true
extra_args:
  - "--disable-gpu"
  - "--no-sandbox"
  - "--disable-dev-shm-usage"
proxy_config:
  server: "http://proxy.example.com:8080"
  username: "user"
  password: "pass"

爬虫配置文件示例 (crawler.yml)

cache_mode: "bypass"
wait_until: "networkidle"
page_timeout: 30000
delay_before_return_html: 0.5
word_count_threshold: 100
scan_full_page: true
scroll_delay: 0.3
process_iframes: false
remove_overlay_elements: true
magic: true
verbose: true
exclude_external_links: true
exclude_social_media_links: true
deep_crawl_strategy:
  type: "bfs"
  max_depth: 3
  max_pages: 50

批量URL处理工作流

Crawl4AI支持批量处理多个URL，通过Python脚本或CLI命令实现高效的数据采集。

Python批量处理示例

import asyncio
from crawl4ai import AsyncWebCrawler

async def batch_crawling_workflow():
    async with AsyncWebCrawler(verbose=True) as crawler:
        # 定义要爬取的URL列表
        urls = [
            "https://news.example.com/technology",
            "https://news.example.com/business", 
            "https://news.example.com/science",
            "https://blog.example.com/ai-trends",
            "https://blog.example.com/machine-learning"
        ]
        
        # 批量执行爬取
        results = await crawler.arun_many(
            urls=urls,
            word_count_threshold=150,
            bypass_cache=True,
            verbose=True,
        )
        
        # 处理结果
        successful_crawls = []
        for result in results:
            if result.success:
                successful_crawls.append({
                    'url': result.url,
                    'title': result.metadata.get('title', 'N/A'),
                    'word_count': len(result.markdown.split()),
                    'content_hash': hash(result.markdown.raw_markdown)
                })
                # 保存Markdown内容到文件
                filename = f"data/{result.url.split('//')[1].replace('/', '_')}.md"
                with open(filename, 'w', encoding='utf-8') as f:
                    f.write(result.markdown.raw_markdown)
        
        return successful_crawls

# 执行批量爬取
if __name__ == "__main__":
    results = asyncio.run(batch_crawling_workflow())
    print(f"成功爬取 {len(results)} 个页面")

定时自动化任务集成

通过系统调度工具（如cron、systemd timers或任务计划程序），您可以轻松设置定时爬取任务。

Linux Cron Job 配置

# 每天凌晨2点执行新闻网站爬取
0 2 * * * /usr/bin/python3 /path/to/your/news_crawler.py >> /var/log/crawl4ai/news.log 2>&1

# 每小时执行一次监控爬取
0 * * * * /usr/local/bin/crwl https://status.example.com -o json -B /etc/crawl4ai/monitor_browser.yml -C /etc/crawl4ai/monitor_crawler.yml >> /var/log/crawl4ai/monitor.log

# 每周一早上6点执行深度爬取
0 6 * * 1 /usr/local/bin/crwl https://blog.example.com --deep-crawl bfs --max-pages 100 -B /etc/crawl4ai/deep_crawl_browser.yml -C /etc/crawl4ai/deep_crawl_crawler.yml >> /var/log/crawl4ai/deep_crawl.log

Systemd Service 配置

创建 /etc/systemd/system/crawl4ai-news.service:

[Unit]
Description=Crawl4AI News Crawling Service
After=network.target

[Service]
Type=oneshot
User=crawluser
Group=crawlgroup
WorkingDirectory=/opt/crawl4ai
ExecStart=/usr/bin/python3 /opt/crawl4ai/news_crawler.py
Environment=PYTHONPATH=/opt/crawl4ai
Environment=CRAWL4AI_BASE_DIRECTORY=/opt/crawl4ai/data
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

创建 /etc/systemd/system/crawl4ai-news.timer:

[Unit]
Description=Run Crawl4AI News Crawler daily at 2 AM
Requires=crawl4ai-news.service

[Timer]
OnCalendar=*-*-* 02:00:00
Persistent=true

[Install]
WantedBy=timers.target

工作流状态管理

flowchart TD
    A[启动爬取工作流] --> B[加载配置文件]
    B --> C[初始化浏览器实例]
    C --> D[处理URL队列]
    D --> E{批量处理模式?}
    E -->|是| F[并发处理多个URL]
    E -->|否| G[顺序处理单个URL]
    F --> H[收集所有结果]
    G --> H
    H --> I[结果后处理]
    I --> J[数据存储]
    J --> K[生成报告]
    K --> L[工作流完成]
    
    subgraph ErrorHandling [错误处理]
        M[网络错误] --> N[重试机制]
        O[解析错误] --> P[跳过或记录]
        Q[资源限制] --> R[等待后继续]
    end
    
    D --> M
    D --> O
    D --> Q

高级自动化特性

1. 智能重试机制

# retry_config.yml
max_retries: 3
retry_delay: 5
backoff_factor: 2
retry_on:
  - "timeout"
  - "network_error"
  - "429_too_many_requests"
  - "500_server_error"

2. 内存监控与调节

from crawl4ai import AsyncWebCrawler, MemoryAwareDispatcher

async def memory_aware_crawling():
    # 使用内存感知调度器
    dispatcher = MemoryAwareDispatcher(
        memory_threshold_percent=85.0,
        critical_threshold_percent=95.0,
        check_interval=2.0
    )
    
    async with AsyncWebCrawler() as crawler:
        results = await crawler.arun_many(
            urls=large_url_list,
            dispatcher=dispatcher,
            verbose=True
        )

3. 分布式爬取工作流

import asyncio
from crawl4ai import AsyncWebCrawler
from redis import Redis
from rq import Queue

# 设置分布式任务队列
redis_conn = Redis(host='redis-server', port=6379)
crawl_queue = Queue('crawling', connection=redis_conn)

def distributed_crawling_workflow():
    # 将爬取任务分发到多个worker
    for url_chunk in chunk_urls(large_url_list, chunk_size=50):
        crawl_queue.enqueue(
            process_url_chunk,
            url_chunk,
            browser_config_path,
            crawler_config_path,
            result_storage='s3://crawl-data/results/'
        )

@crawl_queue.job
def process_url_chunk(urls, browser_config, crawler_config):
    async def crawl_chunk():
        async with AsyncWebCrawler.from_config_files(
            browser_config, crawler_config
        ) as crawler:
            return await crawler.arun_many(urls)
    
    return asyncio.run(crawl_chunk())

监控与日志记录

构建完整的监控体系来跟踪自动化工作流的执行情况：

import logging
from datetime import datetime
from crawl4ai import AsyncWebCrawler

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'/var/log/crawl4ai/crawl_{datetime.now().strftime("%Y%m%d")}.log'),
        logging.StreamHandler()
    ]
)

class MonitoringCrawler:
    def __init__(self):
        self.success_count = 0
        self.failure_count = 0
        self.start_time = None
        
    async def run_with_monitoring(self, urls):
        self.start_time = datetime.now()
        logger.info(f"开始批量爬取 {len(urls)} 个URL")
        
        async with AsyncWebCrawler() as crawler:
            results = await crawler.arun_many(urls)
            
            for result in results:
                if result.success:
                    self.success_count += 1
                    logger.info(f"成功爬取: {result.url}")
                else:
                    self.failure_count += 1
                    logger.error(f"爬取失败: {result.url} - {result.error_message}")
            
            self._generate_report()
    
    def _generate_report(self):
        duration = datetime.now() - self.start_time
        logger.info(f"爬取完成: 成功 {self.success_count}, 失败 {self.failure_count}")
        logger.info(f"总耗时: {duration.total_seconds():.2f} 秒")
        logger.info(f"平均每个页面: {duration.total_seconds()/len(urls):.2f} 秒")

错误处理与恢复

实现健壮的错误处理机制确保工作流的可靠性：

from tenacity import retry, stop_after_attempt, wait_exponential

class ResilientCrawlingWorkflow:
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    async def resilient_crawl(self, url, config):
        try:
            async with AsyncWebCrawler(config=config) as crawler:
                result = await crawler.arun(url)
                if not result.success:
                    raise CrawlingError(f"爬取失败: {result.error_message}")
                return result
        except Exception as e:
            logger.warning(f"爬取尝试失败: {e}")
            raise
    
    async def run_with_recovery(self, urls):
        successful = []
        failed = []
        
        for url in urls:
            try:
                result = await self.resilient_crawl(url, self.config)
                successful.append(result)
            except Exception as e:
                failed.append({'url': url, 'error': str(e)})
                logger.error(f"无法爬取 {url}: {e}")
        
        # 保存失败记录以便后续重试
        if failed:
            self._save_failed_urls(failed)
        
        return successful, failed

通过上述配置和代码示例，您可以构建出强大、可靠且易于维护的自动化爬取工作流，满足各种生产环境下的数据采集需求。

配置文件管理与自定义扩展

Crawl4AI 提供了强大的配置文件管理系统，支持多种格式的配置文件，让用户能够轻松管理和复用复杂的爬取配置。通过配置文件，您可以定义浏览器行为、爬取策略、内容提取规则等，实现高度定制化的网络爬取工作流。

配置文件格式与结构

Crawl4AI 支持多种配置文件格式，包括 YAML、JSON 等，每种格式都有其特定的应用场景和优势。

YAML 配置文件示例

# browser.yml - 浏览器配置
headless: true
viewport_width: 1280
viewport_height: 800
user_agent_mode: "random"
proxy_config:
  server: "http://proxy.example.com:8080"
  username: "user"
  password: "pass"
verbose: true
ignore_https_errors: true
extra_args:
  - "--disable-web-security"
  - "--allow-running-insecure-content"

# crawler.yml - 爬取器配置
word_count_threshold: 5
extraction_strategy:
  type: "llm"
  provider: "openai/gpt-4o"
  instruction: "Extract all product information"
css_selector: ".product-container"
screenshot: true
pdf: false
cache_mode: "bypass"
wait_until: "domcontentloaded"
page_timeout: 60000
scan_full_page: true
scroll_delay: 0.2

JSON 配置文件示例

{
  "extraction_config": {
    "type": "json-css",
    "params": {
      "verbose": true,
      "timeout": 30000
    }
  },
  "schema": {
    "name": "ProductExtractor",
    "baseSelector": ".product-item",
    "fields": [
      {
        "name": "title",
        "selector": "h2.product-title",
        "type": "text",
        "required": true
      },
      {
        "name": "price",
        "selector": ".price",
        "type": "text",
        "transform": "parse_currency"
      },
      {
        "name": "image",
        "selector": "img.product-image",
        "type": "attribute",
        "attribute": "src"
      }
    ]
  }
}

配置文件加载与管理

Crawl4AI 提供了灵活的配置文件加载机制，支持从文件系统加载配置，并支持配置文件的合并和覆盖。

flowchart TD
    A[CLI 命令启动] --> B[解析命令行参数]
    B --> C{是否有配置文件?}
    C -->|是| D[加载配置文件]
    C -->|否| E[使用默认配置]
    D --> F[解析命令行覆盖参数]
    E --> F
    F --> G[合并配置参数]
    G --> H[创建配置对象]
    H --> I[执行爬取任务]

配置加载函数实现

def load_config_file(path: Optional[str]) -> dict:
    """加载配置文件，支持 YAML 和 JSON 格式"""
    if not path:
        return {}
    
    try:
        with open(path) as f:
            if path.endswith((".yaml", ".yml")):
                return yaml.safe_load(f)
            return json.load(f)
    except Exception as e:
        raise click.BadParameter(f'Error loading config file {path}: {str(e)}')

def parse_key_values(ctx, param, value) -> Dict[str, Any]:
    """解析命令行键值对参数，支持多种数据类型"""
    if not value:
        return {}
    result = {}
    pairs = value.split(',')
    for pair in pairs:
        try:
            k, v = pair.split('=', 1)
            # 处理常见值类型
            if v.lower() == 'true': v = True
            elif v.lower() == 'false': v = False
            elif v.isdigit(): v = int(v)
            elif v.replace('.','',1).isdigit(): v = float(v)
            elif v.startswith('[') and v.endswith(']'):
                v = [x.strip() for x in v[1:-1].split(',') if x.strip()]
            elif v.startswith('{') and v.endswith('}'):
                try:
                    v = json.loads(v)
                except json.JSONDecodeError:
                    raise click.BadParameter(f'Invalid JSON object: {v}')
            result[k.strip()] = v
        except ValueError:
            raise click.BadParameter(f'Invalid key=value pair: {pair}')
    return result

全局配置管理

Crawl4AI 提供了全局配置管理系统，允许用户设置默认参数，避免重复配置。

全局配置函数

def get_global_config() -> dict:
    """获取全局配置"""
    config_dir = Path.home() / ".crawl4ai"
    config_file = config_dir / "global.yml"
    
    if not config_file.exists():
        config_dir.mkdir(parents=True, exist_ok=True)
        return {}
        
    with open(config_file) as f:
        return yaml.safe_load(f) or {}

def save_global_config(config: dict):
    """保存全局配置"""
    config_file = Path.home() / ".crawl4ai" / "global.yml"
    with open(config_file, "w") as f:
        yaml.dump(config, f)

全局配置示例

# ~/.crawl4ai/global.yml
DEFAULT_LLM_PROVIDER: "openai/gpt-4o"
DEFAULT_LLM_PROVIDER_TOKEN: "sk-..."
VERBOSE: false
BROWSER_HEADLESS: true
BROWSER_TYPE: "chromium"
CACHE_MODE: "bypass"
USER_AGENT_MODE: "default"

自定义扩展策略

Crawl4AI 支持多种自定义扩展策略，包括内容提取策略、分块策略、标记生成策略等。

自定义提取策略示例

from crawl4ai import ExtractionStrategy
from typing import List, Dict, Any
import json

class CustomExtractionStrategy(ExtractionStrategy):
    def __init__(self, custom_param: str = "default"):
        self.custom_param = custom_param
        
    def extract(self, url: str, html: str, *q, **kwargs) -> List[Dict[str, Any]]:
        """自定义提取逻辑"""
        # 实现您的自定义提取逻辑
        results = []
        
        # 示例：提取所有标题和链接
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, 'html.parser')
        
        for article in soup.find_all('article'):
            title = article.find('h2')
            link = article.find('a')
            
            if title and link:
                results.append({
                    'title': title.get_text().strip(),
                    'url': link.get('href'),
                    'source_url': url,
                    'extraction_method': 'custom'
                })
        
        return results

# 使用自定义策略
custom_strategy = CustomExtractionStrategy(custom_param="my_value")

自定义分块策略示例

from crawl4ai import ChunkingStrategy
from typing import List

class SemanticChunkingStrategy(ChunkingStrategy):
    def __init__(self, model_name: str = "all-MiniLM-L6-v2", threshold: float = 0.8):
        self.model_name = model_name
        self.threshold = threshold
        
    def chunk(self, text: str) -> List[str]:
        """基于语义相似度的分块策略"""
        import numpy as np
        from sentence_transformers import SentenceTransformer
        from sklearn.metrics.pairwise import cosine_similarity
        
        # 加载模型
        model = SentenceTransformer(self.model_name)
        
        # 将文本分割成句子
        sentences = text.split('. ')
        if len(sentences) <= 1:
            return [text]
            
        # 计算句子嵌入
        embeddings = model.encode(sentences)
        
        chunks = []
        current_chunk = []
        
        for i, (sentence, embedding) in enumerate(zip(sentences, embeddings)):
            if not current_chunk:
                current_chunk.append(sentence)
                continue
                
            # 计算与上一个句子的相似度
            prev_embedding = embeddings[i-1]
            similarity = cosine_similarity([embedding], [prev_embedding])[0][0]
            
            if similarity >= self.threshold:
                current_chunk.append(sentence)
            else:
                chunks.append('. '.join(current_chunk) + '.')
                current_chunk = [sentence]
                
        if current_chunk:
            chunks.append('. '.join(current_chunk) + '.')
            
        return chunks

配置文件的最佳实践

1. 模块化配置管理

# configs/
# ├── browsers/
# │   ├── default.yml
# │   ├── stealth.yml
# │   └── mobile.yml
# ├── crawlers/
# │   ├── basic.yml
# │   ├── deep_crawl.yml
# │   └── api_crawl.yml
# └── extractors/
#     ├── product.yml
#     ├── article.yml
#     └── contact.yml

2. 环境特定的配置

# configs/production/browser.yml
headless: true
viewport_width: 1280
user_agent_mode: "random"
proxy_config:
  server: "${PROXY_SERVER}"
  username: "${PROXY_USER}"
  password: "${PROXY_PASS}"
verbose: false

# configs/development/browser.yml  
headless: false
viewport_width: 1920
user_agent_mode: "default"
verbose: true

3. 配置验证与模板

from pydantic import BaseModel, validator
from typing import Optional, List

class BrowserConfigTemplate(BaseModel):
    headless: bool = True
    viewport_width: int = 1080
    viewport_height: int = 600
    user_agent_mode: str = "default"
    proxy_config: Optional[dict] = None
    
    @validator('viewport_width')
    def validate_viewport_width(cls, v):
        if v < 320 or v > 3840:
            raise ValueError('Viewport width must be between 320 and 3840')
        return v
        
    @validator('user_agent_mode')
    def validate_user_agent_mode(cls, v):
        valid_modes = ["default", "random", "mobile"]
        if v not in valid_modes:
            raise ValueError(f'User agent mode must be one of {valid_modes}')
        return v

# 使用模板验证配置
def validate_config(config: dict, template: BaseModel) -> dict:
    try:
        return template(**config).dict()
    except Exception as e:
        raise ValueError(f"Configuration validation failed: {str(e)}")

高级配置技巧

1. 动态配置生成

def generate_dynamic_config(url: str) -> dict:
    """根据URL动态生成配置"""
    if "ecommerce" in url:
        return {
            "extraction_strategy": {
                "type": "json-css",
                "schema": {
                    "baseSelector": ".product",
                    "fields": ["name", "price", "image"]
                }
            },
            "screenshot": True
        }
    elif "blog" in url:
        return {
            "extraction_strategy": {
                "type": "llm",
                "instruction": "Extract article content and metadata"
            },
            "scan_full_page": True
        }
    else:
        return {"verbose": True}

2. 配置继承与覆盖

# base.yml
browser:
  headless: true
  viewport_width: 1280
crawler:
  cache_mode: "bypass"
  verbose: false

# override.yml  
extends: base.yml
browser:
  viewport_width: 1920  # 覆盖基础配置
crawler:
  verbose: true         # 覆盖基础配置
  screenshot: true      # 新增配置

3. 配置版本控制

# config.v1.yml
version: 1.0
config_type: "browser"
parameters:
  headless: true
  viewport: 
    width: 1280
    height: 800
metadata:
  created: "2024-01-01"
  author: "user@example.com"

# config.v2.yml
version: 2.0
config_type: "browser"
parameters:
  headless: true
  viewport:
    width: 1920
    height: 1080
  user_agent:
    mode: "random"
    custom: null
metadata:
  created: "2024-01-15"
  author: "user@example.com"
  migration_notes: "Added user_agent configuration"

通过灵活的配置文件管理和自定义扩展机制，Crawl4AI 让用户能够构建高度定制化、可复用的网络爬取解决方案，满足各种复杂的业务需求。

Crawl4AI通过灵活的配置文件管理和自定义扩展机制，为用户提供了高度定制化、可复用的网络爬取解决方案。无论是简单的单页抓取还是复杂的多页深度抓取，Crawl4AI都能提供高效、灵活的配置选项和自动化工作流支持，满足各种复杂的业务需求。通过模块化配置管理、环境特定配置、配置验证与模板等最佳实践，用户可以构建出强大、可靠且易于维护的自动化数据采集系统。

crawl4ai

🔥🕷️ Crawl4AI: Open-source LLM Friendly Web Crawler & Scrapper

项目地址：https://gitcode.com/GitHub_Trending/craw/crawl4ai

登录后查看全文