OpenClaw Skill示例：从零构建你的第一个智能抓取应用

1次阅读

没有评论

共计 2178 个字符，预计需要花费 6 分钟才能阅读完成。

在动态网页抓取中，传统爬虫常遇到这些痛点：

页面结构频繁变动导致 XPath/CSS 选择器失效
反爬机制触发后缺乏标准化处理流程
需要自行管理代理池和请求频率控制

OpenClaw 通过组件化设计解决了这些问题：

技能市场：可复用抓取逻辑封装为 Skill
智能调度：自动处理限流和重试
可视化调试：实时查看数据抽取过程

安装 Python SDK
```
pip install openclaw-sdk --upgrade
```

配置访问密钥

# ~/.openclaw/config
[default]
api_key = "your_api_key"
region = "ap-southeast-1"

申请测试权限
前往开发者门户申请沙箱环境 access_key

{
  "skill_name": "news_crawler",
  "version": "1.0.0",
  "entry_point": "handler.py",
  "triggers": [{
    "type": "http",
    "config": {
      "method": "GET",
      "path": "/news/{source}"
    }
  }]
}

# 支持 CSS 选择器和 XPath 混合使用
extract_rules = {
  "title": {
    "selector": "div.article h1::text",
    "required": True
  },
  "content": {"xpath": "//div[@class='article-body']//p/text()"}
}

import math
from datetime import timedelta

def exponential_backoff(retry_count):
    """
    指数退避算法
    :param retry_count: 当前重试次数
    :return: 等待秒数
    """
    return min(30, math.pow(2, retry_count) + random.random())

async def fetch_page(url):
    retry = 0
    while retry < 3:
        try:
            return await http.get(url)
        except Exception as e:
            await asyncio.sleep(exponential_backoff(retry))
            retry += 1

在开发者控制台开启Preserve log
过滤 XHR 请求查看 API 调用
检查响应头中的X-RateLimit-Remaining

403 错误：
检查请求头是否包含Authorization
验证时间戳是否在有效期内
数据脱敏：
使用 .raw_html 模式获取原始数据
配置 data_masking: false 参数

# 令牌桶算法实现
class RateLimiter:
    def __init__(self, rate):
        self.tokens = rate
        self.last_check = time.time()

    async def acquire(self):
        now = time.time()
        elapsed = now - self.last_check
        self.tokens += elapsed * self.rate
        self.tokens = min(self.tokens, self.rate)
        self.last_check = now

        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

# config/filters.yaml
rules:
  - pattern: "\\d{4}-\\d{4}-\\d{4}-\\d{4}"
    repl: "[CREDIT_CARD]"
  - pattern: "\\d{3}-\\d{2}-\\d{4}"
    repl: "[SSN]"

from typing import Dict, Optional

async def extract_article(
    html: str, 
    rules: Dict[str, dict]
) -> Optional[Dict]:
    """
    根据规则提取页面内容
    :param html: 原始 HTML
    :param rules: 抽取规则字典
    :return: 结构化数据或 None
    """
    try:
        result = {}
        for field, config in rules.items():
            if config.get("selector"):
                result[field] = css_select(html, config["selector"])
            elif config.get("xpath"):
                result[field] = xpath_select(html, config["xpath"])
        return result
    except Exception as e:
        logger.error(f"Extraction failed: {str(e)}")
        return None