scrapfly-scraper

Scrapfly Scraper

使用 Scrapfly Scraper API 收集网页数据，支持代理轮换、绕过反机器人防护、JavaScript 渲染，以及用于控制浏览器的 JavaScript 场景。

使用场景

• 抓取网页（HTML、JSON、文本、Markdown）
• 绕过反机器人防护（Cloudflare、DataDome、PerimeterX、Kasada 等）
• 通过轮换代理并进行地理定位来收集数据
• 使用无头浏览器渲染 JavaScript 密集型页面
• 使用 JavaScript 场景控制浏览器，执行常见操作（等待选择器、点击、填写元素等）
• 通过复用会话来进行会话持久化抓取
• 捕获浏览器 XHR 调用数据

设置

pip install scrapfly-sdk

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

API 参考

端点： https://api.scrapfly.io/scrape HTTP 方法会转发到上游 URL。 要从网页或 API 获取数据，请使用 GET 方法，该方法为默认方法。 如果 API 资源需要其他方法（例如 POST），可以通过 ScrapeConfig 的 method 参数使用。

ScrapflyClient

from scrapfly import ScrapflyClient, ScrapeConfig, ScrapeApiResponse
import os

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

ScrapeConfig 参数

参数	类型	默认值	描述
url	str	必填	要抓取的目标 URL
method	str	"GET"	HTTP 方法：GET、POST、PUT、PATCH、HEAD、OPTIONS
headers	dict	None	自定义 HTTP 标头
cookies	dict	None	自定义 Cookie（合并到 headers["cookie"]）
body	str	None	POST/PUT/PATCH 请求的原始请求体；用于预先编码负载的情况
data	dict	None	POST/PUT/PATCH 请求的表单/JSON 数据；未提供 body 时，根据 Content-Type 进行编码
timeout	int	None	请求超时时间，单位为毫秒（默认约 150,000 毫秒）
retry	bool	True	网络故障时自动重试
country	str	None	代理国家/地区（ISO 3166-1 alpha-2，例如 "us"、"de"）
proxy_pool	str	None	代理池："basic"（数据中心代理，不启用 Unblocker，1 个积分）、"unblocker"（住宅代理和 Unblocker 反机器人绕过功能，25 个积分）、"unblocker_ultra"（同上，也是唯一支持目标站点或具有独立价格的反机器人保护的代理池；该价格另行计费），或 "auto"（API 根据目标站点的配置文件，在请求前选择代理池：在数据中心代理上使用 Unblocker，或选择目标站点所需的代理池；所有目标站点均可使用，包括要求使用 unblocker_ultra 的目标站点，其价格另行计费；按实际使用的代理池计价）。每个代理池都会自行设置 Unblocker。未设置时，SDK 不会发送 proxy_pool。迁移指南介绍了代理池推出前编写的请求，包括已弃用的代理池名称 "public_datacenter_pool" 和 "public_residential_pool"
session	str	None	会话 ID，用于在多个请求之间保留 Cookie/指纹
session_sticky_proxy	bool	True	为指定的 session 保持相同的代理 IP；在 "unblocker"、"unblocker_ultra" 和 "auto" 上，False 将被拒绝
unblocker	bool	False	已弃用：请使用 proxy_pool（Unblocker 对应 "unblocker_ultra"）。迁移指南介绍了代理池推出前编写的请求。asp 是它的别名；两者同时提供时，以 asp 为准
render_js	bool	False	启用无头浏览器 JavaScript 渲染（+5 个积分）
rendering_wait	int	None	页面加载后等待的时间，单位为毫秒（需要 render_js=True）
rendering_stage	str	"complete"	浏览器就绪阶段："complete" 或 "domcontentloaded"（需要 render_js=True）
wait_for_selector	str	None	要等待的 CSS/XPath 选择器（需要 render_js=True）
js	str	None	要在浏览器中执行的 JavaScript 代码（自动编码）
js_scenario	list	None	浏览器操作列表（点击、填充、滚动、等待等）
auto_scroll	bool	None	渲染期间自动滚动页面，以加载延迟加载的内容（需要 render_js=True）
screenshots	dict	None	截取屏幕截图：{"fullpage": "png"} 或 {"selector": ".element"}（需要 render_js=True）
screenshot_flags	list[str]	None	截图选项："load_images"、"dark_mode"、"block_banners"、"high_quality"、"print_media_format"
format	str	"raw"	输出格式："raw"、"clean_html"、"json"、"markdown"、"text"
format_options	list[str]	None	格式修饰选项（仅适用于 Markdown）："no_images"、"no_links"、"only_content"
extract	dict	None	要应用于响应的原始提取规范（编码后以 extract 发送）
extraction_template	str	None	已保存的服务端提取模板名称
extraction_ephemeral_template	dict	None	仅使用一次的内嵌 JSON 提取模板（ephemeral:）
extraction_prompt	str	None	提供给 Extraction API 的自然语言说明
extraction_model	str	None	Extraction API 要使用的大模型名称
cache	bool	False	启用响应缓存
cache_ttl	int	None	缓存生存时间，单位为秒
cache_clear	bool	False	当 cache=True 时，清除任何现有的缓存响应
dns	bool	False	在 result.dns 中收集 DNS 记录和耗时（速度较慢）
ssl	bool	False	在 result.ssl 中收集 SSL 证书详情（速度较慢）
debug	bool	False	启用调试记录和额外元数据
raise_on_upstream_error	bool	True	上游返回 4xx/5xx HTTP 响应时引发异常
correlation_id	str	None	用于跨系统跟踪请求的自定义 ID
tags	list[str]	None	用于请求整理和分析的自定义标签
lang	list[str]	None	Accept-Language 值，例如 ["en-US", "en"]
os	str	None	覆盖浏览器 OS 指纹，例如 "windows"、"macos"
webhook	str	None	用于异步抓取完成回调的命名 webhook
cost_budget	int	None	Unblocker 在 unblocker、unblocker_ultra 和 auto 代理池上处理请求时可使用的最大积分数

