抓取动态网站

抓取动态网站

这一页介绍 DynamicFetcher 类（旧名 PlayWrightFetcher）。它提供灵活的浏览器自动化能力，拥有多种配置选项，并在底层内置了一些轻量级的隐匿优化。

正如稍后会讲到的，要对页面进行自动化操作，你需要对 Playwright 的 Page API 有一定了解。

Prerequisites

1. 你已经完成或阅读过 Fetcher 基础 页面，理解了 Response 对象 是什么，以及应该使用哪种 fetcher。
2. 你已经完成或阅读过 查询元素 页面，理解了如何从 Selector / Response 对象中查找和提取元素。
3. 你已经完成或阅读过 核心类 页面，了解 Response 类从 Selector 类继承了哪些属性 / 方法。

基本用法

这个 fetcher 的主要导入方式如下；这也是所有 fetcher 通用的导入方式：

from scrapling.fetchers import DynamicFetcher

如何配置解析选项请看这里。

接下来我们会结合示例逐一介绍大部分参数。如果你只想快速查看参数总表，请直接跳到完整参数列表。

fetch 方法对应的异步版本自然是 async_fetch。

这个 fetcher 目前提供三种可按需组合的主要运行方式。

分别是：

1. 原生 Playwright

DynamicFetcher.fetch('https://example.com')

这样使用时会启动一个 Chromium 浏览器并加载页面。Scrapling 会自动做一些速度优化，也会在底层添加少量隐匿处理；除此之外，如果你没有显式启用额外功能，它基本就是一套原生的 Playwright API。

2. 真实 Chrome

DynamicFetcher.fetch('https://example.com', real_chrome=True)

如果你的设备上安装了 Google Chrome，建议使用这个选项。它与第一种方式类似，但会改用你本机已安装的 Google Chrome，而不是 Chromium。这样请求特征会更接近真实用户，更不容易被检测到。

如果你还没有安装 Google Chrome，又想使用这个选项，可以在终端中执行下面的命令，让 Playwright 为该库安装 Chrome，而无需手动安装：

playwright install chrome

3. CDP 连接

DynamicFetcher.fetch('https://example.com', cdp_url='ws://localhost:9222')

你也可以不在本地启动浏览器（Chromium / Google Chrome），而是通过 Chrome DevTools Protocol 连接到远程浏览器。

注意

• 这里曾经有一个 stealth 选项，但从 0.3.13 版本开始，它被移动到了下一页要介绍的 StealthyFetcher 类中，并带来了更多功能。
  
• 这样做能降低新用户的理解成本，也更容易维护；更多原因可见 StealthyFetcher 页面。

完整参数列表

Scrapling 为这个 fetcher 及其 session 类提供了很多选项。为了尽量清晰，我们先在这里列出所有参数，并在后文中通过示例说明大部分用法。

