scrapfly-browser

Scrapfly 云端浏览器

使用 Scrapfly Cloud Browser API 和 Python Playwright 自动化远程云端浏览器，内置代理轮换、反机器人指纹识别和地理定位功能。

适用场景

• 在远程云端浏览器上自动执行浏览器交互（登录流程、填写表单、页面导航）
• 抓取需要完整浏览器自动化的 JavaScript 密集型网站
• 使用托管的浏览器指纹识别绕过反机器人保护
• 通过地理定位代理运行 Playwright 脚本
• 在多个连接间保持持久浏览器会话
• 通过浏览器交互下载文件

设置

pip install playwright

必须通过环境变量 SCRAPFLY_API_KEY 提供 API 密钥，或直接将其传入连接 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" 表示将大型下载作为 IO 流句柄返回，而不是内联 base64

构建连接 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()

Async 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	单个文件的大小：{"size": <bytes>}
ScrapiumBrowser.getDownload	filename（必填），delete（可选）	{"data": "<base64>"} — 单个文件，内容仅传输一次
ScrapiumBrowser.getDownloads	delete（可选）	所有文件的内容会传输两次：files（按名称索引）和 files_by_guid

读取前等待

下载不会排队等待读取操作。请监听 Browser.downloadProgress，直到 state == "completed"，或轮询 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()

会话结束时会清理已下载的文件。在长时间运行的会话中，传入 delete: True 调用 getDownload / getDownloads 可回收磁盘空间。

大文件：请求前先估算回复大小

getDownloads 会将完整负载放在一条 CDP 消息中返回，而超过客户端限制的 WebSocket 消息属于协议错误，并非读取缓慢：客户端会以代码 1009 关闭连接，随之终止会话，并显示为 Target closed / Session closed。

• getDownloads 回复约为文件大小的 2.7 倍——base64 会增加三分之一的体积，而且内容会同时放在 files 和 files_by_guid 下。
• getDownload 回复约为文件大小的 1.34 倍。知道文件名时优先使用它。
• 服务端上限：每个文件 500 MB。 超大文件不会被编码，而且包含该文件的调用会失败：getDownloads 会返回 download too large: "export.csv" (620000000 bytes) exceeds the 524288000 byte limit per file，并且什么也不返回，包括那些大小合适的小文件。 getDownloadsMetadatas 仍会列出超大文件，而 getDownload 仍可按名称获取其他文件——这两个接口可用于恢复。
• 先调用 getDownloadsMetadatas，根据实际大小选择处理方式。文件超过几十 MB 时，使用流式传输，而不是内嵌内容。

使用 downloads_stream=1 流式传输大型下载

在 Cloud Browser WebSocket URL 中添加 downloads_stream=1。这是连接选项，会话开始后无法再启用。随后，getDownloads 会返回更精简的结构——每个文件的内容只在 files_by_guid 下出现一次，guid_names 将 GUID 映射到文件名，任何超过约 8 MB base64 的内容都会通过 streams 作为 IO 句柄返回，而不是内嵌在回复中：

{
  "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。 data 为空但 eof 不为 true 时，并不表示已结束。
• 务必调用 IO.close，包括出错时。未使用的句柄会在 120 秒后回收，到达 eof 时也会释放，但按约定仍应关闭。
• 会话只会处理 scrapfly-download:<n> 句柄。blob: 句柄、Page.printToPDF 流及其他所有 IO 句柄都会原样交给 Chromium 处理。

完全不触发下载：在页面中流式传输 URL 内容

如果已知 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 指纹

重要说明

• 始终使用 connect_over_cdp()（而不是 connect()）连接到 Cloud Browser
• 始终使用 browser.contexts[0] 获取预配置的上下文。不要创建新上下文——Cloud Browser 提供了具有适当指纹配置的上下文
• 始终在 try/finally 块中关闭浏览器，以停止计费。忘记关闭会导致持续产生费用
• 不要设置指纹覆盖项（用户代理、视口、时区等）——这些由系统自动管理
• 会话在闲置 1 小时后过期
• 会话最长持续时间为 30 分钟（1800 秒）
• 每个文件的下载大小上限为 500 MB，任何超大文件都会导致整个 getDownloads 调用失败。对于大小超过数十 MB 的文件，请使用 downloads_stream=1 并读取完 IO 句柄，参见文件下载
• 从 proxy_pool=datacenter 开始，仅在需要绕过反机器人检测时升级到 residential
• 排查问题时，使用 debug=true 录制会话视频