ScrapeApiResponse

response = client.scrape(ScrapeConfig(url="https://httpbin.dev"))

response.content                      # Page content (HTML/JSON/text)
response.scrape_result                # Full result dict
response.status_code                  # Scrapfly API HTTP status code
response.scrape_result["status_code"] # Upsteam HTTP status code
response.headers                      # Response headers
response.context                      # Metadata about features used

示例

基本抓取

from scrapfly import ScrapflyClient, ScrapeConfig
import os

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

result = client.scrape(ScrapeConfig(
    url="https://httpbin.dev",
))
print(result.content)

使用反机器人绕过和地理定位进行抓取

result = client.scrape(ScrapeConfig(
    url="https://httpbin.dev",
    proxy_pool="unblocker", # residential proxies and the Unblocker to bypass antibots
    country="us", # match the proxy with the domain country
))

使用 JavaScript 渲染进行抓取

result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/products",
    render_js=True, # enable JavaScript rendering using a cloud browser
    rendering_wait=5000, # rendering wait for wait for
    wait_for_selector="div.product", # wait a for a selector to be present on the page
))

将抓取结果转换为 Markdown

result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/products",
    format="markdown", # other supported formats are: json, text, clean_html 
))
print(result.content)

带请求正文的 POST 请求

result = client.scrape(ScrapeConfig(
    url="https://httpbin.dev/anything",
    method="POST",
    headers={"Content-Type": "application/json"},
    body='{"query": "search term"}',
))
print(result.content)

JavaScript 场景（浏览器操作）

result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/login",
    render_js=True,
    # browser control
    js_scenario=[
        {
            "fill":{
                "selector":"input[name='username']","clear":True,"value":"user123"
                }
            },{
            "fill":{
                "selector":"input[name='password']","clear":True,"value":"password"
                }
            },{
            "click":{
                "selector":"form > button[type='submit']"
                }
            },{
            "wait_for_navigation":{
                "timeout":5000
                }
            }
    ],
    # request headers
    headers={
        "cookie":"cookiesAccepted=true"
    }
))

