网页抓取引擎

核心功能: 本技能提供中文交互、化工作流场景等能力。

网页抓取引擎

快速开始

1. 确认运行环境满足依赖说明中的要求
2. 在AI Agent对话中调用本技能,提供必要的输入参数
3. 检查输出结果,根据需要进行后续处理

详细的输入输出格式请参考下方章节说明。 基于 Firecrawl / Playwright / Crawl4AI 实现全流程网页数据采集。从搜索发现到结构化提取,从单页抓取到多步骤浏览器交互,无需维护爬虫基础设施。同时提供开源国内替代方案(Crawl4AI),降低对海外 API 的依赖。

主要能力

1. 搜索与发现:关键词搜索发现网页、站点地图扫描、URL 队列构建
2. 单页抓取与结构化提取:Markdown/HTML/纯文本输出 + JSON Schema 自动提取
3. 批量异步爬取:深度控制、路径过滤、速率限制、结果去重
4. 多步骤浏览器交互:点击/表单填写/翻页/登录状态保持
5. 数据处理与导出:清洗、结构化、CSV/JSON/Parquet 导出、数据库写入

部署说明

1. 确认运行环境满足依赖说明中的要求
2. 在AI Agent对话中调用本技能,提供必要的输入参数
3. 检查输出结果,根据需要进行后续处理

使用场景

• 大规模分布式爬虫(单机能力有限,需要 Scrapy Cluster 等架构)
• 实时数据流(本 Skill 为批量采集,非流式)
• APP 内数据抓取(需要 Appium 等移动端工具)
• 加密/混淆内容的破解(仅采集公开可访问内容)
• 违反 robots.txt 或法律法规的爬取(需遵守合规要求)
• 需要登录态的私有数据(技术上可行,但需注意合规与隐私)

使用说明

Step 1: 需求分析与方案选择

1. 明确采集目标:URL/关键词/数据字段
2. 合规检查:检查目标站点 robots.txt、ToS、数据版权
3. 方案选择:
   • 简单页面(无 JS 渲染)→ Firecrawl scrape / requests + BeautifulSoup
   • JS 渲染页面 → Firecrawl / Playwright
   • 大规模爬取 → Crawl4AI(开源,无 API 费用)
   • 多步骤交互 → Playwright
4. 频率规划:避免高频请求导致 IP 封禁

Step 2: 搜索发现

1. 关键词搜索
   • Firecrawl: POST /v1/search 输入关键词,返回相关网页
   • Crawl4AI: 内置搜索 + URL 提取
   • 参数:query/limit/scrapeOptions
   • 输出:URL 列表 + 标题 + 摘要
2. 站点地图发现
   • 获取目标站点 sitemap.xml
   • 过滤不需要的页面
   • 构建爬取队列

Step 3: 单页抓取

1. 基础抓取
   • Firecrawl: POST /v1/scrape 抓取单个 URL
   • Playwright: page.goto(url) + page.content()
   • 输出:Markdown/HTML/纯文本
   • 自动处理 JS 渲染
2. 结构化提取
   • 使用 Schema 定义提取字段(JSON Schema)
   • 自动识别页面内容
   • 输出 JSON 格式
3. 截图
   • 全页截图
   • 可视化验证

Step 4: 批量爬取

1. 异步爬取
   • Firecrawl: POST /v1/crawl 提交 URL 列表,返回 job ID
   • Crawl4AI: 内置并发爬取
2. 爬取策略
   • 深度控制:仅首页/2层/全站
   • 路径过滤:include/exclude 规则
   • 速率限制:并发数控制 + 请求间隔
3. 结果处理
   • 统一格式输出
   • 去重(URL 规范化 + 内容哈希)
   • 导出 CSV/JSON

Step 5: 多步骤浏览器交互

1. 点击操作:点击按钮/链接,等待页面加载,处理弹窗
2. 表单填写:填充输入框,选择下拉菜单,提交表单
3. 翻页:点击"下一页",等待新数据加载,收集每页数据
4. 登录状态:注入 Cookie/Token,处理认证流程,会话保持

Step 6: 数据处理与导出

1. 清洗:移除 HTML 标签,去除空白字符,统一编码(UTF-8)
2. 结构化:JSON Schema 映射,字段类型转换,数据验证
3. 存储:导出 CSV/JSON/Parquet,写入数据库,上传 OSS/COS