参数	说明	可选
url	目标 URL	❌
headless	传入 True 以无头 / 隐藏模式运行浏览器（默认值），传入 False 以有界面 / 可见模式运行。	✔️
disable_resources	丢弃不必要资源的请求以提升速度。会被丢弃的请求类型包括 font、image、media、beacon、object、imageset、texttrack、websocket、csp_report 和 stylesheet。	✔️
cookies	为下一次请求设置 cookies。	✔️
useragent	传入要使用的 User-Agent 字符串。否则，fetcher 会自动生成并使用与当前浏览器及版本一致的真实 User-Agent。	✔️
network_idle	等待页面进入至少 500 毫秒没有网络连接的状态。	✔️
load_dom	默认启用，等待页面中的所有 JavaScript 完整加载并执行完毕（即等待 domcontentloaded 状态）。	✔️
timeout	页面内所有操作和等待使用的超时时间（毫秒）。默认是 30,000 毫秒（30 秒）。	✔️
wait	所有流程完成后，在关闭页面并返回 Response 对象之前，fetcher 额外等待的时间（毫秒）。	✔️
page_action	用于自动化。传入一个接收 page 对象的函数，它会在导航完成后执行，并完成你需要的自动化操作。	✔️
page_setup	一个接收 page 对象的函数，会在导航前运行。适合注册必须在页面加载前就设置好的事件监听器或路由。	✔️
wait_selector	等待某个指定 CSS 选择器进入某个状态。	✔️
init_script	一个 JavaScript 文件的绝对路径；会在该 session 中每个页面创建时执行。	✔️
wait_selector_state	指定 wait_selector 对应选择器要等待到的状态。默认状态是 attached。	✔️
google_search	默认启用。Scrapling 会设置 Google referer 头。	✔️
extra_headers	附加到请求上的额外请求头。如果同时启用 google_search，那么它设置的 referer 会优先于这里设置的 referer。	✔️
proxy	请求使用的代理。它可以是字符串，也可以是仅包含 server、username、password 三个键的字典。	✔️
real_chrome	如果设备上安装了 Chrome，启用后 Fetcher 会启动并使用你的 Chrome 实例。	✔️
locale	指定用户区域设置，例如 en-GB、de-DE 等。它会影响 navigator.language 的值、Accept-Language 请求头的值，以及数字和日期格式化规则。默认使用系统区域设置。	✔️
timezone_id	修改浏览器时区。默认使用系统时区。	✔️
cdp_url	不启动新的浏览器实例，而是连接到这个 CDP URL，通过 CDP 控制真实浏览器。	✔️
user_data_dir	User Data Directory 的路径，用于存储 cookies、本地存储等浏览器会话数据。默认会创建临时目录。仅适用于 session。	✔️
extra_flags	启动浏览器时附加的一组额外浏览器 flags。	✔️
additional_args	传递给 Playwright context 的附加参数，可作为额外设置使用，并且优先级高于 Scrapling 自己的设置。	✔️
selector_config	用于创建最终 Selector / Response 类的自定义解析参数字典。	✔️
blocked_domains	要阻止请求的域名集合。子域名同样会被匹配（例如 "example.com" 也会拦截 "sub.example.com"）。	✔️
block_ads	阻止请求到约 3,500 个已知广告 / 跟踪域名。可以与 blocked_domains 一起使用。	✔️
dns_over_https	使用代理时，通过 Cloudflare 的 DNS-over-HTTPS 路由 DNS 查询，防止 DNS 泄漏。	✔️
proxy_rotator	用于自动轮换代理的 ProxyRotator 实例。不能与 proxy 同时使用。	✔️
retries	请求失败时的重试次数。默认 3 次。	✔️
retry_delay	两次重试之间等待的秒数。默认 1 秒。	✔️
capture_xhr	传入一个正则 URL 模式字符串，用于在页面加载过程中捕获匹配的 XHR / fetch 请求。捕获到的响应可通过 response.captured_xhr 获取。默认值是 None（禁用）。	✔️
executable_path	自定义浏览器可执行文件的绝对路径，用于替代打包的 Chromium。适合非标准安装或自定义浏览器构建。	✔️

在 session 类中，上述所有参数都可以作为整个 session 的全局配置。不过，你仍然可以按请求单独配置其中一些可在浏览器标签页级别生效的参数，例如：google_search、timeout、wait、page_action、page_setup、extra_headers、disable_resources、wait_selector、wait_selector_state、network_idle、load_dom、blocked_domains、proxy 和 selector_config。

注意

1. 根据我的测试，disable_resources 在某些网站上可让请求提速约 25%，也能节省代理流量；但要谨慎使用，因为它也可能导致某些网站永远加载不完。
2. google_search 参数默认对所有请求启用，会把 referer 设为 https://www.google.com/。如果同时传入 extra_headers，它会优先于其中设置的 referer。
3. 从 0.3.13 版本开始，这里的 stealth 选项已移除，改由 StealthyFetcher 类提供；hide_canvas 选项也被移到了那里。disable_webgl 参数同样被移到 StealthyFetcher 中，并重命名为 allow_webgl。
4. 如果你没有设置 user agent 且启用了 headless 模式，fetcher 会自动生成与相同浏览器版本对应的真实 user agent 并使用它。如果你没有设置 user agent 且没有启用 headless 模式，fetcher 会直接使用浏览器默认的 user agent，这与现代标准浏览器中的表现一致。

会话管理

如果你希望在使用同一套配置发起多个请求时保持浏览器持续开启，请使用 DynamicSession / AsyncDynamicSession 类。它们可以接受 fetch 方法支持的全部参数，因此你可以为整个 session 指定统一配置。

from scrapling.fetchers import DynamicSession

# Create a session with default configuration
with DynamicSession(
    headless=True,
    disable_resources=True,
    real_chrome=True
) as session:
    # Make multiple requests with the same browser instance
    page1 = session.fetch('https://example1.com')
    page2 = session.fetch('https://example2.com')
    page3 = session.fetch('https://dynamic-site.com')

    # All requests reuse the same tab on the same browser instance

异步会话用法

