Crawl4AI 快速入门指南：异步网页爬取与AI数据提取实战

Crawl4AI 快速入门指南：异步网页爬取与AI数据提取实战项目概述Crawl4AI 是一个专为AI应用设计的现代化异步网页爬取框架，它简化了从网页中提取结构化数据的过程，特别适合为大型语言模型(LLMs)提供训练数据或实时信息获取。该框架集成了智能内容提取、动态页面处理等高级功能，让开发者能够轻松应对各种复杂的网页爬取场景。环境准备安装依赖首先需要安装Crawl4AI核心库及其依赖项...

卢迁铎Renee

1779人浏览 · 2025-05-30 09:08:00

卢迁铎Renee · 2025-05-30 09:08:00 发布

Crawl4AI 快速入门指南：异步网页爬取与AI数据提取实战

【免费下载链接】crawl4ai 🔥🕷️ Crawl4AI: Open-source LLM Friendly Web Crawler & Scrapper 项目地址: https://gitcode.com/GitHub_Trending/craw/crawl4ai

项目概述

Crawl4AI 是一个专为AI应用设计的现代化异步网页爬取框架，它简化了从网页中提取结构化数据的过程，特别适合为大型语言模型(LLMs)提供训练数据或实时信息获取。该框架集成了智能内容提取、动态页面处理等高级功能，让开发者能够轻松应对各种复杂的网页爬取场景。

环境准备

安装依赖

首先需要安装Crawl4AI核心库及其依赖项：

!pip install crawl4ai
!pip install nest_asyncio
!playwright install

由于Crawl4AI基于异步IO设计，我们需要启用nest_asyncio来兼容Jupyter等环境：

import asyncio
import nest_asyncio
nest_asyncio.apply()

基础爬取示例

简单网页内容获取

最基本的爬取功能只需要几行代码：

from crawl4ai import AsyncWebCrawler

async def simple_crawl():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://www.nbcnews.com/business",
            bypass_cache=True
        )
        print(result.markdown.raw_markdown[:500])  # 打印前500个字符

asyncio.run(simple_crawl())

这段代码会：

创建异步爬虫实例
访问指定URL
返回包含原始Markdown格式的内容
打印前500个字符作为示例

高级功能探索

动态内容处理

现代网页大量使用JavaScript动态加载内容，Crawl4AI提供了多种方式处理这种情况：

async def crawl_dynamic_content():
    async with AsyncWebCrawler(verbose=True) as crawler:
        js_code = [
            "const loadMoreButton = Array.from(document.querySelectorAll('button'))"
            ".find(button => button.textContent.includes('Load More'));"
            "loadMoreButton && loadMoreButton.click();"
        ]
        result = await crawler.arun(
            url="https://www.nbcnews.com/business",
            js_code=js_code,
            bypass_cache=True,
        )
        print(result.markdown.raw_markdown[:500])

关键参数说明：

js_code: 执行自定义JavaScript代码
wait_for: 可指定CSS选择器或函数，等待特定元素出现
verbose: 开启详细日志

内容清洗与优化

Crawl4AI内置了智能内容清洗功能：

async def clean_content():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://janineintheworld.com/places-to-visit-in-central-mexico",
            excluded_tags=['nav', 'footer', 'aside'],  # 排除导航等非主要内容
            remove_overlay_elements=True,  # 移除弹窗等覆盖元素
            word_count_threshold=10,  # 内容块最小字数阈值
        )
        print(f"清洗前长度: {len(result.markdown.raw_markdown)}")
        print(f"清洗后长度: {len(result.markdown.fit_markdown)}")
        print(result.markdown.fit_markdown[:1000])

链接分析与过滤

Crawl4AI可以智能分类和分析页面链接：

async def link_analysis():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://www.nbcnews.com/business",
            exclude_external_links=True,  # 排除外部链接
            exclude_social_media_links=True  # 排除社交媒体链接
        )
        print(f"内部链接数: {len(result.links['internal'])}")
        print(f"外部链接数: {len(result.links['external'])}")
        for link in result.links['internal'][:5]:
            print(f"链接: {link['href']}\n文本: {link['text']}\n")

媒体资源处理

提取页面中的图片等媒体资源：

async def media_handling():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://www.nbcnews.com/business",
            exclude_external_images=False,  # 包含外部图片
            screenshot=True  # 生成页面截图
        )
        for img in result.media['images'][:5]:
            print(f"图片URL: {img['src']}, 描述: {img['alt']}, 相关性评分: {img['score']}")

高级技巧

使用钩子自定义流程

Crawl4AI提供了多种钩子(Hook)来扩展功能：

async def custom_hook_workflow():
    async with AsyncWebCrawler() as crawler:
        # 设置导航前钩子
        crawler.crawler_strategy.set_hook(
            "before_goto", 
            lambda page: print("[Hook] 准备导航...")
        )
        
        result = await crawler.arun(
            url="https://crawl4ai.com",
            bypass_cache=True
        )
        print(result.markdown.raw_markdown[:500])

可用钩子包括：

on_browser_created: 浏览器创建时触发
before_goto: 页面导航前触发
after_goto: 页面导航后触发
on_execution_started: JavaScript执行前触发
before_return_html: 返回HTML前触发

会话保持爬取

对于需要登录或多步骤操作的场景，可以使用会话保持：

async def session_based_crawl():
    async with AsyncWebCrawler() as crawler:
        # 第一步操作
        await crawler.arun(url="https://example.com/login")
        
        # 第二步操作，保持相同会话
        result = await crawler.arun(url="https://example.com/dashboard")
        print(result.markdown.raw_markdown)