scrapfly-crawler

Scrapfly Crawler

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

适用场景

• 爬取整个网站或特定版块
• 发现网站上的所有页面／URL
• 批量收集多个页面的内容
• 在整个网站范围内大规模提取结构化数据
• 下载网站存档（WARC/HAR 格式）
• 根据爬取的文档构建 LLM.txt 文件

设置

pip install scrapfly-sdk

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

SDK 导入

from scrapfly import ScrapflyClient, CrawlerConfig, Crawl

Webhook 的其他导入项：

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

CrawlerConfig 参数

参数	类型	默认值	描述
url	str	必填	爬取的起始 URL
page_limit	int	None	爬取的最大页面数
max_depth	int	None	从起始 URL 开始遍历链接的最大深度
max_duration	int	None	爬取的最长时长，单位为秒
exclude_paths	list[str]	None	要跳过的 URL 路径模式（支持 *，最多 100 个）。不能与 include_only_paths 同时使用
include_only_paths	list[str]	None	仅爬取与这些路径模式匹配的 URL（支持 *，最多 100 个）。不能与 exclude_paths 同时使用
ignore_base_path_restriction	bool	False	允许爬取同一域名下的任意路径，而不仅限于种子 URL 的基础路径
follow_external_links	bool	False	跟踪指向外部域名的链接
allowed_external_domains	list[str]	None	当 follow_external_links=True 时，外部域名的允许列表（支持 *，最多 250 个）
headers	dict	None	以 JSON 对象形式提供自定义 HTTP 标头
delay	int	None	请求之间的延迟，单位为毫秒（0-15000）
user_agent	str	None	自定义 User-Agent 字符串。在 proxy_pool="unblocker"、"unblocker_ultra" 或 "auto" 时，Unblocker 会保持启用，并替换 Chrome User-Agent；其他值则按原样发送
max_concurrency	int	None	并发爬取请求的最大数量
rendering_delay	int	None	页面加载后、提取前的等待时间，单位为毫秒（0-25000）。0 会禁用浏览器渲染
use_sitemaps	bool	False	如果存在 sitemap.xml，则从中发现 URL
respect_robots_txt	bool	False	遵循 robots.txt 规则和 Disallow 指令
ignore_no_follow	bool	False	忽略链接上的 rel="nofollow" 属性
cache	bool	False	为爬取的页面启用缓存层
cache_ttl	int	None	缓存生存时间，单位为秒（0-604800）
cache_clear	bool	False	强制刷新缓存页面
content_formats	list[str]	None	要提取的内容格式：html、clean_html、markdown、text
extraction_rules	dict	None	用于结构化数据的自定义提取规则
unblocker	bool	False	已弃用：请使用 proxy_pool（Unblocker 对应 "unblocker_ultra"）。迁移指南介绍了代理池推出前编写的请求。asp 是它的别名；两者都提供时，以 asp 为准
proxy_pool	str	None	每个爬取页面所使用的代理池："basic"（数据中心代理，Unblocker 关闭）、"unblocker"（住宅代理和 Unblocker）、"unblocker_ultra"（同上，也是目标网站或具有独立价格的反机器人保护所接受的唯一代理池），或 "auto"（API 在每次请求前根据目标网站的配置选择代理池：在数据中心代理上启用 Unblocker，或使用目标网站所需的代理池；所有目标网站均接受此选项，包括要求使用 unblocker_ultra 的目标网站，其价格会另行计入）。每种代理池都会自行设置 Unblocker。未设置时，SDK 不会发送 proxy_pool。迁移指南介绍了代理池推出前编写的请求，也包括已弃用的代理池名称 public_datacenter_pool 和 public_residential_pool
country	str	None	代理国家/地区选择（ISO 代码）
webhook_name	str	None	在控制面板中配置的 Webhook 名称
webhook_events	list[str]	None	要订阅的事件（例如：crawler_started、crawler_url_visited、crawler_finished）
max_api_credit	int	None	此次爬取可使用的最大 API 积分

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

# Create, start, and wait for crawl in one chain
# 2. Create and run crawler
crawl = Crawl(
    client,
    CrawlerConfig(url='https://web-scraping.dev/products', page_limit=5)
).crawl().wait()

# 3. Get results
pages = crawl.warc().get_pages()

# 4. Process results
print(f"Crawled {len(pages)} pages:")
for page in pages:
    print(f"  • {page['url']} ({page['status_code']})")

# 5. Access specific URLs
html = crawl.read('https://web-scraping.dev/products')
if html:
    print(f"\nMain page is {len(html):,} bytes")

底层工作流：启动、轮询、下载

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 crawler
start_response = client.start_crawl(config)
print(f"Crawler started: {start_response.uuid}")

# 2. Poll for status
while True:
    status = client.get_crawl_status(start_response.uuid)
    print(f"Status: {status.status} - {status.progress_pct:.1f}%")
    print(f"Crawled: {status.urls_crawled}/{status.urls_discovered} pages")

    if status.is_complete:
        break
    elif status.is_failed:
        print("Crawl failed!")
        break

    time.sleep(5)

# 3. Download and process results
if status.is_complete:
    artifact = client.get_crawl_artifact(start_response.uuid)
    pages = artifact.get_pages()
    print(f"Downloaded {len(pages)} pages")

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

    # Save to file
    artifact.save('crawl_results.warc.gz')

按路径筛选进行抓取

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,
    proxy_pool='unblocker',  # residential proxies and the Unblocker on every page
    content_formats=['html'],
)
crawl = Crawl(client, config).crawl().wait()

如果目标站点或反机器人防护有单独的价格，basic 和 unblocker 会对每个页面返回 ERR::PROXY::UNBLOCKER_ULTRA_REQUIRED 错误（HTTP 422，不计费，不重试；若种子 URL 被拒绝，抓取将以 seed_url_failed 结束）：请使用 proxy_pool='unblocker_ultra' 重新运行抓取。

使用提取规则抓取

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 的内容

# Read HTML (from WARC - fast)
content = crawl.read('https://web-scraping.dev/product/1')
if content:
    print(f"URL: {content.url}")
    print(f"Status: {content.status_code}")
    print(f"Duration: {content.duration}s")
    print(content.content)

# Read markdown (from contents API)
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)} chars")

批量获取内容

# Get markdown for multiple URLs in a single API call (max 100 per request)
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)} chars")

下载 WARC/HAR 存档

# WARC format (default)
artifact = crawl.warc()
artifact.save('crawl_results.warc.gz')

# HAR format
artifact = crawl.har()
artifact.save('crawl_results.har')

# Iterate through records
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/*'], # only allow doc resources
    page_limit=50,
    max_depth=5,
    follow_external_links=False,
    content_formats=['markdown'],
)

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

# Get URLs from WARC
warc_artifact = crawl.warc()
all_urls = [
    record.url for record in warc_artifact.iter_responses()
    if record.status_code == 200
]

# Retrieve markdown content in batches
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)

# Build llms-full.txt
lines = ["# Documentation\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"URLs discovered: {stats['urls_discovered']}")
print(f"URLs crawled: {stats['urls_crawled']}")
print(f"Progress: {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 控制大型网站的积分消耗
• 每个页面按其代理池的价格计费：basic 为 1 个积分，Unblocker 池为 25 个积分，使用浏览器渲染时额外加 5 个积分；unblocker_ultra 会加上目标站点或防护自身的价格（若有）；auto 按其实际运行所用代理池的价格计费
• 使用 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，以免抓取整个网络