scrapfly-browser

Scrapfly Cloud Browser

Scrapfly Cloud Browser API を Python Playwright とともに使用して、プロキシの自動ローテーション、アンチボットフィンガープリント、地域ターゲティングを組み込んだリモートクラウドブラウザーを自動化します。

使用する場面

• リモートクラウドブラウザーでブラウザー操作を自動化する場合（ログインフロー、フォーム入力、ナビゲーション）
• 完全なブラウザー自動化が必要な、JavaScript を多用するサイトをスクレイピングする場合
• 管理されたブラウザーフィンガープリントでアンチボット保護を回避する場合
• 地域ターゲティングされたプロキシ経由で Playwright スクリプトを実行する場合
• 複数の接続にわたってブラウザーの永続セッションを維持する場合
• ブラウザー操作を通じてファイルをダウンロードする場合

セットアップ

pip install playwright

API キーは、環境変数 SCRAPFLY_API_KEY を介して指定するか、接続 URL に直接渡す必要があります。

API リファレンス

接続： WebSocket CDP（Chrome DevTools Protocol）

wss://browser.scrapfly.io?api_key=YOUR_API_KEY

接続パラメーター

パラメーター	型	デフォルト	説明
api_key	str	必須	Scrapfly API キー
proxy_pool	str	"datacenter"	プロキシの種類："datacenter" または "residential"。Web Scraping API の basic、unblocker、unblocker_ultra、auto は Cloud Browser には適用されず、指定すると ERR::BROWSER::CONFIG_ERROR で拒否されます
os	str	ランダム	OS フィンガープリント："linux"、"windows"、または "macos"
country	str	None	プロキシの国（ISO 3166-1 alpha-2、例："us"、"de"）
session	str	None	ブラウザー状態を保持するセッション ID
auto_close	str	"true"	切断時にセッションを閉じます。維持するには "false" を設定します
timeout	int	900	セッションの最大継続時間（秒、900-1800）
debug	str	"false"	セッションの動画録画を有効にします
downloads_stream	str	None	"1" を指定すると、大きなダウンロードをインラインの base64 ではなく IO ストリームハンドルとして返します

接続 URL の作成

import os
from urllib.parse import urlencode

params = {
    "api_key": os.environ["SCRAPFLY_API_KEY"],
    "proxy_pool": "datacenter",
    "os": "linux",
}
BROWSER_WS = f"wss://browser.scrapfly.io?{urlencode(params)}"

例

基本的なページナビゲーション

from playwright.sync_api import sync_playwright
import os

API_KEY = os.environ["SCRAPFLY_API_KEY"]
BROWSER_WS = f"wss://browser.scrapfly.io?api_key={API_KEY}&proxy_pool=datacenter&os=linux"

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto("https://web-scraping.dev")
        print("Title:", page.title())
        print("Content:", page.content()[:500])
    finally:
        browser.close()

非同期 Playwright

from playwright.async_api import async_playwright
import asyncio
import os

API_KEY = os.environ["SCRAPFLY_API_KEY"]
BROWSER_WS = f"wss://browser.scrapfly.io?api_key={API_KEY}&proxy_pool=datacenter&os=linux"

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp(BROWSER_WS)
        try:
            context = browser.contexts[0]
            page = context.pages[0] if context.pages else await context.new_page()

            await page.goto("https://web-scraping.dev")
            print("Title:", await page.title())
        finally:
            await browser.close()

asyncio.run(main())

フォーム入力を含むログインフロー

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto("https://web-scraping.dev/login")
        page.fill("input[name=username]", "user123")
        page.fill("input[name=password]", "password")
        page.click("button[type=submit]")
        page.wait_for_selector("div#secret-message")

        print("Logged in:", page.url)
    finally:
        browser.close()

スクリーンショットを撮る

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto("https://web-scraping.dev")
        page.screenshot(path="screenshot.png", full_page=True)
    finally:
        browser.close()

セレクターを使ってデータをスクレイピングする

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto("https://web-scraping.dev/products")
        page.wait_for_selector("div.product")

        products = page.query_selector_all("div.product")
        for product in products:
            name = product.query_selector("a").inner_text()
            price = product.query_selector(".price").inner_text()
            print(f"{name}: {price}")
    finally:
        browser.close()

レジデンシャルプロキシを使った地域ターゲティング閲覧

