scrapfly-crawler

Scrapfly クローラー

Python SDK を使用して Scrapfly Crawler 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 のパラメーター

パラメーター	型	デフォルト	説明
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() は 1 リクエストあたり最大 100 件の URL に対応しており、コンテンツを効率よく一括取得できます
• 本番環境では、セキュリティのため Webhook の署名を検証してください
• Web 全体のクロールを避けるため、follow_external_links は慎重に使用してください