Scrapfly 爬虫

通过 Python SDK 使用 Scrapfly 爬虫 API，可调度并管理全站爬取任务，支持自动链接发现、深度控制和结构化数据提取。

适用场景

• 爬取整个网站或特定部分
• 发现站点中的所有页面/URL
• 批量收集多个页面的内容
• 在整个站点上大规模提取结构化数据
• 下载网站存档（WARC/HAR 格式）
• 从爬取的文档中生成 LLM.txt 文件

安装设置

pip install scrapfly-sdk

API 密钥必须通过环境变量 SCRAPFLY_API_KEY 提供，或直接传递给客户端。

SDK 导入

from scrapfly import ScrapflyClient, CrawlerConfig, Crawl

用于 Webhook 的额外导入：

from scrapfly import (
    webhook_from_payload,
    CrawlStartedWebhook,
    CrawlUrlDiscoveredWebhook,
    CrawlUrlFailedWebhook,
    CrawlCompletedWebhook,
)

CrawlerConfig 参数

Webhook 事件

• "crawler_started" - 爬虫任务开始执行
• "crawler_url_visited" - 成功爬取了一个 URL
• "crawler_url_skipped" - URL 被跳过
• "crawler_url_discovered" - 发现了新的 URL
• "crawler_url_failed" - 某个 URL 爬取失败
• "crawler_stopped" - 因错误或限制而停止爬取
• "crawler_cancelled" - 爬取被手动取消
• "crawler_finished" - 爬取任务成功完成

示例

快速入门：爬取并获取结果

import os
from scrapfly import ScrapflyClient, CrawlerConfig, Crawl

client = ScrapflyClient(key=os.environ["SCRAPFLY_API_KEY"])

# 创建、启动并等待爬取完成（链式调用）
# 2. 创建并运行爬虫
crawl = Crawl(
    client,
    CrawlerConfig(url='https://web-scraping.dev/products', page_limit=5)
).crawl().wait()

# 3. 获取结果
pages = crawl.warc().get_pages()

# 4. 处理结果
print(f"已爬取 {len(pages)} 个页面：")
for page in pages:
    print(f"  • {page['url']} ({page['status_code']})")

# 5. 访问特定 URL 内容
html = crawl.read('https://web-scraping.dev/products')
if html:
    print(f"\n主页大小为 {len(html):,} 字节")

低级流程：启动、轮询、下载

import os
import time
from scrapfly import ScrapflyClient, CrawlerConfig

client = ScrapflyClient(key=os.environ["SCRAPFLY_API_KEY"])

config = CrawlerConfig(
    url='https://web-scraping.dev',
    page_limit=10,
    max_depth=2,
    content_formats=['html', 'markdown']
)

# 1. 启动爬虫
start_response = client.start_crawl(config)
print(f"爬虫已启动：{start_response.uuid}")

# 2. 轮询状态
while True:
    status = client.get_crawl_status(start_response.uuid)
    print(f"状态：{status.status} - {status.progress_pct:.1f}%")
    print(f"已爬取：{status.urls_crawled}/{status.urls_discovered} 个页面")

    if status.is_complete:
        break
    elif status.is_failed:
        print("爬取失败！")
        break

    time.sleep(5)

# 3. 下载并处理结果
if status.is_complete:
    artifact = client.get_crawl_artifact(start_response.uuid)
    pages = artifact.get_pages()
    print(f"已下载 {len(pages)} 个页面")

    for page in pages:
        print(f"  {page['url']}: {page['status_code']} ({len(page['content'])} 字节)")

    # 保存到文件
    artifact.save('crawl_results.warc.gz')

带路径过滤的爬取

# Scrapfly 爬虫技能 (Scrapfly Crawler)

## 基础爬取

config = CrawlerConfig(

url='https://web-scraping.dev/',

page_limit=50,

max_depth=5,

include_only_paths=['*/product/*'],

content_formats=['markdown', 'text'],

)

crawl = Crawl(client, config).crawl().wait()

## 绕过反爬机制的爬取

config = CrawlerConfig(

url='https://web-scraping.dev',

page_limit=50,

asp=True,

content_formats=['html'],

)

crawl = Crawl(client, config).crawl().wait()

## 使用提取规则的爬取

config = CrawlerConfig(

url='https://web-scraping.dev/products',

page_limit=100,

max_depth=3,

include_only_paths=['/products/*'],

extraction_rules={

"/products/*": {

"type": "model",

"value": "product",

},

},

)

crawl = Crawl(client, config).crawl().wait()