国内外抓取方案对照

用法示例

示例1: 竞品价格监控(输入→输出)

输入:

参数说明

用户请求:抓取3个竞品电商网站的产品价格,每周更新
竞品: 京东、淘宝、拼多多
商品: iPhone 15 Pro 256GB

输出(scrape-result.json):

from firecrawl import FirecrawlApp
import json
app = FirecrawlApp(api_key=os.environ['FIRECRAWL_API_KEY'])
schema = {
    "type": "object",
    "properties": {
        "product_name": {"type": "string"},
        "price": {"type": "number"},
        "currency": {"type": "string"},
        "availability": {"type": "boolean"},
        "platform": {"type": "string"},
        "url": {"type": "string"},
        "scraped_at": {"type": "string"}
    }
results = [
    {
        "product_name": "Apple iPhone 15 Pro 256GB",
        "price": 7999.00,
        "currency": "CNY",
        "availability": true,
        "platform": "京东",
        "url": "https://item.jd.com/100未指定.html",
        "scraped_at": "2026-01-15T10:00:00Z"
    },
    {
        "product_name": "Apple iPhone 15 Pro 256GB",
        "price": 7599.00,
        "currency": "CNY",
        "availability": true,
        "platform": "拼多多",
        "url": "https://mobile.yangkeduo.com/未指定.html",
        "scraped_at": "2026-01-15T10:00:00Z"
    }
]
import pandas as pd
df = pd.DataFrame(results)
df.to_csv('price_monitor.csv', index=False)
report = f"""
| 平台 | 价格 | 可用性 |
|:---:|:---:|:---:|
| 京东 | ¥7999 | 有货 |
| 拼多多 | ¥7599 | 有货 |
最低价: 拼多多 ¥7599
最高价: 京东 ¥7999
价差: ¥400 (5%)
"""

示例2: Crawl4AI 国内开源替代(输入→输出)

输入:

用户请求: 使用开源工具抓取某新闻网站最新文章
要求: 不依赖海外 API,完全本地运行
目标: https://news.example.com

输出(scrape.py):

from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction_strategy import JsonCssExtractionStrategy
import asyncio
import json
async def scrape_news():
    async with AsyncWebCrawler() as crawler:
        extraction_strategy = JsonCssExtractionStrategy({
            "name": "News Articles",
            "baseSelector": "article.news-item",
            "fields": [
                {"name": "title", "selector": "h2.title", "type": "text"},
                {"name": "url", "selector": "a", "type": "attribute", "attribute": "href"},
                {"name": "date", "selector": ".date", "type": "text"},
                {"name": "summary", "selector": ".summary", "type": "text"}
            ]
        })
        result = await crawler.arun(
            url="https://news.example.com",
            extraction_strategy=extraction_strategy,
            bypass_cache=True
        )
        articles = json.loads(result.extracted_content)
        print(f"抓取到 {len(articles)} 篇文章")
        with open('articles.json', 'w', encoding='utf-8') as f:
            json.dump(articles, f, ensure_ascii=False, indent=2)
asyncio.run(scrape_news())

示例3: 多步骤表单提交(输入→输出)

输入:

用户请求: 自动填写某网站的查询表单,翻页采集结果
目标: https://search.example.com
步骤: 1)填写关键词 2)选择日期 3)点击搜索 4)翻页采集

输出(scrape.py):

from playwright.async_api import async_playwright
import pandas as pd
import asyncio
async def scrape_with_form():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        await page.goto('https://search.example.com')
        await page.fill('#keyword', '人工智能')
        await page.select_option('#date-range', 'last-30-days')
        await page.click('button[type="submit"]')
        await page.wait_for_selector('.result-item')
        all_results = []
        while True:
            items = await page.query_selector_all('.result-item')
            for item in items:
                title = await (await item.query_selector('.title')).inner_text()
                url = await (await item.query_selector('a')).get_attribute('href')
                all_results.append({'title': title, 'url': url})
            next_btn = await page.query_selector('.next-page:not([disabled])')
            if not next_btn:
                break
            await next_btn.click()
            await page.wait_for_selector('.result-item')
        df = pd.DataFrame(all_results)
        df.to_csv('search_results.csv', index=False)
        print(f"采集完成,共 {len(all_results)} 条结果")
        await browser.close()
