OpenClaw 新手教程

OpenClaw是一个强大的数据抓取工具，可以帮助你从各种网站提取结构化数据,本教程将带你从零开始学习OpenClaw的基本使用。

安装OpenClaw

方法1：使用pip安装

pip install openclaw

方法2：从源码安装

git clone https://github.com/openclaw/openclaw.git
cd openclaw
pip install -e .

基础概念

1 主要组件

• Crawler：负责网页爬取
• Parser：负责解析HTML/XML
• Storage：负责数据存储
• Scheduler：负责任务调度

第一个爬虫项目

1 创建爬虫脚本

# my_first_spider.py
from openclaw import Claw, Request
from openclaw.items import Item, Field
# 定义数据模型
class BookItem(Item):= Field()
    author = Field()
    price = Field()
    description = Field()
# 创建爬虫类
class BookSpider(Claw):
    name = "book_spider"
    start_urls = ["https://books.toscrape.com/"]
    async def parse(self, response):
        # 提取书籍信息
        books = response.xpath('//article[@class="product_pod"]')
        for book in books:
            item = BookItem()
            item['title'] = book.xpath('.//h3/a/@title').get()
            item['author'] = book.xpath('.//p[@class="author"]/text()').get()
            item['price'] = book.xpath('.//p[@class="price_color"]/text()').get()
            yield item
        # 处理分页
        next_page = response.xpath('//li[@class="next"]/a/@href').get()
        if next_page:
            yield Request(url=response.urljoin(next_page), callback=self.parse)
if __name__ == "__main__":
    spider = BookSpider()
    spider.run()

2 运行爬虫

python my_first_spider.py

配置爬虫

1 基础配置

# settings.py
CONCURRENT_REQUESTS = 16
DOWNLOAD_DELAY = 1
USER_AGENT = "MySpider/1.0 (+http://www.example.com)"
ROBOTSTXT_OBEY = True

2 中间件配置

# middlewares.py
from openclaw.middlewares import DownloaderMiddleware
class CustomMiddleware(DownloaderMiddleware):
    async def process_request(self, request):
        # 在请求前处理
        request.headers['X-Custom-Header'] = 'value'
        return request
    async def process_response(self, request, response):
        # 在响应后处理
        if response.status == 403:
            # 处理403错误
            pass
        return response

数据处理与存储

1 数据清洗管道

# pipelines.py
from openclaw.pipelines import ItemPipeline
class CleanPipeline(ItemPipeline):
    async def process_item(self, item):
        # 清洗价格数据
        if 'price' in item:
            item['price'] = item['price'].replace('£', '').strip()
        # 清洗作者数据
        if 'author' in item:
            item['author'] = item['author'].strip()
        return item
class JSONPipeline(ItemPipeline):
    def __init__(self):
        import json
        self.json = json
    async def process_item(self, item):
        # 将数据保存为JSON格式
        with open('books.json', 'a') as f:
            f.write(self.json.dumps(dict(item)) + '\n')
        return item

2 数据库存储

# db_pipeline.py
import sqlite3
from openclaw.pipelines import ItemPipeline
class SQLitePipeline(ItemPipeline):
    def __init__(self):
        self.conn = sqlite3.connect('books.db')
        self.create_table()
    def create_table(self):
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS books (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT,
                author TEXT,
                price REAL,
                description TEXT
            )
        ''')
        self.conn.commit()
    async def process_item(self, item):
        cursor = self.conn.cursor()
        cursor.execute('''
            INSERT INTO books (title, author, price, description)
            VALUES (?, ?, ?, ?)
        ''', (
            item.get('title'),
            item.get('author'),
            item.get('price'),
            item.get('description', '')
        ))
        self.conn.commit()
        return item

高级功能

1 处理JavaScript渲染的页面

from openclaw.splash import SplashRequest
class JSEnabledSpider(Claw):
    name = "js_spider"
    def start_requests(self):
        yield SplashRequest(
            url="https://example.com/dynamic-content",
            callback=self.parse,
            args={'wait': 2}  # 等待2秒让JS执行
        )
    async def parse(self, response):
        # 现在可以解析JavaScript生成的内容
        data = response.xpath('//div[@class="dynamic-content"]/text()').get()
        yield {'data': data}

2 使用代理

# 在settings.py中配置
PROXY_POOL = [
    'http://proxy1:port',
    'http://proxy2:port',
    'http://proxy3:port'
]
# 或者在爬虫中动态设置
class ProxySpider(Claw):
    name = "proxy_spider"
    async def start_requests(self):
        proxy = "http://your-proxy:port"
        yield Request(
            url="https://example.com",
            callback=self.parse,
            meta={'proxy': proxy}
        )

3 处理登录

class LoginSpider(Claw):
    name = "login_spider"
    def start_requests(self):
        # 首先访问登录页面
        yield Request(
            url="https://example.com/login",
            callback=self.login
        )
    async def login(self, response):
        # 提交登录表单
        return FormRequest.from_response(
            response,
            formdata={
                'username': 'your_username',
                'password': 'your_password'
            },
            callback=self.after_login
        )
    async def after_login(self, response):
        # 检查登录是否成功
        if "Logout" in response.text:
            # 登录成功，开始爬取
            yield Request(
                url="https://example.com/protected-page",
                callback=self.parse_protected
            )
    async def parse_protected(self, response):
        # 解析受保护页面的内容
        pass

错误处理与调试

1 错误处理

class RobustSpider(Claw):
    name = "robust_spider"
    async def parse(self, response):
        try:
            # 尝试提取数据
            data = response.xpath('//div[@class="content"]/text()').get()
            if not data:
                # 数据为空时的处理
                self.logger.warning("No data found on page: %s", response.url)
            yield {'data': data}
        except Exception as e:
            # 记录错误但继续运行
            self.logger.error("Error parsing %s: %s", response.url, str(e))

2 调试模式

# 启用详细日志
python my_spider.py --loglevel DEBUG
# 限制请求数量进行测试
python my_spider.py --max-requests 10

最佳实践

1 遵守robots.txt

# 始终尊重robots.txt
ROBOTSTXT_OBEY = True

2 设置合理的延迟

# 避免对服务器造成压力
DOWNLOAD_DELAY = 1  # 1秒延迟
CONCURRENT_REQUESTS = 4  # 限制并发数

3 使用User-Agent轮换

# 在settings.py中
USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
    # 添加更多User-Agent
]

项目结构示例

my_openclaw_project/
├── spiders/
│   ├── __init__.py
│   ├── books.py
│   └── news.py
├── items.py
├── pipelines.py
├── middlewares.py
├── settings.py
└── requirements.txt

下一步学习

1. 阅读官方文档：查看OpenClaw的官方文档了解更多高级功能
2. 学习XPath/CSS选择器：提高数据提取能力
3. 了解反爬虫策略：学习如何应对网站的反爬虫机制
4. 学习分布式爬虫：了解如何使用多个服务器进行大规模爬取

常见问题

Q: 如何处理验证码？

A: 可以使用第三方验证码识别服务,或添加人工识别机制。

Q: 爬虫被屏蔽了怎么办？

A: 尝试使用代理IP、降低爬取频率、更换User-Agent。

Q: 如何增量爬取？

A: 记录已爬取的URL或使用数据库记录爬取状态。

Q: 如何处理大量数据？

A: 使用数据库存储,考虑分布式爬虫架构。

希望这个教程能帮助你开始使用OpenClaw！记得始终遵守法律法规和网站的robots.txt规则,合理使用爬虫技术。

标签： OpenClaw 新手教程