抓取强防护动态网站

抓取强防护动态网站

这一页介绍 StealthyFetcher 类。它与 DynamicFetcher 非常相似，包括支持的浏览器、自动化方式以及对 Playwright API 的使用。主要区别在于：这个类提供了更高级的反机器人防护绕过能力；其中大部分会在底层自动处理，其余部分则由你自行按需启用。

和 DynamicFetcher 一样，要对页面进行自动化操作，你仍然需要掌握一些 Playwright 的 Page API 知识，后面会详细说明。

Prerequisites

1. 你已经完成或阅读过 DynamicFetcher 页面，因为这个类是在它的基础上构建的，所以这里不会重复同样的信息。
2. 你已经完成或阅读过 Fetcher 基础 页面，理解了 Response 对象 是什么，以及应该使用哪种 fetcher。
3. 你已经完成或阅读过 查询元素 页面，理解了如何从 Selector / Response 对象中查找和提取元素。
4. 你已经完成或阅读过 核心类 页面，了解 Response 类从 Selector 类继承了哪些属性 / 方法。

基本用法

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

from scrapling.fetchers import StealthyFetcher

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

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

它能做什么？

StealthyFetcher 可以看作是 DynamicFetcher 的隐匿增强版本。它具备的一些能力包括：

1. 自动绕过各类 Cloudflare Turnstile / Interstitial 挑战。
2. 绕过 CDP 运行时泄漏和 WebRTC 泄漏。
3. 隔离 JavaScript 执行环境，移除大量 Playwright 指纹，并避免一些已知的机器人行为特征被检测到。
4. 通过向 canvas 注入噪声，降低基于 canvas 的指纹识别风险。
5. 自动修补已知的无头模式检测方法，并提供选项来对抗时区不匹配攻击。
6. 以及更多反防护选项……

完整参数列表

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。	✔️
solve_cloudflare	启用后，fetcher 会在把响应返回给你之前，自动解决所有类型的 Cloudflare Turnstile / Interstitial 挑战。	✔️
block_webrtc	强制 WebRTC 遵循代理设置，以避免本地 IP 地址泄漏。	✔️
hide_canvas	为 canvas 操作添加随机噪声，以防止基于 canvas 的指纹识别。	✔️
allow_webgl	默认启用。关闭后会彻底禁用 WebGL 和 WebGL 2.0 支持。不建议禁用，因为现在很多 WAF 都会检查 WebGL 是否可用。	✔️
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、solve_cloudflare、blocked_domains、proxy 和 selector_config。

注意

1. 它本质上与 DynamicFetcher 的参数几乎一致，只是额外增加了：solve_cloudflare、block_webrtc、hide_canvas 和 allow_webgl。capture_xhr 则与 DynamicFetcher 共用。
2. 根据我的测试，disable_resources 在某些网站上可让请求提速约 25%，也能节省代理流量；但要谨慎使用，因为它也可能导致某些网站永远加载不完。
3. google_search 参数默认对所有请求启用，会把 referer 设为 https://www.google.com/。如果同时传入 extra_headers，它会优先于其中设置的 referer。
4. 如果你没有设置 user agent 且启用了 headless 模式，fetcher 会自动生成与相同浏览器版本对应的真实 user agent 并使用它。如果你没有设置 user agent 且没有启用 headless 模式，fetcher 会直接使用浏览器默认的 user agent，这与现代标准浏览器中的表现一致。

示例

通过示例理解会更容易，所以接下来我们会逐项介绍大部分参数。由于这个类和 DynamicFetcher 是同一体系，你也可以前往那一页查看更多示例；这里不会重复所有内容。

Cloudflare 与隐匿选项

# Automatic Cloudflare solver
page = StealthyFetcher.fetch('https://nopecha.com/demo/cloudflare', solve_cloudflare=True)

# Works with other stealth options
page = StealthyFetcher.fetch(
    'https://protected-site.com',
    solve_cloudflare=True,
    block_webrtc=True,
    real_chrome=True,
    hide_canvas=True,
    google_search=True,
    proxy='http://username:***@host:port',  # It can also be a dictionary with only the keys 'server', 'username', and 'password'.
)

solve_cloudflare 参数会启用对各类 Cloudflare Turnstile / Interstitial 挑战的自动检测与求解，包括：

• JavaScript 挑战（托管式）
• 交互式挑战（例如点击验证框）
• 隐形挑战（后台自动验证）

甚至也能处理内嵌 captcha 的自定义页面。

重要说明

1. 有些网站使用了自定义实现，因此在验证码通过后，你可能还需要配合 wait_selector，以确保 Scrapling 会继续等待真正的网站内容加载完成。为了尽可能通用，这类极端边缘情况有时仍需要手动补充等待条件。
2. 使用 Cloudflare 求解器时，timeout 最好至少设置为 60 秒，以留出足够的挑战求解时间。
3. 这一功能可以与代理及其他隐匿选项无缝配合使用。

浏览器自动化

这里就需要你对 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 = StealthyFetcher.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 StealthyFetcher.async_fetch('https://example.com', page_action=scroll_page)

等待条件

# Wait for the selector
page = StealthyFetcher.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 刚好相反。

真实场景示例（Amazon）

下面的示例仅用于教学；原文说明它由 AI 生成，这也从侧面展示了通过 AI 配合 Scrapling 的使用门槛并不高。

