FusionCrawler

统一的 AI 驱动全栈爬虫平台，融合 Scrapling / Firecrawl / MediaCrawler / Crawl4AI / EasySpider / CloakBrowser 六个顶级开源项目的核心能力。

六层架构

Layer 0: 基础设施层    — 存储 / 缓存 / 代理 / 日志 / 配置
Layer 1: 浏览器引擎层  — HTTP / Playwright / CloakBrowser 三引擎
Layer 2: 内容处理层    — 自适应解析 / LLM 提取 / 过滤 / 格式转换
Layer 3: 智能调度层    — AI Agent / 任务调度 / 爬取策略 / 断点续爬
Layer 4: 用户交互层    — API / SDK / CLI / 可视化设计器
Layer 5: 平台适配层    — 通用 Web + 10 个主流平台适配器

平台适配层

支持 10 个主流平台 + 通用 Web 适配器：

快速开始

安装

pip install -e ".[dev]"

Python SDK

from fusion_crawler import FusionCrawler

client = FusionCrawler()
result = client.scrape("https://example.com")
print(result.markdown)

平台适配器使用

from fusion_crawler.platform.registry import AdapterRegistry

registry = AdapterRegistry()

# 根据 URL 自动匹配适配器
adapter = registry.match_url("https://weibo.com/123")
if adapter:
    post = await adapter.get_post_detail("123")
    print(post.title, post.content[:100])

# 搜索内容
bilibili = registry.get("bilibili")
videos = await bilibili.search("Python教程")
for v in videos:
    print(v.title, v.url)

CLI

# 爬取网页
fusion-crawler https://example.com

# 使用特定平台适配器
fusion-crawler --platform weibo "关键词"

# 启动 API 服务
fusion-crawler api
# 访问 http://localhost:8000/docs

API 服务

fusion-crawler api
curl http://localhost:8000/api/v1/health

LLM 补充模块

FusionCrawler 提供四个可选 LLM/Agent 增强模块：

from fusion_crawler.processing.llm.schema import SchemaValidator
from fusion_crawler.processing.llm.cost_controller import LLMCostController
from fusion_crawler.orchestrator.resume_manager import CrawlResumeManager
from fusion_crawler.orchestrator.tool_registry import ToolRegistry, ToolParameter

# Schema 校验
validator = SchemaValidator()
report = validator.validate([{"name": "Alice", "age": "30"}],
    {"properties": {"name": {"type": "string"}, "age": {"type": "integer"}}})
print(report.valid)  # True (age 自动从 string 转为 int)

# LLM 成本控制
controller = LLMCostController(monthly_budget_usd=50.0)
controller.track_call("gpt-4o-mini", prompt_tokens=500, completion_tokens=200)
print(controller.stats().total_cost)

# 断点续爬
manager = CrawlResumeManager()
await manager.initialize()
task_id = await manager.save_checkpoint(Checkpoint(
    task_id="task-1", seed_urls=["https://example.com"],
    visited_urls={"https://example.com/a"}, pending_urls=["https://example.com/b"],
))

# 工具注册
registry = ToolRegistry()
@registry.register(name="scrape", description="抓取页面",
    parameters=[ToolParameter(name="url", type="string", required=True)])
async def scrape(url: str) -> dict:
    return {"url": url, "status": 200}

测试

# 全量测试（936 用例）
python -m pytest tests/ -v

# 仅平台适配层测试
python -m pytest tests/platform/ -v

# 指定模块测试
python -m pytest tests/processing/llm/ tests/orchestrator/ -v

# 集成测试
python -m pytest tests/test_platform_integration.py -v

Docker 部署

快速启动（单命令）

# 1. 复制环境配置
cp .env.example .env

# 2. 构建并启动
make deploy
# 或: docker-compose up -d

# 3. 访问 http://localhost:8000

含 Nginx 反向代理

make up-nginx
# 或: docker-compose --profile with-nginx up -d

生产环境部署

# 完全重建（无缓存）
make deploy-clean

# 查看日志
make logs

# 进入容器
make shell

环境变量

前端开发

cd frontend
npm install
npm run dev     # 开发模式，端口 5173
npm run build   # 生产构建

项目结构

fusion-crawler/
├── fusion_crawler/          # 核心 Python 包
│   ├── api/                 # API 服务层（FastAPI）
│   ├── cli/                 # CLI 命令行（Typer）
│   ├── core/                # 核心接口 / 数据模型 / 异常体系
│   ├── orchestrator/        # 智能调度层（策略 / Agent / 工具注册 / 断点续爬）
│   ├── processing/          # 内容处理层（解析 / 过滤 / LLM 提取 / 格式转换）
│   ├── browser/             # 浏览器引擎层（HTTP / Playwright / 反检测）
│   ├── platform/            # 平台适配层（11 个适配器）
│   └── infrastructure/      # 基础设施层（存储 / 缓存 / 代理 / 日志）
├── frontend/                # 可视化设计器（React + Vite + TypeScript）
├── tests/                   # 测试（936 用例，覆盖所有模块）
├── docs/                    # 文档
├── deploy/                  # 部署配置（Docker 入口 / Nginx）
├── .github/workflows/       # CI/CD 工作流
├── pyproject.toml           # 项目配置与依赖管理
└── docker-compose.yml       # 容器化部署

开发阶段

许可证

MIT