Crawl4AI 爬虫工具（一）使用 Docker Compose 部署和最小实践

前言

正好想学习一种 Selenium + Chrome 以外的爬虫工具，作为 AI 工作流中的一个环境，就选 LLM 友好的 Crawl4AI 吧。

方案概述

1. 安装 Docker 环境
2. 使用 docker-compose 部署容器
3. 编写 Python3 脚本进行测试

操作步骤

一、安装 Docker 环境

参考：Ubuntu 20.04 从官方源安装最新的 Docker

二、使用 docker-dompose 部署容器

官方仓库：unclecode/crawl4ai

官方的 docker-compose.yaml 文件中涉及到很多 AI 提供商的 API Key 配置，我们本次的最小实践暂时不涉及 AI 功能因此跳过。
最终我的文件配置：

version: '3.8'

services:
  crawl4ai:
    image: unclecode/crawl4ai:all-amd64
    container_name: crawl4ai
    restart: always
    ports:
      - "11235:11235"
    environment:
      - CRAWL4AI_API_TOKEN=YourApiToken

我的服务部署在公网上，不希望所有人都可以使用，因此添加 CRAWL4AI_API_TOKEN 来进行认证。
之后除了 /health，访问其他接口都需要提供该 Token。

启动容器：

docker-compose up -d

如果出现下面的错误：

exec /usr/local/bin/uvicorn: exec format error

说明你的 Docker 容器镜像架构和系统的不符，unclecode/crawl4ai:all 有 all-arm64 和 all-amd64 两种，需要视实际情况选择。

启动完成后，访问 http://IP:11235/health 测试一下：

三、编写 Python3 脚本进行测试

因为我们没有在本地部署 crawl4ai，因此不需要安装 crawl4ai 的 Python 包。
使用 requests 构造响应的请求参数，之后调用 crawl4ai 容器的远程 API 即可实现爬取。
测试脚本：

import time
import json
import requests


CRAWL4AI_HOST = "127.0.0.1"
CRAWL4AI_PORT = 11235
CRAWL4AI_API_TOKEN = "YourApiToken"


def main():
    # 1. 发布爬取任务
    data = {
        "urls": ["https://www.baidu.com"],
        "browser_config": {},
        "crawler_config": {},
    }
    url = "{}://{}:{}/crawl".format("https" if CRAWL4AI_PORT == 443 else "http", CRAWL4AI_HOST, CRAWL4AI_PORT)
    headers = {
        "Authorization": f"Bearer {CRAWL4AI_API_TOKEN}",
        "Content-Type": "application/json",
    }
    try:
        response = requests.post(url, json=data, headers=headers)
        # print(response.json())
    except Exception as e:
        print(e)
    # 2. 获取 Task ID
    task_id = response.json()["task_id"]
    print(f"任务 ID 为 {task_id}")
    # 3. 获取爬取结果
    url = "{}://{}:{}/task/{}".format("https" if CRAWL4AI_PORT == 443 else "http", CRAWL4AI_HOST, CRAWL4AI_PORT, task_id)
    for _ in range(60):
        try:
            response = requests.get(url, headers=headers)
            if response.json()["status"] in ["pending", "processing"]:
                time.sleep(1)
            else:
                print("网站标题：{}".format(response.json()["results"][0]["metadata"]["title"]))
                with open("response.json", "w") as f:
                    f.write(json.dumps(response.json(), indent=4, ensure_ascii=False))
                break
        except Exception as e:
            print(e)


if __name__ == "__main__":
    main()

执行结果：

结束。