def scrape_amazon_product(url):
    # Use StealthyFetcher to bypass protection
    page = StealthyFetcher.fetch(url)

    # Extract product details
    return {
        'title': page.css('#productTitle::text').get().clean(),
        'price': page.css('.a-price .a-offscreen::text').get(),
        'rating': page.css('[data-feature-name="averageCustomerReviews"] .a-popover-trigger .a-color-base::text').get(),
        'reviews_count': page.css('#acrCustomerReviewText::text').re_first(r'[\d,]+'),
        'features': [
            li.get().clean() for li in page.css('#feature-bullets li span::text')
        ],
        'availability': page.css('#availability')[0].get_all_text(strip=True),
        'images': [
            img.attrib['src'] for img in page.css('#altImages img')
        ]
    }

会话管理

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

from scrapling.fetchers import StealthySession

# Create a session with default configuration
with StealthySession(
    headless=True,
    real_chrome=True,
    block_webrtc=True,
    solve_cloudflare=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://nopecha.com/demo/cloudflare')

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

异步会话用法

import asyncio
from scrapling.fetchers import AsyncStealthySession

async def scrape_multiple_sites():
    async with AsyncStealthySession(
        real_chrome=True,
        block_webrtc=True,
        solve_cloudflare=True,
        timeout=60000,  # 60 seconds for Cloudflare challenges
        max_pages=3
    ) as session:
        # Make async requests with shared browser configuration
        pages = await asyncio.gather(
            session.fetch('https://site1.com'),
            session.fetch('https://site2.com'),
            session.fetch('https://protected-site.com')
        )
        return pages

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

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

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

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

会话的优势

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

使用 Camoufox 作为引擎

在 0.3.13 之前，这个 fetcher 曾使用一个定制版 Camoufox 作为底层引擎；后来由于多方面原因，被 patchright 取代。如果你发现 Camoufox 在你的设备上运行稳定、没有明显的高内存问题，并且你仍想继续使用它，也是可以的。

首先，如果你还没有安装过 Camoufox 库、浏览器以及 Firefox 所需的系统依赖，需要先执行：

pip install camoufox
playwright install-deps firefox
camoufox fetch

然后你可以继承 StealthySession 并按下面的方式设置：

from scrapling.fetchers import StealthySession
from playwright.sync_api import sync_playwright
from camoufox.utils import launch_options as generate_launch_options

class StealthySession(StealthySession):
    def start(self):
        """Create a browser for this instance and context."""
        if not self.playwright:
            self.playwright = sync_playwright().start()
            # Configure camoufox run options here
            launch_options = generate_launch_options(**{"headless": True, "user_data_dir": ''})
            # Here's an example, part of what we have been doing before v0.3.13
            launch_options = generate_launch_options(**{
                "geoip": False,
                "proxy": self._config.proxy,
                "headless": self._config.headless,
                "humanize": True if self._config.solve_cloudflare else False,  # Better enable humanize for Cloudflare, otherwise it's up to you
                "i_know_what_im_doing": True,  # To turn warnings off with the user configurations
                "allow_webgl": self._config.allow_webgl,
                "block_webrtc": self._config.block_webrtc,
                "os": None,
                "user_data_dir": self._config.user_data_dir,
                "firefox_user_prefs": {
                    # This is what enabling `enable_cache` does internally, so we do it from here instead
                    "browser.sessionhistory.max_entries": 10,
                    "browser.sessionhistory.max_total_viewers": -1,
                    "browser.cache.memory.enable": True,
                    "browser.cache.disk_cache_ssl": True,
                    "browser.cache.disk.smart_size.enabled": True,
                },
                # etc...
            })
            self.context = self.playwright.firefox.launch_persistent_context(**launch_options)
        else:
            raise RuntimeError("Session has been already started")

之后你就可以像以前一样正常使用它，甚至继续用它解决 Cloudflare 挑战：

with StealthySession(solve_cloudflare=True, headless=True) as session:
    page = session.fetch('https://sergiodemo.com/security/challenge/legacy-challenge')
    if page.css('#page-not-found-404'):
        print('Cloudflare challenge solved successfully!')

AsyncStealthySession 类也适用同样的思路，只是有一些细节差异：

from scrapling.fetchers import AsyncStealthySession
from playwright.async_api import async_playwright
from camoufox.utils import launch_options as generate_launch_options

class AsyncStealthySession(AsyncStealthySession):
    async def start(self):
        """Create a browser for this instance and context."""
        if not self.playwright:
            self.playwright = await async_playwright().start()
            # Configure camoufox run options here
            launch_options = generate_launch_options(**{"headless": True, "user_data_dir": ''})
            # or set the launch options as in the above example
            self.context = await self.playwright.firefox.launch_persistent_context(**launch_options)
        else:
            raise RuntimeError("Session has been already started")

async with AsyncStealthySession(solve_cloudflare=True, headless=True) as session:
    page = await session.fetch('https://sergiodemo.com/security/challenge/legacy-challenge')
    if page.css('#page-not-found-404'):
        print('Cloudflare challenge solved successfully!')

祝你使用顺利！:)

何时使用

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

• 需要绕过反机器人防护
• 需要更可靠的浏览器指纹
• 需要完整的 JavaScript 支持
• 希望自动获得更多隐匿特性
• 需要浏览器自动化
• 需要处理 Cloudflare 防护