asyncio.run(scrape_with_form())

异常恢复方案

环境要求

运行环境

• Agent平台: Claude Code / Cursor / Codex / Gemini CLI / Windsurf 等支持SKILL.md的任意Agent
• 操作系统: Windows / macOS / Linux
• 运行时: Python 3.10+ 或 Node.js 18+

第三方依赖

API Key 配置

• Firecrawl: API Key 从 firecrawl.dev 获取(免费 500 页/月)
• Crawl4AI: 无需 API Key(开源,完全本地)
• Playwright: 无需 API Key(开源)
• 安全要求:
  • API Key 仅存储在环境变量,不写入源码
  • 抓取的 Cookie/Token 不写入日志
  • 敏感数据(用户信息)脱敏后存储

可用性分类

• 分类: MD+EXEC
• 说明: 纯Markdown方法论,需 exec 能力运行 Python 脚本

案例展示

以下案例展示了skill的工作流程和预期输出效果，由LLM按照skill定义的流程生成。

案例1: 竞品价格监控(Firecrawl+JSON Schema提取+CSV导出+对比报告)

输入:

目标: 监控3个电商平台iPhone 15 Pro 256GB价格
平台: 京东、拼多多、天猫
Schema字段: product_name, price, currency, availability, platform, url, scraped_at
频率: 每周一次
输出: CSV文件 + Markdown价格对比报告

LLM生成输出 (output/price-monitor/scrape.py):

import os
import json
import csv
from datetime import datetime
from firecrawl import FirecrawlApp
environ['FIRECRAWL_API_KEY'])
price_schema = {
    "type": "object",
    "properties": {
        "product_name": {"type": "string", "description": "商品名称"},
        "price": {"type": "number", "description": "当前售价"},
        "original_price": {"type": "number", "description": "原价"},
        "currency": {"type": "string", "description": "货币"},
        "availability": {"type": "boolean", "description": "是否有货"},
        "discount": {"type": "string", "description": "优惠信息"}
    },
    "required": ["product_name", "price", "currency", "availability"]
}
targets = [
    {"platform": "京东", "url": "https://item.jd.com/100074466498.html"},
    {"platform": "拼多多", "url": "https://mobile.yangkeduo.com/goods.html?goods_id=420001234567"},
    {"platform": "天猫", "url": "https://detail.tmall.com/item.htm?id=678901234567"},
]
results = []
scraped_at = datetime.utcnow().isoformat() + "Z"
for target in targets:
    try:
        response = app.scrape_url(target['url'], params={
            'formats': ['json'],
            'jsonOptions': {
                'prompt': '提取商品名称、当前售价、原价、货币、是否有货、优惠信息',
                'schema': price_schema
            },
            'waitFor': 3000  # 等待JS渲染
        })
        data = response.get('json', {})
        data['platform'] = target['platform']
        data['url'] = target['url']
        data['scraped_at'] = scraped_at
        results.append(data)
        print(f"[{target['platform']}] {data.get('product_name', 'N/A')} - ¥{data.get('price', 'N/A')}")
    except Exception as e:
        print(f"[{target['platform']}] 抓取失败: {e}")
        results.append({
            'platform': target['platform'],
            'url': target['url'],
            'scraped_at': scraped_at,
            'error': str(e)
        })
csv_file = 'price_monitor.csv'
fieldnames = ['platform', 'product_name', 'price', 'original_price', 'currency',
              'availability', 'discount', 'url', 'scraped_at', 'error']
with open(csv_file, 'w', newline='', encoding='utf-8-sig') as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore')
    writer.writeheader()
    writer.writerows(results)
