Scrapfly 抓取器

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

适用场景

• 抓取网页内容（HTML、JSON、文本、Markdown）
• 绕过反机器人保护（Cloudflare、DataDome、PerimeterX、Kasada 等）
• 使用带地理定位的轮换代理收集数据
• 使用无头浏览器渲染 JavaScript 重度页面
• 通过 JavaScript 场景控制浏览器执行常见操作（等待选择器、点击、填写表单等）
• 会话复用，用于保持会话状态的抓取
• 捕获浏览器 XHR 请求数据

设置

pip install scrapfly-sdk

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

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 参数

ScrapeApiResponse

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

response.content                      # 页面内容（HTML/JSON/文本）
response.scrape_result                # 完整结果字典
response.status_code                  # Scrapfly API HTTP 状态码
response.scrape_result["status_code"] # 上游 HTTP 状态码
response.headers                      # 响应头部
response.context                      # 使用功能的元数据

示例

基础抓取

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",
    asp=True,  # 启用 ASP 以绕过反机器人机制
    country="us",  # 将代理与目标域名所在国家匹配
    proxy_pool="public_residential_pool",  # 使用住宅代理池，模拟真实 ISP IP 地址
))

使用 JavaScript 渲染进行抓取

result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/products",
    render_js=True,  # 启用云浏览器渲染 JavaScript
    rendering_wait=5000,  # 渲染等待时间（毫秒）
    wait_for_selector="div.product",  # 等待页面上指定选择器出现
))

以 Markdown 格式抓取内容

result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/products",
    format="markdown",  # 支持的其他格式包括：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,
    # 浏览器控制指令
    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
            }
        }
    ],
    # 请求头设置
    headers={
        "cookie": "cookiesAccepted=true"
    }
))

# 访问登录后的页面元素
print(result.selector.css("div#secret-message::text").get())

内置 Parsel 选择器

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

# 使用内置的 Parsel 选择器
selector = result.selector
for product in selector.css("div.product"):
    print("商品名称:", product.xpath(".//a/text()").get())  # 使用 XPath
    print("商品价格:", product.css("div.price::text").get())  # 使用 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())

使用会话实现状态化抓取

# 第一次请求：登录
result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/login",
    render_js=True,
    # 浏览器操作指令
    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
            }
        }
    ],
    # 请求头设置
    headers={
        "cookie": "cookiesAccepted=true"
    },
    session="logged-in-session"
))

# 第二次请求：使用相同会话访问受保护页面
result = client.scrape(ScrapeConfig(
    url="https://web-scraping.dev/login",
    session="logged-in-session"
))

# 使用内置选择器访问内容
print(result.selector.css("div#secret-message::text").get())

缓存响应结果

result = client.scrape(ScrapeConfig(
    url="https://example.com",
    cache=True,
    cache_ttl=3600,  # 缓存时长为 1 小时
))

错误处理

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

try:
    result = client.scrape(ScrapeConfig(url="https://httpbin.dev", asp=True))
except ScrapflyThrottleError as e:
    print(f"请求过于频繁，建议在 {e.retry_delay} 秒后重试")
except UpstreamHttpClientError as e:
    print(f"目标服务器返回 4xx 错误：{e.message}")
except UpstreamHttpServerError as e:
    print(f"目标服务器返回 5xx 错误：{e.message}")
except ScrapflyProxyError as e:
    print(f"代理错误：{e.message}")
except ScrapflyError as e:
    print(f"Scrapfly 错误：{e.message}")

重要注意事项

• render_js=True 是使用 rendering_wait、wait_for_selector、js_scenario、screenshots 等功能的前提
• 使用 asp=True 时，推荐搭配 proxy_pool="public_residential_pool" 以获得更真实的访问行为
• 使用 format="markdown" 可获取结构清晰的内容，适合用于大模型处理
• 使用 session 参数可在多个请求间保持状态（如登录状态）
• concurrent_scrape 方法会自动处理速率限制