BROWSER_WS = (
    f"wss://browser.scrapfly.io?"
    f"api_key={API_KEY}"
    f"&proxy_pool=residential"
    f"&country=de"
    f"&os=windows"
)

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto("https://web-scraping.dev")
        print("Page content from DE proxy:", page.title())
    finally:
        browser.close()

永続セッション（複数ステップのワークフロー）

フォーム入力を含むログインフローの例と同じログイン認証情報およびセレクターを使用します。

LOGIN_URL = "https://web-scraping.dev/login"
USERNAME = "user123"
PASSWORD = "password"

# Step 1: Login and keep session alive
BROWSER_WS = (
    f"wss://browser.scrapfly.io?"
    f"api_key={API_KEY}"
    f"&session=my-session"
    f"&auto_close=false"
    f"&proxy_pool=datacenter"
)

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto(LOGIN_URL)
        page.fill("input[name=username]", USERNAME)
        page.fill("input[name=password]", PASSWORD)
        page.click("button[type=submit]")
        page.wait_for_selector("div#secret-message")
        print("Login complete, session preserved")
    finally:
        browser.close()  # Disconnects but session stays alive

# Step 2: Reconnect to same session (still authenticated with same credentials)
BROWSER_WS = (
    f"wss://browser.scrapfly.io?"
    f"api_key={API_KEY}"
    f"&session=my-session"
    f"&proxy_pool=datacenter"
)

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto(LOGIN_URL)
        page.wait_for_selector("div#secret-message")
        print("Accessed protected page (still logged in):", page.title())
    finally:
        browser.close()

無限スクロールのスクレイピング

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto("https://web-scraping.dev/testimonials")
        page.wait_for_selector("span.rating")

        all_items = []
        for _ in range(10):  # Scroll 10 times
            items = page.query_selector_all("span.rating")
            all_items = items
            page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            page.wait_for_timeout(2000)

        print(f"Collected {len(all_items)} reviews")
    finally:
        browser.close()

ページ内で JavaScript を実行する

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        page.goto("https://web-scraping.dev")

        # Extract data using JavaScript
        data = page.evaluate("""
            () => {
                return {
                    title: document.title,
                    links: Array.from(document.querySelectorAll('a')).map(a => ({
                        text: a.innerText,
                        href: a.href,
                    })),
                };
            }
        """)
        print(f"Found {len(data['links'])} links")
    finally:
        browser.close()

複数ページのナビゲーション

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()

        urls = [
            "https://web-scraping.dev/products",
            "https://web-scraping.dev/products?page=2",
            "https://web-scraping.dev/products?page=3",
        ]

        results = []
        for url in urls:
            page.goto(url)
            results.append({"url": url, "content": content})

        print(f"Scraped {len(results)} pages")
    finally:
        browser.close()

ファイルのダウンロード

ページ内で開始されたダウンロード（ログイン後のクリック、アンチボットフローなど）は セッションによって取得され、CDP 経由で ScrapiumBrowser.* コマンドを使って読み取れます。これらはブラウザーレベルのコマンドです。CDP セッションで送信してください。

コマンド	パラメーター	戻り値
ScrapiumBrowser.hasDownloads	なし	{"result": true|false}
ScrapiumBrowser.getDownloadsMetadatas	なし	{"metadata": {"<filename>": <bytes>}} — サイズのみ。内容や状態は含まれません
ScrapiumBrowser.getDownloadMetadatas	filename	1 ファイルの {"size": <bytes>}
ScrapiumBrowser.getDownload	filename（必須）、delete（任意）	{"data": "<base64>"} — 1 ファイル。内容は一度だけ転送されます
ScrapiumBrowser.getDownloads	delete（任意）	すべてのファイルの内容。files（名前別）と files_by_guid の両方で、2 回転送されます

読み取る前に待つ

ダウンロードの完了前に読み取りを待機させるキューはありません。state == "completed" の Browser.downloadProgress を待ち受けるか、ファイルが現れてサイズの変化が止まるまで getDownloadsMetadatas をポーリングしてください。クリックの直後に getDownloads を呼び出すと {} が返され、検索時に KeyError が発生します。

import base64
import time

