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"

执行 vs 指导

Fetcher 选择指南

┌─────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│   Fetcher       │────▶│ DynamicFetcher   │────▶│ StealthyFetcher  │
│   (HTTP)        │     │ (Browser/JS)     │     │ (Anti-bot)       │
└─────────────────┘     └──────────────────┘     └──────────────────┘
     最快                 渲染 JS 内容              Cloudflare、
     静态页面             SPA、React/Vue           Turnstile 等

决策树

1. 是否为静态 HTML？ → 使用 Fetcher（速度快 10–100 倍）
2. 是否需执行 JavaScript？ → 使用 DynamicFetcher
3. 是否被封禁？ → 使用 StealthyFetcher
4. 是否涉及复杂会话？ → 使用 Session 变体（如 FetcherSession、StealthySession）

MCP 获取模式

• fetch_page — HTTP Fetcher
• fetch_dynamic — 基于浏览器（Playwright）
• fetch_stealthy — 反爬绕过模式

反爬升级阶梯

第 1 级：礼貌型 HTTP 请求

# MCP 调用：fetch_page 并传入参数
{
  "url": "https://example.com",
  "headers": {"User-Agent": "..."},
  "delay": 2.0
}

第 2 级：会话持久化

# 使用会话管理跨请求的 Cookie / 状态
FetcherSession(impersonate="chrome")  # TLS 指纹模拟

第 3 级：隐身模式

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

第 4 级：代理轮换

详见 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

Spider 框架（大规模爬取）

何时使用 Spider 而非直接调用 Fetcher？

• ✅ Spider：目标页数 ≥ 10，需并发、断点续爬、代理轮换等能力
• ✅ 直接调用：仅需抓取 1–5 页，流程简单、提取快速

基础 Spider 模式

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")

进阶：多会话 Spider

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")

Spider 功能特性

• 暂停/续爬：通过 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 方法

# 按属性查找
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('Add to Cart', tag='button')
page.find_by_regex(r'\$\d+\.\d{2}')

# DOM 导航
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">(.*?)',
        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 Fetcher —— 比浏览器快 10–100 倍
2. 模拟浏览器指纹 —— 使用 impersonate='chrome' 实现 TLS 指纹欺骗
3. 支持 HTTP/3 —— FetcherSession(http3=True)
4. 限制资源加载 —— 在 Dynamic/Stealthy 模式中启用 disable_resources=True
5. 连接池复用 —— 在多个请求间复用会话实例

安全守则（始终遵守）

• 仅爬取您被授权访问的内容
• 尊重 robots.txt 及网站服务条款（ToS）
• 大规模爬取时务必设置请求间隔（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 —— 连通性与反爬指标测试脚本