Scrapling MCP — 网页抓取指导

引导层 + MCP 集成

本技能用于策略与模式设计。执行阶段请通过 mcporter 调用 Scrapling 的 MCP 服务。

快速开始（MCP）

1. 安装支持 MCP 的 Scrapling

pip install scrapling[mcp]
# 或安装完整功能：
pip install scrapling[mcp,playwright]
python -m playwright install chromium

2. 添加到 OpenClaw MCP 配置

{
  "mcpServers": {
    "scrapling": {
      "command": "python",
      "args": ["-m", "scrapling.mcp"]
    }
  }
}

3. 通过 mcporter 调用

mcporter call scrapling fetch_page --url "https://example.com"

执行与引导的区别

抓取器选择指南

┌─────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│   抓取器        │────▶│ 动态抓取器       │────▶│ 隐蔽抓取器       │
│   (HTTP)        │     │ (浏览器/JS 渲染)  │     │ (反反爬)         │
└─────────────────┘     └──────────────────┘     └──────────────────┘
     最快                JS 渲染页面           Cloudflare、Turnstile 等
     静态页面            SPA、React/Vue         常见反爬挑战

决策树

1. 页面是静态 HTML 吗？ → 使用 Fetcher（速度提升 10–100 倍）
2. 需要执行 JavaScript 吗？ → 使用 DynamicFetcher
3. 频繁被封禁？ → 使用 StealthyFetcher
4. 涉及复杂会话？ → 使用会话变体

MCP 抓取模式

• fetch_page — HTTP 抓取器
• fetch_dynamic — 基于 Playwright 的浏览器抓取
• fetch_stealthy — 反反爬绕过模式

反爬升级阶梯

第一级：友好 HTTP 请求

# MCP 调用：fetch_page 并配置选项
{
  "url": "https://example.com",
  "headers": {"User-Agent": "..."},
  "delay": 2.0
}

第二级：会话持久化

# 使用会话保持 Cookie / 状态
FetcherSession(impersonate="chrome")  # 模拟浏览器 TLS 指纹

第三级：隐蔽模式

# MCP 调用：fetch_stealthy
StealthyFetcher.fetch(
    url,
    headless=True,
    solve_cloudflare=True,  # 自动解决 Turnstile
    network_idle=True
)

第四级：代理轮换

详见 references/proxy-rotation.md

自适应抓取（抗脆弱性）

Scrapling 可通过自适应选择器应对网站结构变更：

# 首次运行 —— 保存 DOM 特征指纹
products = page.css('.product', auto_save=True)

# 后续运行 —— 若 DOM 结构变化，自动重新定位
products = page.css('.product', adaptive=True)

MCP 使用方式：

mcporter call scrapling css_select \\
  --selector ".product" \\
  --adaptive true \\
  --auto-save true

蜘蛛框架（大规模爬取）

何时使用蜘蛛 vs 直接抓取：

• ✅ 使用蜘蛛：需处理 10+ 页面、需要并发、支持断点续爬、代理轮换
• ✅ 直接抓取：仅 1–5 个页面、快速提取、流程简单

基础蜘蛛模式

from scrapling.spiders import Spider, Response

class ProductSpider(Spider):
    name = "products"
    start_urls = ["https://example.com/products"]
    concurrent_requests = 10
    download_delay = 1.0

    async def parse(self, response: Response):
        for product in response.css('.product'):
            yield {
                "name": product.css('h2::text').get(),
                "price": product.css('.price::text').get(),
                "url": response.url
            }

        # 处理分页
        next_page = response.css('.next a::attr(href)').get()
        if next_page:
            yield response.follow(next_page)

# 启动并支持断点续爬
result = ProductSpider(crawldir="./crawl_data").start()
result.items.to_jsonl("products.jsonl")

高级：多会话蜘蛛

from scrapling.spiders import Spider, Request, Response
from scrapling.fetchers import FetcherSession, AsyncStealthySession

class MultiSessionSpider(Spider):
    name = "multi"
    start_urls = ["https://example.com/"]

    def configure_sessions(self, manager):
        manager.add("fast", FetcherSession(impersonate="chrome"))
        manager.add("stealth", AsyncStealthySession(headless=True), lazy=True)

    async def parse(self, response: Response):
        for link in response.css('a::attr(href)').getall():
            if "/protected/" in link:
                yield Request(link, sid="stealth")
            else:
                yield Request(link, sid="fast")

蜘蛛功能特性

• 暂停/恢复：通过 crawldir 参数保存检查点
• 流式处理：支持 async for item in spider.stream() 实时处理数据
• 自动重试：可配置在请求被屏蔽时自动重试
• 导出功能：内置 to_json()、to_jsonl() 导出方法