# access the element under login
print(result.selector.css("div#secret-message::text").get())

内置 Parsel 选择器

result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/products",
    render_js=True
))

# access the built-in Parsel selector
selector = result.selector
for product in selector.css("div.product"):
    print("product name:", product.xpath(".//a/text()").get()) # using xpath
    print("product price:", product.css("div.price::text").get()) # using css

并发抓取

import asyncio
from scrapfly import ScrapflyClient, ScrapeConfig

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

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

configs = [ScrapeConfig(url=url) for url in urls]

async def concurrent_scraping():
    results = []
    async for result in client.concurrent_scrape(configs):
        results.append(result)

    for result in results:
        print(result.content[:100])
        print("===========")

asyncio.run(concurrent_scraping())

使用会话进行有状态抓取

# First request: login
result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/login",
    render_js=True,
    # browser control
    js_scenario=[
        {
            "fill":{
                "selector":"input[name='username']","clear":True,"value":"user123"
                }
            },{
            "fill":{
                "selector":"input[name='password']","clear":True,"value":"password"
                }
            },{
            "click":{
                "selector":"form > button[type='submit']"
                }
            },{
            "wait_for_navigation":{
                "timeout":5000
                }
            }
    ],
    # request headers
    headers={
        "cookie":"cookiesAccepted=true"
    },
    session="logged-in-session"
))

# Second request: access protected page with same session
result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/login",
    session="logged-in-session"
))

# access the built-in selector
print(result.selector.css("div#secret-message::text").get())

缓存响应

result = client.scrape(ScrapeConfig(
    url="https://example.com",
    cache=True,
    cache_ttl=3600,  # Cache for 1 hour
))

错误处理

from scrapfly import ScrapflyClient, ScrapeConfig
from scrapfly.errors import (
    ScrapflyError,
    UpstreamHttpClientError,
    UpstreamHttpServerError,
    ScrapflyProxyError,
    ScrapflyThrottleError,
)

try:
    result = client.scrape(ScrapeConfig(url="https://httpbin.dev", proxy_pool="unblocker"))
except ScrapflyThrottleError as e:
    print(f"Rate limited, retry after {e.retry_delay}s")
except UpstreamHttpClientError as e:
    print(f"Target returned 4xx: {e.message}")
except UpstreamHttpServerError as e:
    print(f"Target returned 5xx: {e.message}")
except ScrapflyProxyError as e:
    print(f"Proxy error: {e.message}")
except ScrapflyError as e:
    print(f"Scrapfly error: {e.message}")

需要单独付费的目标或反机器人防护支持使用 proxy_pool="unblocker_ultra" 和 "auto"，这会增加相应费用；使用 basic 和 unblocker 时会返回 ERR::PROXY::UNBLOCKER_ULTRA_REQUIRED（HTTP 422，不计费）。请改用 unblocker_ultra 重新发送：

try:
    result = client.scrape(ScrapeConfig(url=url, proxy_pool="unblocker"))
except ScrapflyError as e:
    if e.code != "ERR::PROXY::UNBLOCKER_ULTRA_REQUIRED":
        raise
    result = client.scrape(ScrapeConfig(url=url, proxy_pool="unblocker_ultra"))

重要说明

• rendering_wait、wait_for_selector、js、js_scenario 和 screenshots 需要设置 render_js=True
• 根据网站需要选择代理池：数据中心代理可访问的网站使用 "basic"；需要住宅代理和 Unblocker 的网站使用 "unblocker"；API 返回 ERR::PROXY::UNBLOCKER_ULTRA_REQUIRED 时使用 "unblocker_ultra"；也可以使用 "auto"，让 API 根据目标网站的配置文件进行选择
• unblocker=True 及其别名 asp=True 已弃用：请使用 proxy_pool（Unblocker 对应 "unblocker_ultra"）。迁移指南介绍了代理池推出前编写的请求
• 使用 format="markdown" 获取适合大模型使用的整洁内容
• 使用 session 参数在多个请求间保持状态
• concurrent_scrape 方法会自动处理速率限制