print(f"\nCSV导出: {csv_file}")
valid_results = [r for r in results if 'price' in r]
if valid_results:
    prices = [(r['platform'], r['price']) for r in valid_results]
    min_platform, min_price = min(prices, key=lambda x: x[1])
    max_platform, max_price = max(prices, key=lambda x: x[1])
    report = f"""# 价格监控报告
**抓取时间**: {scraped_at}
**商品**: iPhone 15 Pro 256GB
| 平台 | 商品名称 | 售价 | 原价 | 是否有货 | 优惠 |
|:------:|--------|:-------|:------:|--------|:-------|
"""
    for r in valid_results:
        report += f"| {r['platform']} | {r.get('product_name','N/A')} | ¥{r.get('price','N/A')} | ¥{r.get('original_price','-')} | {'有货' if r.get('availability') else '缺货'} | {r.get('discount','-')} |\n"
    report += f"""
- **最低价**: {min_platform} ¥{min_price}
- **最高价**: {max_platform} ¥{max_price}
- **价差**: ¥{max_price - min_price} ({(max_price - min_price) / min_price * 100:.1f}%)
"""
    if max_price - min_price > 200:
        report += f"- 价差较大(¥{max_price - min_price}),建议在{min_platform}购买\n"
    else:
        report += "- 各平台价格接近,建议选择物流最快的平台\n"
    report += "- 拼多多价格通常最低,但需注意百亿补贴商品真伪\n"
    report += "- 京东自营售后最有保障,适合追求稳妥的用户\n"
    with open('price_report.md', 'w', encoding='utf-8') as f:
        f.write(report)
    print(f"报告导出: price_report.md")

LLM生成输出 (output/price-monitor/price_monitor.csv):

platform,product_name,price,original_price,currency,availability,discount,url,scraped_at
京东,Apple iPhone 15 Pro 256GB 钛金属,7999.00,8999.00,CNY,True,plus会员-100,item.jd.com/100074466498.html,2026-01-15T10:00:00Z
拼多多,Apple iPhone 15 Pro 256GB,7599.00,8999.00,CNY,True,百亿补贴,yangkeduo.com/goods.html?id=420001234567,2026-01-15T10:00:00Z
天猫,Apple/苹果 iPhone 15 Pro 256GB,7799.00,8999.00,CNY,True,88VIP-50,detail.tmall.com/item.htm?id=678901234567,2026-01-15T10:00:00Z

效果验证: ✓JSON Schema定义6字段(含required约束) ✓Firecrawl scrape_url含waitFor等待JS渲染 ✓异常处理try-catch记录错误不影响其他平台 ✓CSV导出使用utf-8-sig防Excel中文乱码 ✓Markdown报告含价格对比表+最低/最高价分析+购买建议 ✓价差百分比计算

案例2: Crawl4AI新闻网站批量爬取(开源+CSS提取+去重+JSON导出)

输入:

目标: 抓取36氪最新科技文章
工具: Crawl4AI(开源,无API费用,国内可用)
要求: 完全本地运行,不依赖海外服务
提取字段: 标题、链接、发布时间、摘要、作者
去重: URL规范化 + 内容哈希
导出: JSON + Markdown摘要

LLM生成输出 (output/news-scraper/scrape_36kr.py):

import asyncio
import json
import hashlib
from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction_strategy import JsonCssExtractionStrategy
extraction_strategy = JsonCssExtractionStrategy({
    "name": "36Kr Articles",
    "baseSelector": "div.article-item",
    "fields": [
        {"name": "title", "selector": "a.article-item-title", "type": "text"},
        {"name": "url", "selector": "a.article-item-title", "type": "attribute", "attribute": "href"},
        {"name": "summary", "selector": "a.article-item-description", "type": "text"},
        {"name": "author", "selector": "span.article-item-author", "type": "text"},
        {"name": "date", "selector": "span.article-item-time", "type": "text"}
    ]
})
def normalize_url(url: str) -> str:
    """移除查询参数和锚点,统一为小写"""
    from urllib.parse import urlparse, urlunparse
    parsed = urlparse(url)
    return urlunparse((parsed.scheme, parsed.netloc, parsed.path, '', '', '')).lower()
def content_hash(article: dict) -> str:
    """基于标题+摘要生成哈希"""
    content = f"{article.get('title','')}{article.get('summary','')}"
    return hashlib.md5(content.encode('utf-8')).hexdigest()
async def scrape_36kr():
    async with AsyncWebCrawler(
        headless=True,
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        delay_before_return_html=2
    ) as crawler:
            url="https://36kr.com/information/technology/",
            extraction_strategy=extraction_strategy,
            bypass_cache=True
        )
        if not result.extracted_content:
            print("未提取到内容")
            return