## 读取特定 URL 的内容

从 WARC 读取 HTML（快速）

content = crawl.read('https://web-scraping.dev/product/1')

if content:

print(f"URL: {content.url}")

print(f"状态码: {content.status_code}")

print(f"耗时: {content.duration}s")

print(content.content)

从内容 API 读取 Markdown

content = crawl.read('https://web-scraping.dev/product/1', format='markdown')

if content:

print(content.content)

## 遍历匹配模式的 URL

for content in crawl.read_iter(pattern="*/products?page=*", format="markdown"):

print(f"{content.url}: {len(content.content)} 字符")

## 批量内容获取

单次 API 请求中获取多个 URL 的 Markdown 内容（最多 100 个）

urls = ['https://web-scraping.dev/product/1', 'https://web-scraping.dev/product/2']

contents = crawl.read_batch(urls, formats=['markdown'])

for url, formats_dict in contents.items():

markdown = formats_dict.get('markdown', '')

print(f"{url}: {len(markdown)} 字符")

## 下载 WARC/HAR 归档文件

WARC 格式（默认）

artifact = crawl.warc()

artifact.save('crawl_results.warc.gz')

HAR 格式

artifact = crawl.har()

artifact.save('crawl_results.har')

遍历响应记录

for record in artifact.iter_responses():

print(f"{record.url}: {record.status_code}")

## 从爬取的文档构建 LLM.txt 文件

import os

from scrapfly import ScrapflyClient, CrawlerConfig, Crawl

client = ScrapflyClient(key=os.environ["SCRAPFLY_API_KEY"])

config = CrawlerConfig(

url='https://web-scraping.dev/',

include_only_paths=['/docs', '/docs/*'], # 仅允许文档资源

page_limit=50,

max_depth=5,

follow_external_links=False,

content_formats=['markdown'],

)

crawl = Crawl(client, config).crawl().wait(poll_interval=5, verbose=True)

从 WARC 获取所有 URL

warc_artifact = crawl.warc()

all_urls = [

record.url for record in warc_artifact.iter_responses()

if record.status_code == 200

]

分批获取 Markdown 内容

all_contents = {}

for i in range(0, len(all_urls), 100):

batch_urls = all_urls[i:i + 100]

batch_contents = crawl.read_batch(batch_urls, formats=['markdown'])

all_contents.update(batch_contents)

构建 llms-full.txt

lines = ["# 文档\n"]

for url, formats_dict in all_contents.items():

markdown = formats_dict.get('markdown', '').strip()

if markdown:

lines.append(f"---\n\n### {url}\n\n{markdown}\n")

with open("llms-full.txt", "w", encoding="utf-8") as f:

f.write("\n".join(lines))

## 爬取统计信息

stats = crawl.stats()

print(f"发现的 URL 数量: {stats['urls_discovered']}")

print(f"已爬取的 URL 数量: {stats['urls_crawled']}")

print(f"进度: {stats['progress_pct']:.1f}%")

## 主要 SDK 类

| 类 | 描述 |
|-------|-------------|
| `ScrapflyClient` | 主客户端，负责认证和 API 调用 |
| `CrawlerConfig` | 爬取任务配置 |
| `Crawl` | 高级爬取生命周期管理器（启动、等待、读取） |
| `CrawlerStartResponse` | `client.start_crawl()` 的返回结果 |
| `CrawlerStatusResponse` | `client.get_crawl_status()` 的返回结果，包含 `is_complete`、`is_failed`、`progress_pct` |
| `CrawlerArtifactResponse` | `client.get_crawl_artifact()` 的返回结果，包含 `get_pages()`、`iter_responses()`、`save()` 方法 |
| `CrawlContent` | 单个 URL 的内容，包含 `url`、`content`、`status_code`、`duration` |

## 重要注意事项

- 使用 `page_limit` 控制大型网站的费用支出
- 使用 `include_only_paths` 和 `exclude_paths` 限定爬取范围，聚焦于相关部分
- `exclude_paths` 与 `include_only_paths` 互斥，不可同时使用
- 爬取任务为异步执行，简单场景使用 `Crawl.crawl().wait()`，需要更精细控制时可轮询 `client.get_crawl_status()`
- WARC/HAR 归档文件可通过 `crawl.warc()` / `crawl.har()` 获取，用于离线处理
- `read_batch()` 支持单次请求最多 100 个 URL，实现高效批量内容获取
- 生产环境中应验证 Webhook 签名以确保安全
- 使用 `follow_external_links` 时需谨慎，避免意外爬取整个互联网