import asyncio
from scrapling.fetchers import AsyncDynamicSession

async def scrape_multiple_sites():
    async with AsyncDynamicSession(
        network_idle=True,
        timeout=30000,
        max_pages=3
    ) as session:
        # Make async requests with shared browser configuration
        pages = await asyncio.gather(
            session.fetch('https://spa-app1.com'),
            session.fetch('https://spa-app2.com'),
            session.fetch('https://dynamic-content.com')
        )
        return pages

你可能已经注意到了 max_pages 参数。这是一个新参数，用来让 fetcher 创建一个轮换式浏览器标签页池。也就是说，它不再强制所有请求都复用同一个标签页，而是允许你设置同时最多可存在多少个页面 / 标签页。每次请求时，库会先关闭所有已经完成任务的标签页，然后检查当前标签页数量是否低于允许的最大值：

1. 如果仍在允许范围内，fetcher 会新建一个标签页，然后按正常流程继续。
2. 否则，它会每隔不到 1 秒持续检查一次是否可以新建标签页，最多等待 60 秒；如果仍不行，就抛出 TimeoutError。这通常发生在目标网站长时间无响应时。

这套逻辑允许同一个浏览器在同一时间抓取多个 URL，不仅节省资源，更重要的是速度非常快 :)

在 0.3 和 0.3.1 版本中，这个池曾经会复用已经完成任务的标签页，以进一步节省资源 / 时间。但这种逻辑后来被证明有缺陷，因为几乎不可能完全避免前一个请求配置对后一个请求造成污染。

会话的优势

• 浏览器复用：通过复用同一浏览器实例，大幅提升后续请求速度。
• Cookie 持久化：像真实浏览器一样自动处理 cookies 和会话状态。
• 一致的指纹：所有请求共享同一个浏览器指纹。
• 更高的内存效率：相比每次抓取都启动一个新浏览器，资源使用更合理。

示例

通过示例理解会更直观，下面我们来看看。

资源控制

# Disable unnecessary resources
page = DynamicFetcher.fetch('https://example.com', disable_resources=True)  # Blocks fonts, images, media, etc.

域名拦截

# Block requests to specific domains (and their subdomains)
page = DynamicFetcher.fetch('https://example.com', blocked_domains={"ads.example.com", "tracker.net"})

网络控制

# Wait for network idle (Consider fetch to be finished when there are no network connections for at least 500 ms)
page = DynamicFetcher.fetch('https://example.com', network_idle=True)

# Custom timeout (in milliseconds)
page = DynamicFetcher.fetch('https://example.com', timeout=30000)  # 30 seconds

# Proxy support (It can also be a dictionary with only the keys 'server', 'username', and 'password'.)
page = DynamicFetcher.fetch('https://example.com', proxy='http://username:***@host:port')

下载文件

page = DynamicFetcher.fetch('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/docs/assets/main_cover.png')

with open(file='main_cover.png', mode='wb') as f:
    f.write(page.body)

Response 对象的 body 属性始终返回 bytes。

导航前设置

如果你需要在页面跳转前就注册事件监听器、路由或脚本，请使用 page_setup。这个函数会接收 page 对象，并在调用 page.goto() 之前执行。

from playwright.sync_api import Page

def capture_websockets(page: Page):
    page.on("websocket", lambda ws: print(f"WebSocket opened: {ws.url}"))

page = DynamicFetcher.fetch('https://example.com', page_setup=capture_websockets)

异步版本：

from playwright.async_api import Page

async def capture_websockets(page: Page):
    page.on("websocket", lambda ws: print(f"WebSocket opened: {ws.url}"))

page = await DynamicFetcher.async_fetch('https://example.com', page_setup=capture_websockets)

它可以与 page_action 组合使用——page_setup 在导航前执行，page_action 在导航后执行。

浏览器自动化

这里就需要你对 Playwright 的 Page API 有一定了解。你传入的函数会拿到 Playwright API 中的 page 对象，执行你想要的动作，然后 fetcher 再继续后续流程。

这个函数会在等待 network_idle 完成后（如果启用了该选项）、等待 wait_selector 之前执行，因此它并不局限于自动化用途；你可以按自己的需要改动页面状态。

下面的示例里，我使用页面的 鼠标事件 先滚动页面，再移动鼠标。

from playwright.sync_api import Page

def scroll_page(page: Page):
    page.mouse.wheel(10, 0)
    page.mouse.move(100, 400)
    page.mouse.up()

page = DynamicFetcher.fetch('https://example.com', page_action=scroll_page)