loads(result.extracted_content)
        print(f"原始抓取: {len(articles)} 篇文章")
        seen_urls = set()
        seen_hashes = set()
        unique_articles = []
        for article in articles:
            url = article.get('url', '')
            if url.startswith('/'):
                article['url'] = f"https://36kr.com{url}"
            norm_url = normalize_url(article['url'])
            if norm_url in seen_urls:
                continue
            seen_urls.add(norm_url)
            chash = content_hash(article)
            if chash in seen_hashes:
                continue
            seen_hashes.add(chash)
            article['title'] = article.get('title', '').strip()
            article['summary'] = article.get('summary', '').strip()
            article['author'] = article.get('author', '36氪').strip()
            article['date'] = article.get('date', '').strip()
            unique_articles.append(article)
        print(f"去重后: {len(unique_articles)} 篇文章")
json', 'w', encoding='utf-8') as f:
            json.dump(unique_articles, f, ensure_ascii=False, indent=2)
        print("JSON导出: articles.json")
        with open('articles_summary.md', 'w', encoding='utf-8') as f:
            f.write(f"# 36氪科技文章摘要\n\n")
            f.write(f"**抓取时间**: {asyncio.get_event_loop().time()}\n")
            f.write(f"**文章数量**: {len(unique_articles)} 篇\n\n")
            f.write("---\n\n")
            for i, article in enumerate(unique_articles, 1):
                f.write(f"## {i}. {article['title']}\n\n")
                f.write(f"- **作者**: {article['author']}\n")
                f.write(f"- **时间**: {article['date']}\n")
                f.write(f"- **链接**: {article['url']}\n")
                f.write(f"- **摘要**: {article['summary'][:100]}...\n\n")
> 注: 本SKILL.md超过500行上限, 已截断尾部非核心章节以满足L1格式要求。完整内容见版本库历史。
## 问答集锦
### Q1: 网页抓取引擎支持哪些输入格式？
A1: Firecrawl/P。支持文本指令和结构化参数输入，具体格式参考使用流程章节。
### Q2: 需要配置API Key吗？
A2: 是的，部分功能需要配置对应平台的API Key。请在依赖说明章节查看具体要求，并通过环境变量安全配置。
### Q3: 命令行执行失败怎么办？
A3: 检查命令参数是否正确，确认运行环境支持exec能力。如遇权限问题，请参照错误处理章节排查。
## 安全指引
### 安全风险防范
| 风险项 | 等级 | 防护措施 | 验证方法 |
| --- | --- | --- | --- |
| API密钥泄露 | 高 | 通过环境变量配置，禁止硬编码 | 定期检查代码和配置文件 |
| 命令执行风险 | 高 | 仅执行白名单命令，避免拼接用户输入 | 使用沙箱环境测试 |
| 网络通信安全 | 中 | 使用HTTPS协议，验证SSL证书 | 定期检查证书有效期 |
| 敏感数据暴露 | 高 | 输出结果中不包含密钥、令牌等敏感信息 | 日志脱敏审查 |
| 未授权访问 | 中 | 限制访问权限，实施认证机制 | 定期审计访问日志 |
使用前请确认已阅读依赖说明章节，确保运行环境满足安全要求。
### 网页抓取引擎通用排查步骤
1. **检查输入参数**: 确认所有必填参数已提供且格式正确
2. **查看日志输出**: 定位具体错误行和异常类型
3. **验证环境配置**: 确认依赖库版本和运行环境满足要求
4. **逐步调试**: 缩小问题范围,隔离故障模块
## 异常处置
针对网页抓取引擎使用中可能遇到的常见问题,提供以下排查方案:
| 错误类型 | 原因分析 | 解决方案 |
|---------|---------|---------|
| API认证失败(401) | API密钥错误或过期 | 检查密钥配置,重新生成token |
| 接口限流(429) | 请求频率超出限制 | 降低调用频率,启用重试退避策略 |
| 响应超时(504) | 网络延迟或服务端负载过高 | 增加超时阈值,检查网络连接 |
| 文件不存在 | 路径错误或文件未创建 | 检查路径拼写,确认文件已生成 |
| 文件格式不支持 | 扩展名不在支持列表中 | 转换为支持的格式后重试 |
| 权限不足 | 当前用户无读写权限 | 检查文件权限,以管理员身份运行 |
| 命令执行失败 | 参数错误或环境依赖缺失 | 检查命令语法,确认依赖已安装 |
| 进程超时 | 命令执行时间过长 | 增加超时设置,优化命令参数 |
| 网络连接失败 | DNS解析失败或防火墙拦截 | 检查网络配置,确认代理设置 |
## 帮助文档