CLI 与交互式 Shell

终端数据提取（无需编写代码）

# 提取为 Markdown
scrapling extract get 'https://example.com' content.md

# 提取特定元素
scrapling extract get 'https://example.com' content.txt \\
  --css-selector '.article' \\
  --impersonate 'chrome'

# 隐蔽模式
scrapling extract stealthy-fetch 'https://protected.com' content.md \\
  --no-headless \\
  --solve-cloudflare

交互式 Shell

scrapling shell

# 在 Shell 中操作：
>>> page = Fetcher.get('https://example.com')
>>> page.css('h1::text').get()
>>> page.find_all('div', class_='item')

解析器 API（超越 CSS/XPath）

BeautifulSoup-Style 方法

# 通过属性查找
page.find_all('div', {'class': 'product', 'data-id': True})
page.find_all('div', class_='product', id=re.compile(r'item-\\d+'))

# 文本搜索
page.find_by_text('添加到购物车', tag='button')
page.find_by_regex(r'\\$\\d+\\.\\d{2}')

# 导航
first = page.css('.product')[0]
parent = first.parent
siblings = first.next_siblings
children = first.children

# 相似性匹配
similar = first.find_similar()  # 查找视觉或结构上相似的元素
below = first.below_elements()  # DOM 中位于其下方的元素

自动生成的选择器

# 获取任意元素的稳定选择器
element = page.css('.product')[0]
selector = element.auto_css_selector()  # 返回稳定的 CSS 路径
xpath = element.auto_xpath()

代理轮换

from scrapling.spiders import ProxyRotator

# 循环轮换
rotator = ProxyRotator([
    "http://proxy1:8080",
    "http://proxy2:8080",
    "http://user:pass@proxy3:8080"
], strategy="cyclic")

# 与任意会话结合使用
with FetcherSession(proxy=rotator.next()) as session:
    page = session.get('https://example.com')

常用技巧

分页模式

# 页码分页
for page_num in range(1, 11):
    url = f"https://example.com/products?page={page_num}"
    ...

# 下一页按钮
while next_page := response.css('.next a::attr(href)').get():
    yield response.follow(next_page)

# 无限滚动（DynamicFetcher）
with DynamicSession() as session:
    page = session.fetch(url)
    page.scroll_to_bottom()
    items = page.css('.item').getall()

登录会话

with StealthySession(headless=False) as session:
    # 登录
    login_page = session.fetch('https://example.com/login')
    login_page.fill('input[name="username"]', 'user')
    login_page.fill('input[name="password"]', 'pass')
    login_page.click('button[type="submit"]')

    # 现在会话已携带 Cookie
    protected_page = session.fetch('https://example.com/dashboard')

Next.js 数据提取

# 提取 __NEXT_DATA__ 中的 JSON
import json
import re

next_data = json.loads(
    re.search(
        r'__NEXT_DATA__" type="application/json">(.*?)</script>',
        page.html_content,
        re.S
    ).group(1)
)
props = next_data['props']['pageProps']

输出格式

# JSON（美化输出）
result.items.to_json('output.json')

# JSONL（流式输出，每行一个）
result.items.to_jsonl('output.jsonl')

# Python 对象
for item in result.items:
    print(item['title'])

性能优化建议

1. 尽可能使用 HTTP 获取器 — 比浏览器快 10–100 倍
2. 模拟浏览器行为 — 使用 impersonate='chrome' 实现 TLS 指纹伪装
3. 支持 HTTP/3 — 在 FetcherSession 中启用 http3=True
4. 限制资源加载 — 在 Dynamic/Stealthy 模式中设置 disable_resources=True
5. 连接池复用 — 在多个请求间复用会话以提升效率

安全守则（始终遵守）

• 仅爬取您有权访问的内容
• 尊重 robots.txt 和服务条款
• 大规模爬取时添加延迟（download_delay）
• 未经许可不得绕过付费墙或身份验证
• 严禁爬取个人或敏感数据

参考文档

• references/mcp-setup.md — 详细的 MCP 配置说明
• references/anti-bot.md — 反爬虫策略处理方法
• references/proxy-rotation.md — 代理配置与轮换方案
• references/spider-recipes.md — 高级爬取模式示例
• references/api-reference.md — 快速 API 参考手册
• references/links.md — 官方文档链接汇总

工具脚本

• scripts/scrapling_scrape.py — 快速一次性数据提取脚本
• scripts/scrapling_smoke_test.py — 测试连接性及反爬指标