当然，如果你使用异步版 fetch，那么这个函数本身也必须是异步函数。

from playwright.async_api import Page

async def scroll_page(page: Page):
   await page.mouse.wheel(10, 0)
   await page.mouse.move(100, 400)
   await page.mouse.up()

page = await DynamicFetcher.async_fetch('https://example.com', page_action=scroll_page)

等待条件

# Wait for the selector
page = DynamicFetcher.fetch(
    'https://quotes.toscrape.com/js-delayed/',
    wait_selector='.quote',
    wait_selector_state='visible'
)

如果启用了这项功能，那么这是 fetcher 在返回响应前执行的最后一次等待。你把 CSS 选择器传给 wait_selector，再把期望状态传给 wait_selector_state，fetcher 就会等待该状态满足。如果你没有显式传入状态，默认值是 attached，也就是等待元素出现在 DOM 中。

在这之后，如果 load_dom 仍然启用（默认值），fetcher 会再次检查所有 JavaScript 是否已完成加载与执行（即达到 domcontentloaded 状态），否则就继续等待。如果你启用了 network_idle，那么 fetcher 还会像前面解释的那样，再等待一次 network_idle 达成。

fetcher 可等待的状态包括以下几种（来源）：

• attached：等待元素出现在 DOM 中。
• detached：等待元素不再出现在 DOM 中。
• visible：等待元素拥有非空的边界框，且没有 visibility:hidden。注意，没有内容或设置了 display:none 的元素边界框为空，因此不算可见。
• hidden：等待元素从 DOM 中移除，或边界框为空，或 visibility:hidden。它与 visible 刚好相反。

捕获 XHR / Fetch 请求

很多单页应用会通过后台 API 调用（XHR / fetch）加载数据。你可以在 session 级别传入 capture_xhr 的正则 URL 模式来捕获这些请求：

from scrapling.fetchers import DynamicSession

with DynamicSession(capture_xhr=r"https://api\.example\.com/.*", headless=True) as session:
    page = session.fetch('https://example.com')

    # Access captured XHR responses
    for xhr in page.captured_xhr:
        print(xhr.url, xhr.status)
        print(xhr.body)  # Raw response body as bytes

captured_xhr 中的每一项都是完整的 Response 对象，具有相同的属性（.url、.status、.headers、.body 等）。当 capture_xhr 没有设置或其值为 None 时，captured_xhr 会是一个空列表。

一些隐匿特性

page = DynamicFetcher.fetch(
    'https://example.com',
    google_search=True,
    useragent='Mozilla/5.0...',  # Custom user agent
    locale='en-US',  # Set browser locale
)

通用示例

from scrapling.fetchers import DynamicFetcher

def scrape_dynamic_content():
    # Use Playwright for JavaScript content
    page = DynamicFetcher.fetch(
        'https://example.com/dynamic',
        network_idle=True,
        wait_selector='.content'
    )

    # Extract dynamic content
    content = page.css('.content')

    return {
        'title': content.css('h1::text').get(),
        'items': [
            item.text for item in content.css('.item')
        ]
    }

代理轮换

from scrapling.fetchers import DynamicSession, ProxyRotator

# Set up proxy rotation
rotator = ProxyRotator([
    "http://proxy1:8080",
    "http://proxy2:8080",
    "http://proxy3:8080",
])

# Use with session - rotates proxy automatically with each request
with DynamicSession(proxy_rotator=rotator, headless=True) as session:
    page1 = session.fetch('https://example1.com')
    page2 = session.fetch('https://example2.com')

    # Override rotator for a specific request
    page3 = session.fetch('https://example3.com', proxy='http://specific-proxy:8080')

请记住，默认情况下，所有基于浏览器的 fetcher 和 session 都会使用持久化浏览器上下文以及标签页池。不过，由于浏览器无法为每个标签页单独设置代理，当你使用 ProxyRotator 时，fetcher 会自动为每个代理分别创建独立的 context，并在每个 context 中只开一个标签页。等该标签页完成任务后，标签页和对应 context 都会被关闭。

何时使用

当你满足以下需求时，可以使用 DynamicFetcher：

• 需要浏览器自动化
• 希望拥有多种浏览器运行方式
• 想使用真实 Chrome 浏览器
• 需要自定义浏览器配置
• 只需要少量隐匿选项

如果你想在不增加太多配置复杂度的情况下获得更强的隐匿能力与控制力，可以继续阅读 StealthyFetcher。