def wait_for_download(cdp, filename, timeout=60):
    """Poll until the file exists and its size is stable across two reads."""
    deadline = time.monotonic() + timeout
    last = -1
    while time.monotonic() < deadline:
        size = cdp.send("ScrapiumBrowser.getDownloadsMetadatas")["metadata"].get(filename)
        if size is not None and size == last:
            return size
        last = size if size is not None else -1
        time.sleep(0.5)
    raise TimeoutError(f"{filename} did not finish downloading")


with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()
        cdp = context.new_cdp_session(page)

        page.goto("https://web-scraping.dev/file-download")
        page.click("#download-btn")

        size = wait_for_download(cdp, "download-sample.pdf")

        # One file by name: content is carried once, so the reply is ~1.34x the file.
        result = cdp.send("ScrapiumBrowser.getDownload", {"filename": "download-sample.pdf"})
        open("download-sample.pdf", "wb").write(base64.b64decode(result["data"]))
    finally:
        browser.close()

ダウンロードしたファイルはセッション終了時に削除されます。長時間実行するセッション中にディスク領域を回収するには、getDownload / getDownloads に delete: True を渡してください。

大きなファイル：リクエストする前に応答サイズを見積もる

getDownloads はペイロード全体を 1つの CDP メッセージに含めて応答します。クライアントの上限を超える WebSocket メッセージは、読み込みに時間がかかるのではなくプロトコルエラーになります。クライアントはコード 1009 で接続を閉じ、セッションも終了するため、Target closed / Session closed として表面化します。

• getDownloads の応答サイズはファイルの約 2.7倍 です。base64 による増加分が3分の1あり、コンテンツは files と files_by_guid の両方に含まれます。
• getDownload の応答サイズはファイルの約 1.34倍 です。ファイル名がわかっている場合はこちらを優先してください。
• サーバー側の上限：1ファイルあたり500 MB。 上限を超えるファイルはエンコードされず、そのファイルが含まれる呼び出しは失敗します。getDownloads は download too large: "export.csv" (620000000 bytes) exceeds the 524288000 byte limit per file と応答し、問題なく取得できる小さなファイルも含め、何も返しません。getDownloadsMetadatas には上限超過のファイルも引き続き表示され、getDownload ではほかのファイルを名前で取得できます。この2つの組み合わせが復旧手段です。
• まず getDownloadsMetadatas を呼び出し、実際のサイズに応じて処理を分岐してください。数十MBを超える場合は、インライン化せずにストリーミングしてください。

downloads_stream=1 で大きなダウンロードをストリーミングする

Cloud Browser の WebSocket URL に downloads_stream=1 を追加します。これは接続オプションであり、セッション開始後に有効化することはできません。その場合、getDownloads はより小さな形式で応答します。各ファイルのコンテンツは files_by_guid に1回だけ含まれ、GUID とファイル名の対応は guid_names に格納されます。また、base64 で約8 MBを超えるコンテンツは、インラインではなく IO ハンドルとして streams に返されます。

{
  "files": {},
  "files_by_guid": {"9F2C1A4E-...": "JVBERi0xLjQKJe..."},
  "guid_names": {"9F2C1A4E-...": "invoice.pdf", "B71D0E88-...": "export.csv"},
  "streams": {
    "B71D0E88-...": {"stream": "scrapfly-download:1", "size": 91250688, "name": "export.csv"}
  }
}

stream の値は通常の IO ハンドルで、IO.read / IO.close を使って読み取ります。

import base64
from urllib.parse import urlencode
from playwright.sync_api import sync_playwright

params = {"api_key": API_KEY, "proxy_pool": "datacenter", "downloads_stream": "1"}
BROWSER_WS = f"wss://browser.scrapfly.io?{urlencode(params)}"

CHUNK = 8 * 1024 * 1024  # base64 characters, not raw bytes (32 MB is the cap)


def drain(cdp, handle, destination):
    """Read a scrapfly-download: handle to eof. Sequential only, no offset."""
    encoded = []
    try:
        while True:
            read = cdp.send("IO.read", {"handle": handle, "size": CHUNK})
            if read.get("data"):
                encoded.append(read["data"])
            if read.get("eof"):
                break
    finally:
        try:
            cdp.send("IO.close", {"handle": handle})
        except Exception:
            pass

    # Chunks are cut on base64 group boundaries, so they concatenate as text
    # and decode once.
    with open(destination, "wb") as out:
        out.write(base64.b64decode("".join(encoded)))


with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(BROWSER_WS)
    try:
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else context.new_page()
        cdp = context.new_cdp_session(page)

        page.goto("https://web-scraping.dev/file-download")
        page.click("#download-btn")
        wait_for_download(cdp, "download-sample.pdf")

        result = cdp.send("ScrapiumBrowser.getDownloads")

        # Small files still arrive inline, large ones as handles. Handle both.
        for guid, b64 in result.get("files_by_guid", {}).items():
            open(result["guid_names"][guid], "wb").write(base64.b64decode(b64))
        for guid, stream in result.get("streams", {}).items():
            drain(cdp, stream["stream"], stream["name"])
    finally:
        browser.close()

ダウンロードハンドルのルール（Chromium 独自の IO ストリームとは異なります）：

• 読み取りは順次行われます。 offset を渡すとエラーになります。バイト列は読み取るたびに解放されます。
• ここでの size は、生のバイト数ではなくbase64文字数です。 チャンクの上限は32 MBで、size が送信されない場合のデフォルトは4 MBです。
• eof が true になるまでループしてください。 eof がない空の data は終端ではありません。
• エラー時も含め、必ず IO.close を呼び出してください。 放棄されたハンドルは120秒後に回収され、eof に到達すると解放されますが、閉じることが仕様上の要件です。
• セッションで処理されるのは scrapfly-download:<n> ハンドルのみです。blob: ハンドル、Page.printToPDF のストリーム、その他すべての IO ハンドルは、Chromium にそのまま渡されます。

ダウンロードなし：ページから URL の内容をストリーミングする

URL がわかっていて、ブラウザーのダウンロード機能が不要な場合は、ページ内で取得して Blob をストリーミングします。大きなデータがネットワークを通過することはなく、この方法では500 MBの上限も適用されません。

CHUNK = 10 * 1024 * 1024  # raw bytes here; the reply inflates ~1.34x

evaluated = cdp.send("Runtime.evaluate", {
    "expression": f"fetch({url!r}).then(r => r.blob())",
    "awaitPromise": True,
})
resolved = cdp.send("IO.resolveBlob", {"objectId": evaluated["result"]["objectId"]})
handle = "blob:" + resolved["uuid"]
# Drain with IO.read / IO.close, decoding each chunk when base64Encoded is true.

Page.printToPDF は transferMode: "ReturnAsStream" を受け取り、同じ方法でハンドルを返します。

大きな本文には Network.getResponseBody を使用しないでください。 Chromium はリクエスト前に大きなレスポンス本文をインスペクターのキャッシュから削除することがあり（Request content was evicted from inspector cache）、テキストのコンテンツタイプでは本文がすでにデコードされた状態で返されるため、バイナリペイロードが気付かないうちに破損します。

ボット対策機能

Cloud Browser はブラウザーのフィンガープリントを自動的に管理します。

• 実際のブラウザーに一致する TLS/JA3 フィンガープリント
• User-Agent、デバイスのメトリクス、タイムゾーン、ロケール（country に基づく）
• WebGL と Canvas のフィンガープリントのエミュレーション
• HTTP/2 フィンガープリント

重要な注意事項

• Cloud Browser への接続には、必ず connect_over_cdp() を使用してください（connect() は使用しないでください）。
• 事前に設定されたコンテキストを取得するには、必ず browser.contexts[0] を使用してください。新しいコンテキストは作成しないでください。Cloud Browser には、適切なフィンガープリント設定が施されたコンテキストが用意されています。
• 課金を停止するため、必ず try/finally ブロック内でブラウザーを閉じてください。閉じ忘れると、課金が継続します。
• フィンガープリントの上書き（ユーザーエージェント、ビューポート、タイムゾーンなど）は設定しないでください。これらは自動的に管理されます。
• セッションは、1 時間操作がないと期限切れになります。
• セッションの最大継続時間は 30 分（1800 秒）です。
• ダウンロードできるファイルの最大サイズは 1 ファイルあたり 500 MB です。上限を超えるファイルが 1 つでもあると、getDownloads の呼び出し全体が失敗します。数十 MB を超えるファイルでは、downloads_stream=1 を使用して IO ハンドルを最後まで読み取ってください。詳しくはファイルのダウンロードを参照してください。
• proxy_pool=datacenter から始め、ボット対策の回避が必要な場合にのみ residential に切り替えてください。
• トラブルシューティングの際にセッションの動画を記録するには、debug=true を使用してください。