动态页面爬取（Selenium/Puppeteer）

动态页面爬取原理

动态页面 vs 静态页面

静态页面：服务器端渲染完成后返回完整的 HTML 文档，页面内容在 HTML 源码中直接可见。爬虫只需发送 HTTP 请求获取 HTML 并解析即可提取数据。

动态页面：页面内容由 JavaScript 在浏览器端动态生成和渲染，HTML 源码中不包含最终展示的数据。爬虫需要执行 JavaScript 或直接捕获 API 接口才能获取数据。

CSR（Client-Side Rendering）与 SSR（Server-Side Rendering）区别

特性	CSR（客户端渲染）	SSR（服务端渲染）
HTML 内容	初始 HTML 只包含根节点和脚本引用	返回完整渲染后的 HTML
SEO 友好度	低，搜索引擎不易抓取	高，内容直接可见
首屏加载速度	较慢，需等待 JS 下载执行	较快，直接展示内容
爬虫难度	高，需要执行 JS 或抓取 API	低，可直接解析 HTML
常见框架	Vue CLI / Create React App	Next.js / Nuxt.js

JS 执行与页面渲染

动态页面依赖 JavaScript 执行来完成以下操作：

通过 fetch / XMLHttpRequest 异步加载数据
使用 DOM API 动态创建和修改页面元素
通过路由库实现前端路由切换
通过 WebSocket 接收实时推送数据

常见 SPA 框架页面特点

Vue.js 页面：使用 v-if / v-for 等指令动态控制 DOM；数据通过 data / vuex / pinia 管理；页面挂载点通常为 <div id="app">；异步请求使用 axios 或 fetch
React 页面：使用 JSX 语法，通过 useState / useEffect 管理状态和副作用；虚拟 DOM diff 算法更新真实 DOM；数据流单向；常见于中后台管理系统
Angular 页面：使用 TypeScript，依赖注入机制；zone.js 实现变更检测；模板语法和指令系统

渲染机制

浏览器渲染流程

HTML Parse -> CSSOM -> Render Tree -> Layout -> Paint -> Composite

HTML Parse：浏览器解析 HTML 文档生成 DOM（Document Object Model）树
CSSOM：解析 CSS 生成 CSSOM（CSS Object Model）树
Render Tree：将 DOM 树和 CSSOM 树合并生成 Render Tree
Layout（Reflow）：计算每个元素的位置和大小
Paint：将元素绘制到屏幕上
Composite：将分层合成的页面展示给用户

当页面中存在 JavaScript 动态修改 DOM 时，上述流程会部分或全部重走，引发回流（Reflow）和重绘（Repaint）。

异步数据加载

动态页面通常通过异步请求从后端 API 获取数据：

javascript

// 使用 fetch 加载数据
async function loadData() {
  const response = await fetch('/api/products?page=1');
  const data = await response.json();
  // 动态渲染到页面
  document.getElementById('product-list').innerHTML = renderProducts(data);
}

// 使用 XMLHttpRequest
const xhr = new XMLHttpRequest();
xhr.open('GET', '/api/products');
xhr.onload = function() {
  const data = JSON.parse(xhr.responseText);
  renderProducts(data);
};
xhr.send();

HTTP 请求分析

爬取动态页面时，分析网络请求是关键步骤：

XHR / Fetch 请求：打开浏览器 DevTools 的 Network 面板，筛选 XHR / Fetch 类型的请求，找到返回实际数据的 API 接口
请求头分析：关注 Authorization、Cookie、X-CSRF-Token 等鉴权头
请求体分析：POST 请求的 payload 格式（JSON / FormData）
响应数据格式：通常为 JSON，也可能为 XML 或 Protobuf

Network 面板分析

浏览器 DevTools 的 Network 面板提供以下功能：

查看所有网络请求的详细信息（URL、方法、状态码、请求头、响应体）
筛选特定类型的请求（XHR、JS、CSS、IMG、Media、Font、Doc、WS、Manifest）
搜索请求内容（通过关键词搜索所有请求的 URL 和响应体）
模拟网络条件（限速、离线）
记录请求时间线（Timing 面板：Queueing、DNS、TCP、TLS、Request、Response）

预加载数据提取

部分动态页面会将初始数据嵌入 HTML 中，避免首屏异步请求：

html

<!-- 通过 <script> 标签嵌入 JSON 数据 -->
<script>
  window.__INITIAL_STATE__ = {
    products: [...],
    user: {...}
  };
</script>

<!-- 通过 data-* 属性嵌入 -->
<div id="app" data-config='{"apiKey":"xxx"}'></div>

<!-- 通过隐藏的 pre 或 textarea 标签 -->
<pre id="server-data" style="display:none">
{"products":[...]}
</pre>

爬虫在解析 HTML 后，可以直接从这些预加载数据中提取初始内容，减少渲染等待时间。

Selenium 爬虫

Selenium 架构

WebDriver

Selenium WebDriver 是浏览器自动化控制的核心接口，通过标准化的协议（WebDriver Protocol / W3C WebDriver 规范）与浏览器通信。

爬虫脚本 -> Selenium Client Library -> WebDriver Protocol -> Browser Driver -> 浏览器

Selenium Client Library：Python / Java / C# / Ruby / JavaScript 等语言的客户端绑定
WebDriver Protocol：基于 HTTP 的 RESTful API，规范了浏览器操作的 JSON Wire Protocol 或 W3C WebDriver 标准
Browser Driver：各浏览器厂商实现的 WebDriver 服务，如 ChromeDriver、GeckoDriver（Firefox）、EdgeDriver

浏览器驱动

浏览器	驱动名称	下载地址
Chrome	ChromeDriver	https://chromedriver.chromium.org/
Firefox	GeckoDriver	https://github.com/mozilla/geckodriver
Edge	EdgeDriver	https://developer.microsoft.com/en-us/microsoft-edge/tools/webdriver/
Safari	safaridriver	内置于 macOS

Remote WebDriver

Remote WebDriver 允许在远程机器上执行浏览器操作，实现浏览器与爬虫脚本的分离部署：

python

from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities

# 连接到远程 Selenium Server
driver = webdriver.Remote(
    command_executor='http://localhost:4444/wd/hub',
    desired_capabilities=DesiredCapabilities.CHROME
)

typescript

import { Builder } from 'selenium-webdriver';

const driver = await new Builder()
  .usingServer('http://localhost:4444/wd/hub')
  .forBrowser('chrome')
  .build();

Grid 分布式

Selenium Grid 支持在多台机器上分布式运行浏览器，实现并发爬取：

Hub（中心节点）：接收测试请求并分发到各个 Node
Node（工作节点）：在各自机器上启动浏览器执行任务
MaxSession：每个 Node 可同时运行的浏览器实例数

bash

# 启动 Hub
java -jar selenium-server-standalone.jar -role hub

# 启动 Node 并注册到 Hub
java -jar selenium-server-standalone.jar -role node -hub http://localhost:4444/grid/register

元素定位

8 种定位方式

Selenium 提供了 8 种元素定位方法：

定位方式	Python By 常量	示例	说明
ID	`By.ID`	`find_element(By.ID, "username")`	按 id 属性定位，唯一且最快
Name	`By.NAME`	`find_element(By.NAME, "email")`	按 name 属性定位
Class Name	`By.CLASS_NAME`	`find_element(By.CLASS_NAME, "title")`	按 class 属性定位
Tag Name	`By.TAG_NAME`	`find_element(By.TAG_NAME, "div")`	按 HTML 标签名定位
CSS Selector	`By.CSS_SELECTOR`	`find_element(By.CSS_SELECTOR, ".list > .item:first-child")`	使用 CSS 选择器
XPath	`By.XPATH`	`find_element(By.XPATH, "//div[@class='content']//a")`	使用 XPath 语法
Link Text	`By.LINK_TEXT`	`find_element(By.LINK_TEXT, "点击查看详情")`	按链接完整文本定位
Partial Link Text	`By.PARTIAL_LINK_TEXT`	`find_element(By.PARTIAL_LINK_TEXT, "查看")`	按链接部分文本定位

findElement 与 findElements

python

from selenium import webdriver
from selenium.webdriver.common.by import By

driver = webdriver.Chrome()
driver.get('https://example.com/products')

# findElement: 返回第一个匹配元素，未找到则抛出 NoSuchElementException
first_product = driver.find_element(By.CSS_SELECTOR, '.product-item')

# findElements: 返回匹配元素列表，未找到则返回空列表
all_products = driver.find_elements(By.CSS_SELECTOR, '.product-item')
print(f'共找到 {len(all_products)} 个商品')

# XPath 定位示例
element = driver.find_element(By.XPATH, '//div[contains(@class, "price") and contains(text(), "￥")]')

# 相对定位：查找父元素下的子元素
parent = driver.find_element(By.ID, 'product-list')
children = parent.find_elements(By.TAG_NAME, 'li')

typescript

import { By, Builder } from 'selenium-webdriver';

const driver = await new Builder().forBrowser('chrome').build();
await driver.get('https://example.com/products');

// findElement
const firstProduct = await driver.findElement(By.css('.product-item'));

// findElements
const allProducts = await driver.findElements(By.css('.product-item'));
console.log(`共找到 ${allProducts.length} 个商品`);

// XPath
const element = await driver.findElement(By.xpath('//div[contains(text(), "￥")]'));

相对定位（Relative Locators）

Selenium 4 引入了相对定位 API，根据元素之间的位置关系查找元素：

python

from selenium.webdriver.support.relative_locator import locate_with

# near: 靠近某个元素
password_field = driver.find_element(By.ID, 'password')
confirm_field = driver.find_element(locate_with(By.TAG_NAME, 'input').near(password_field))

# above: 在某个元素上方
login_button = driver.find_element(By.ID, 'login-btn')
hint_text = driver.find_element(locate_with(By.TAG_NAME, 'span').above(login_button))

# below: 在某个元素下方
below_element = driver.find_element(locate_with(By.TAG_NAME, 'div').below(login_button))

# to_left_of / to_right_of: 左/右侧
left_element = driver.find_element(locate_with(By.ID, 'left').to_left_of(login_button))

等待策略

隐式等待（Implicitly Wait）

设置全局等待时间，在查找元素时如果元素未出现则等待指定时间：

python

driver = webdriver.Chrome()
driver.implicitly_wait(10)  # 全局等待 10 秒
# 后续所有 find_element 调用都会最多等待 10 秒
element = driver.find_element(By.ID, 'dynamic-content')

隐式等待的缺点：

全局生效，无法针对特定元素设置不同等待时间
只作用于元素查找，不处理元素状态（如是否可点击、可见）
与显式等待混用时可能产生意外行为

显式等待（Explicit Wait）

针对特定条件设置等待时间，更加精确可控：

python

from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

driver = webdriver.Chrome()
driver.get('https://example.com')

# 等待元素可见
element = WebDriverWait(driver, 10).until(
    EC.visibility_of_element_located((By.ID, 'dynamic-content'))
)

# 等待元素可点击
button = WebDriverWait(driver, 10).until(
    EC.element_to_be_clickable((By.CSS_SELECTOR, '.submit-btn'))
)

# 等待元素包含特定文本
WebDriverWait(driver, 10).until(
    EC.text_to_be_present_in_element((By.CLASS_NAME, 'status'), '加载完成')
)

# 等待元素消失（用于等待加载动画消失）
WebDriverWait(driver, 10).until(
    EC.invisibility_of_element_located((By.CLASS_NAME, 'spinner'))
)

ExpectedConditions 常用条件

条件	说明
`visibility_of_element_located`	元素在 DOM 中且可见
`element_to_be_clickable`	元素可见且可点击
`presence_of_element_located`	元素存在于 DOM 中
`invisibility_of_element_located`	元素不可见或不存在
`text_to_be_present_in_element`	元素包含指定文本
`alert_is_present`	弹出 Alert 对话框
`frame_to_be_available_and_switch_to_it`	可用帧且可切换
`element_selection_state_to_be`	元素选中状态

FluentWait

FluentWait 提供了更灵活的等待配置，可自定义轮询间隔和忽略特定异常：

python

from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import NoSuchElementException, StaleElementReferenceException
import time

# 自定义 FluentWait
wait = WebDriverWait(
    driver,
    timeout=30,                # 最大等待时间
    poll_frequency=0.5,        # 每 0.5 秒检查一次
    ignored_exceptions=[       # 忽略的异常
        NoSuchElementException,
        StaleElementReferenceException
    ]
)

element = wait.until(
    EC.presence_of_element_located((By.ID, 'lazy-loaded-content'))
)

自定义等待条件

当内置的 ExpectedConditions 不够用时，可以自定义等待条件：

python

class element_has_scrolled:
    """等待页面滚动到指定位置"""
    def __init__(self, target_y):
        self.target_y = target_y

    def __call__(self, driver):
        current_y = driver.execute_script('return window.scrollY')
        return current_y >= self.target_y

# 使用自定义条件
WebDriverWait(driver, 10).until(element_has_scrolled(1000))

typescript

class ElementHasScrolled {
  constructor(private targetY: number) {}

  async __call__(driver: any): Promise<boolean> {
    const currentY = await driver.executeScript('return window.scrollY');
    return currentY >= this.targetY;
  }
}

await driver.wait(new ElementHasScrolled(1000), 10000);

Selenium 实例

页面滚动

python

# 滚动到页面底部
driver.execute_script('window.scrollTo(0, document.body.scrollHeight)')

# 滚动到指定元素
element = driver.find_element(By.ID, 'target')
driver.execute_script('arguments[0].scrollIntoView(true)', element)

# 渐进式滚动（触发无限加载）
window_height = driver.execute_script('return document.body.scrollHeight')
for i in range(0, window_height, 300):
    driver.execute_script(f'window.scrollTo(0, {i})')
    time.sleep(0.3)

下拉加载

python

import time

def scroll_to_load_all(max_attempts=10):
    """模拟下拉加载更多内容"""
    previous_height = driver.execute_script('return document.body.scrollHeight')

    for attempt in range(max_attempts):
        # 滚动到底部
        driver.execute_script('window.scrollTo(0, document.body.scrollHeight)')
        time.sleep(2)  # 等待内容加载

        # 检查页面高度是否变化
        new_height = driver.execute_script('return document.body.scrollHeight')
        if new_height == previous_height:
            print(f'第 {attempt + 1} 次尝试后无更多内容加载')
            break
        previous_height = new_height

点击展开与翻页

python

# 点击展开更多详情
expand_buttons = driver.find_elements(By.CSS_SELECTOR, '.expand-btn')
for btn in expand_buttons:
    try:
        btn.click()
        time.sleep(0.5)
    except Exception as e:
        print(f'点击展开失败: {e}')

# 翻页操作
def paginate():
    current_page = 1
    while True:
        # 获取当前页数据
        items = driver.find_elements(By.CSS_SELECTOR, '.item')
        print(f'第 {current_page} 页, 共 {len(items)} 条')

        # 查找下一页按钮
        try:
            next_btn = driver.find_element(By.CSS_SELECTOR, '.pagination .next:not(.disabled)')
            next_btn.click()
            time.sleep(2)
            current_page += 1
        except:
            print('翻页结束')
            break

文件上传

python

from selenium.webdriver.common.by import By

# 方法1：直接通过 input[type=file] 上传
file_input = driver.find_element(By.CSS_SELECTOR, 'input[type="file"]')
file_input.send_keys('C:\\path\\to\\file.txt')

# 方法2：通过第三方上传组件（需先暴露文件输入框）
driver.execute_script('document.getElementById("file-input").style.display = "block"')
driver.find_element(By.ID, 'file-input').send_keys('C:\\path\\to\\image.jpg')

截屏

python

# 全页面截图
driver.save_screenshot('full_page.png')

# 元素截图
element = driver.find_element(By.CLASS_NAME, 'target-area')
element.screenshot('element.png')

# 设置窗口大小后截图
driver.set_window_size(1920, 1080)
driver.save_screenshot('desktop_view.png')

typescript

// 全页面截图
await driver.takeScreenshot().then(data => {
  require('fs').writeFileSync('screenshot.png', data, 'base64');
});

窗口切换与多标签

python

from selenium.webdriver.common.by import By

# 获取当前窗口句柄
main_window = driver.current_window_handle

# 点击打开新标签页的链接
driver.find_element(By.LINK_TEXT, '在新标签页中打开').click()

# 切换到新标签页
all_windows = driver.window_handles
for window in all_windows:
    if window != main_window:
        driver.switch_to.window(window)
        break

# 在新标签页中操作
print(driver.title)

# 关闭新标签页并切回主窗口
driver.close()
driver.switch_to.window(main_window)

Alert 对话框处理

python

from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# 等待 Alert 出现
alert = WebDriverWait(driver, 5).until(EC.alert_is_present())

# 获取 Alert 文本
alert_text = alert.text
print(f'Alert 内容: {alert_text}')

# 接受（确认）
alert.accept()

# 或取消
alert.dismiss()

# 如果是 Prompt，可输入文本
alert.send_keys('输入内容')
alert.accept()

Frame 切换

python

# 切换到 iframe
iframe = driver.find_element(By.CSS_SELECTOR, 'iframe#content')
driver.switch_to.frame(iframe)

# 在 iframe 内操作
inner_element = driver.find_element(By.ID, 'inside-iframe')
print(inner_element.text)

# 切回主文档
driver.switch_to.default_content()

# 切换到嵌套 iframe
driver.switch_to.frame('outer-frame')
driver.switch_to.frame('inner-frame')

Selenium Stealth 反检测

问题背景

Selenium 控制的浏览器会被检测出以下特征：

navigator.webdriver 属性为 true
chrome.runtime 存在且包含自动化相关标记
User-Agent 中包含 HeadlessChrome 标识
CDP（Chrome DevTools Protocol）连接痕迹
浏览器的指纹特征与正常用户不同

stealth.min.js

通过注入 JavaScript 修改浏览器环境变量：

python

from selenium import webdriver

driver = webdriver.Chrome()

# 读取 stealth.min.js 并注入
with open('stealth.min.js', 'r') as f:
    stealth_js = f.read()

driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
    'source': stealth_js
})

driver.get('https://example.com')

undetected-chromedriver

第三方库，自动绕过主流反爬检测：

python

import undetected_chromedriver as uc

# 自动处理 ChromeDriver 版本匹配和反检测
driver = uc.Chrome()
driver.get('https://example.com')

# 自定义选项
options = uc.ChromeOptions()
options.headless = True
driver = uc.Chrome(options=options)

修改 navigator.webdriver

python

from selenium import webdriver

driver = webdriver.Chrome()

# 在页面加载前注入脚本修改 navigator 属性
driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
    'source': '''
        Object.defineProperty(navigator, 'webdriver', {
            get: () => undefined
        });

        // 覆盖 navigator.plugins
        Object.defineProperty(navigator, 'plugins', {
            get: () => [1, 2, 3, 4, 5]
        });

        // 覆盖 navigator.languages
        Object.defineProperty(navigator, 'languages', {
            get: () => ['zh-CN', 'zh', 'en']
        });

        // 覆盖 chrome.runtime
        window.chrome = {
            runtime: {}
        };
    '''
})

driver.get('https://example.com')

移除 CDP 痕迹

python

from selenium import webdriver

options = webdriver.ChromeOptions()

# 禁用自动化扩展和提示
options.add_experimental_option('excludeSwitches', ['enable-automation'])
options.add_experimental_option('useAutomationExtension', False)

# 设置窗口大小和位置使行为更自然
options.add_argument('--window-size=1920,1080')
options.add_argument('--start-maximized')

driver = webdriver.Chrome(options=options)

# 移除 webdriver 属性
driver.execute_script('Object.defineProperty(navigator, "webdriver", {get: () => undefined})')

参数修改

python

from selenium import webdriver

options = webdriver.ChromeOptions()

# 禁用自动化标志
options.add_argument('--disable-blink-features=AutomationControlled')

# 禁用 GPU 加速和沙箱
options.add_argument('--disable-gpu')
options.add_argument('--no-sandbox')

# 禁用共享内存
options.add_argument('--disable-dev-shm-usage')

# 设置 User-Agent
options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36')

# 禁用自动化扩展
options.add_experimental_option('excludeSwitches', ['enable-automation'])
options.add_experimental_option('useAutomationExtension', False)

driver = webdriver.Chrome(options=options)

Puppeteer/Playwright 爬虫

Puppeteer 核心 API

Puppeteer 是一个 Node.js 库，通过 Chrome DevTools Protocol 控制 Chrome/Chromium 浏览器。

launch 与 browser 控制

typescript

import puppeteer from 'puppeteer';

// 启动浏览器
const browser = await puppeteer.launch({
  headless: false,           // 无头模式
  slowMo: 50,                // 操作延迟（毫秒），模拟人类操作速度
  defaultViewport: {
    width: 1920,
    height: 1080
  },
  args: [
    '--no-sandbox',
    '--disable-setuid-sandbox',
    '--disable-dev-shm-usage',
    '--disable-gpu'
  ]
});

// 创建新页面
const page = await browser.newPage();

// 关闭浏览器
await browser.close();

page.goto 与导航

typescript

// 导航到页面
await page.goto('https://example.com', {
  waitUntil: 'networkidle2',   // 等待网络空闲
  timeout: 30000                // 超时时间
});

// waitUntil 选项:
// - 'load': 触发 load 事件
// - 'domcontentloaded': DOM 解析完成
// - 'networkidle0': 500ms 内没有网络连接
// - 'networkidle2': 500ms 内网络连接数不超过 2

waitForSelector 与等待

typescript

// 等待元素出现
await page.waitForSelector('.product-item', {
  timeout: 10000,
  visible: true          // 等待元素可见
});

// 等待元素消失（如加载动画）
await page.waitForSelector('.spinner', { hidden: true });

// 等待网络空闲
await page.waitForNetworkIdle({ idleTime: 500 });

// 等待自定义条件
await page.waitForFunction(() => {
  return document.querySelectorAll('.item').length > 10;
});

evaluate 与 $$eval

typescript

// evaluate: 在浏览器上下文中执行代码
const title = await page.evaluate(() => {
  return document.title;
});

const pageData = await page.evaluate(() => {
  const items = document.querySelectorAll('.product');
  return Array.from(items).map(item => ({
    name: item.querySelector('.name')?.textContent?.trim(),
    price: item.querySelector('.price')?.textContent?.trim(),
    link: (item.querySelector('a') as HTMLAnchorElement)?.href
  }));
});

// $eval: 对单个元素执行操作
const firstItemName = await page.$eval('.product-item', (el) => {
  return el.textContent?.trim();
});

// $$eval: 对元素列表执行操作
const allPrices = await page.$$eval('.price', (elements) => {
  return elements.map(el => el.textContent?.trim());
});

click 与 type

typescript

// 点击操作
await page.click('.login-btn');
await page.click('#submit-btn', { delay: 100 });  // 模拟人类点击延迟

// 输入文本
await page.type('#username', 'admin', { delay: 50 });  // 模拟键盘输入速度
await page.type('#password', 'password123', { delay: 30 });

// 键盘操作
await page.keyboard.press('Enter');
await page.keyboard.press('Escape');
await page.keyboard.down('Control');
await page.keyboard.press('A');
await page.keyboard.up('Control');

// 下拉选择
await page.select('select#country', 'CN');

screenshot 与 pdf

typescript

// 全页面截图
await page.screenshot({
  path: 'screenshot.png',
  fullPage: true
});

// 元素截图
const element = await page.$('.target-area');
if (element) {
  await element.screenshot({ path: 'element.png' });
}

// 生成 PDF
await page.pdf({
  path: 'page.pdf',
  format: 'A4',
  printBackground: true,
  margin: { top: '20px', right: '20px', bottom: '20px', left: '20px' }
});

请求拦截（request interception）

typescript

// 启用请求拦截
await page.setRequestInterception(true);

page.on('request', (request) => {
  const url = request.url();

  // 拦截并阻止不必要的资源加载
  if (request.resourceType() === 'image' ||
      request.resourceType() === 'font' ||
      request.resourceType() === 'media') {
    request.abort();
  } else {
    request.continue();
  }
});

page.on('response', (response) => {
  // 分析响应
  const url = response.url();
  const status = response.status();

  // 捕获 API 响应数据
  if (url.includes('/api/products')) {
    response.json().then(data => {
      console.log('API 响应数据:', data);
    }).catch(() => {});
  }
});

Playwright

Playwright 是微软开发的跨浏览器自动化测试库，支持 Chromium、Firefox 和 WebKit。

多浏览器支持

typescript

import { chromium, firefox, webkit } from 'playwright';

// Chromium
const browser = await chromium.launch({ headless: true });

// Firefox
const browser = await firefox.launch({ headless: true });

// WebKit (Safari)
const browser = await webkit.launch({ headless: true });

auto-wait（自动等待）

Playwright 的核心优势之一是内置自动等待机制：

typescript

const page = await browser.newPage();
await page.goto('https://example.com');

// 这些操作都会自动等待元素就绪
await page.click('.submit-btn');      // 自动等待元素可见、稳定、可点击
await page.fill('#username', 'admin'); // 自动等待元素可见
await page.check('#agree');           // 自动等待元素可见
await page.selectOption('#country', 'CN'); // 自动等待元素可见

// 不包含 auto-wait 的显式操作
await page.dispatchEvent('.btn', 'click');  // 直接派发事件，不等待

网络拦截（route）

typescript

const page = await browser.newPage();

// 拦截网络请求
await page.route('**/*.{png,jpg,jpeg,gif,svg}', route => route.abort());
await page.route('**/*.css', route => route.abort());
await page.route('**/*.woff2', route => route.abort());

// 修改 API 响应（Mock）
await page.route('**/api/products', async route => {
  const response = await route.fetch();
  const json = await response.json();
  // 修改响应数据
  json.products = json.products.slice(0, 5);
  await route.fulfill({ response, json });
});

// 获取请求详情
page.on('request', request => {
  console.log(`请求: ${request.method()} ${request.url()}`);
});

page.on('response', response => {
  console.log(`响应: ${response.status()} ${response.url()}`);
});

API Request Context

Playwright 允许在不启动浏览器的情况下直接发送 HTTP 请求：

typescript

import { request } from 'playwright';

// 创建 API Request Context
const apiContext = await request.newContext({
  baseURL: 'https://api.example.com',
  extraHTTPHeaders: {
    'Authorization': 'Bearer token123',
    'User-Agent': 'Mozilla/5.0 ...'
  }
});

// 发送 GET 请求
const response = await apiContext.get('/products?page=1');
const data = await response.json();
console.log(data);

// 发送 POST 请求
const postResponse = await apiContext.post('/login', {
  data: { username: 'admin', password: 'pass' }
});

// 发送带表单数据的 POST 请求
const formResponse = await apiContext.post('/upload', {
  multipart: {
    file: { name: 'test.txt', mimeType: 'text/plain', buffer: Buffer.from('content') }
  }
});

await apiContext.dispose();

Puppeteer Stealth

puppeteer-extra + puppeteer-extra-plugin-stealth

typescript

import puppeteer from 'puppeteer-extra';
import StealthPlugin from 'puppeteer-extra-plugin-stealth';

// 使用 Stealth 插件
puppeteer.use(StealthPlugin());

// 启动 Stealth 增强的浏览器
const browser = await puppeteer.launch({
  headless: true,
  args: ['--no-sandbox']
});

const page = await browser.newPage();
await page.goto('https://example.com');

// 检查检测结果（在检测网站如 https://bot.sannysoft.com 上验证）
const result = await page.evaluate(() => {
  return {
    webdriver: navigator.webdriver,
    plugins: navigator.plugins.length,
    languages: navigator.languages,
    chrome: !!window.chrome
  };
});
console.log(result);

修改指纹：WebGL / WebRTC / Battery / OS 平台伪装

typescript

import puppeteer from 'puppeteer';

const browser = await puppeteer.launch({ headless: true });
const page = await browser.newPage();

// 设置 Viewport
await page.setViewport({
  width: 1920,
  height: 1080,
  deviceScaleFactor: 1
});

// 设置 User-Agent
await page.setUserAgent(
  'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
);

// 注入修改指纹的脚本
await page.evaluateOnNewDocument(() => {
  // 修改 WebGL 指纹
  const getParameter = WebGLRenderingContext.prototype.getParameter;
  WebGLRenderingContext.prototype.getParameter = function(parameter: number) {
    if (parameter === 37445) {
      return 'Intel Inc.';  // 伪造渲染器厂商
    }
    if (parameter === 37446) {
      return 'Intel Iris OpenGL Engine';  // 伪造渲染器名称
    }
    return getParameter.call(this, parameter);
  };

  // 修改 WebRTC 指纹
  Object.defineProperty(navigator, 'connection', {
    get: () => ({
      effectiveType: '4g',
      rtt: 50,
      downlink: 10,
      saveData: false
    })
  });

  // 修改电池 API
  if ('getBattery' in navigator) {
    (navigator as any).getBattery = () => {
      return Promise.resolve({
        charging: true,
        level: 0.85,
        chargingTime: 0,
        dischargingTime: Infinity
      });
    };
  }

  // 修改平台信息
  Object.defineProperty(navigator, 'platform', {
    get: () => 'Win32'
  });

  Object.defineProperty(navigator, 'hardwareConcurrency', {
    get: () => 8
  });

  Object.defineProperty(navigator, 'deviceMemory', {
    get: () => 8
  });
});

Playwright vs Puppeteer 对比

特性	Puppeteer	Playwright
开发方	Google	Microsoft
浏览器支持	Chrome/Chromium	Chromium + Firefox + WebKit
语言支持	JavaScript/TypeScript	JS/TS + Python + Java + .NET
自动等待	需手动调用 waitForSelector	内置 auto-wait 机制
网络拦截	request interception	route API，更简洁
API Context	无	内置，无需浏览器即可发请求
移动端模拟	手动配置	内置 device descriptors
Selector 引擎	仅 CSS/XPath	CSS + XPath + 文本选择器
Browser Context	无	支持隔离的 Context
Trace Viewer	无	内置录制和回放
测试框架集成	Jest 生态	内置 Test Runner
社区生态	插件丰富（stealth 等）	较新但增长迅速
性能	轻量	更优的并行和资源管理

Playwright Browser Context 示例

typescript

import { chromium } from 'playwright';

const browser = await chromium.launch();

// 创建隔离的 Browser Context（每个 Context 有独立的 cookies/localStorage）
const context1 = await browser.newContext({
  viewport: { width: 1920, height: 1080 },
  userAgent: 'Mozilla/5.0 ...',
  locale: 'zh-CN',
  timezoneId: 'Asia/Shanghai',
  geolocation: { latitude: 39.9042, longitude: 116.4074 },
  permissions: ['geolocation']
});

const context2 = await browser.newContext({
  viewport: { width: 375, height: 812 },   // 模拟移动端
  userAgent: 'Mozilla/5.0 (iPhone ...)',
  isMobile: true,
  hasTouch: true
});

const page1 = await context1.newPage();
const page2 = await context2.newPage();

await page1.goto('https://example.com');
await page2.goto('https://example.com');

动态页面数据处理

API 接口直接抓取

Network 拦截捕获 API

通过浏览器自动化工具拦截网络请求，直接从 API 接口获取结构化数据，而非解析 DOM：

python

from selenium import webdriver
from selenium.webdriver.common.by import By
import json

driver = webdriver.Chrome()
api_responses = []

# 通过 performance log 获取网络请求
driver.get('https://example.com')

# 方法1: 通过 performance.log
logs = driver.execute_script("""
    var entries = performance.getEntriesByType('resource');
    return entries.filter(e => e.initiatorType == 'fetch' || e.initiatorType == 'xmlhttprequest')
                  .map(e => ({name: e.name, type: e.initiatorType}));
""")

for log in logs:
    print(f'API 请求: {log["name"]}')

# 方法2: 通过 DevTools 捕获响应体
driver.execute_cdp_cmd('Network.enable', {})
driver.execute_cdp_cmd('Network.setCacheDisabled', {'cacheDisabled': True})

driver.get('https://example.com')

# 获取 Network 日志
network_logs = driver.execute_script("""
    var resources = performance.getEntriesByType('resource');
    return JSON.stringify(resources.filter(r => r.initiatorType === 'fetch'));
""")

typescript

import puppeteer from 'puppeteer';

const browser = await puppeteer.launch();
const page = await browser.newPage();

const capturedData: any[] = [];

// 拦截 XHR/Fetch 请求的响应
page.on('response', async (response) => {
  const url = response.url();

  // 只捕获目标 API
  if (url.includes('/api/')) {
    try {
      const contentType = response.headers()['content-type'] || '';
      if (contentType.includes('json')) {
        const data = await response.json();
        capturedData.push({ url, data });
        console.log(`捕获 API: ${url}`);
      }
    } catch (e) {
      // 忽略解析错误
    }
  }
});

await page.goto('https://example.com', { waitUntil: 'networkidle2' });
console.log(`共捕获 ${capturedData.length} 个 API 响应`);

抓包工具 Charles / Fiddler

抓包工具作为代理服务器，拦截所有 HTTP/HTTPS 流量：

Charles：设置 SSL Proxying -> 安装根证书 -> 设置系统代理为 127.0.0.1:8888 -> 捕获 API 请求 -> Export 为 Har 格式
Fiddler：安装根证书 -> 设置代理端口 8888 -> 使用 AutoResponder 修改响应 -> 使用 Composer 重放请求

python

# 通过抓包工具导出的 HAR 文件解析 API
import json

with open('capture.har', 'r') as f:
    har = json.load(f)

api_entries = []
for entry in har['log']['entries']:
    url = entry['request']['url']
    if '/api/' in url:
        response_body = entry['response']['content']['text']
        api_entries.append({
            'url': url,
            'method': entry['request']['method'],
            'status': entry['response']['status'],
            'data': json.loads(response_body) if response_body else None
        })

print(f'从 HAR 文件中提取 {len(api_entries)} 个 API 请求')

直接调用 API 替代渲染

分析出 API 接口后，可直接用 HTTP 客户端调用，无需启动浏览器：

python

import requests

# 分析 Network 面板后找到的 API
api_url = 'https://api.example.com/v1/products'
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'X-Requested-With': 'XMLHttpRequest',
    'Referer': 'https://example.com/products',
    'Cookie': 'session_id=abc123; token=xyz789'
}

params = {
    'page': 1,
    'pageSize': 20,
    'sort': 'price_asc'
}

response = requests.get(api_url, headers=headers, params=params)
data = response.json()

# 直接处理结构化数据
for product in data['products']:
    print(f"{product['name']} - ￥{product['price']}")

typescript

import axios from 'axios';

const api = axios.create({
  baseURL: 'https://api.example.com',
  headers: {
    'User-Agent': 'Mozilla/5.0 ...',
    'X-Requested-With': 'XMLHttpRequest',
    'Referer': 'https://example.com/products'
  }
});

async function fetchProducts(page: number) {
  const response = await api.get('/v1/products', {
    params: { page, pageSize: 20 }
  });
  return response.data;
}

提取数据清洗

HTML 结构不稳定与 XPath 容错

动态页面的 DOM 结构可能随版本变化，需要使用容错策略：

python

from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException
import re

def safe_extract(driver, selectors, attribute=None):
    """多级选择器容错提取"""
    for selector in selectors:
        try:
            elements = driver.find_elements(By.CSS_SELECTOR, selector)
            if elements:
                if attribute:
                    return [el.get_attribute(attribute) for el in elements]
                else:
                    return [el.text.strip() for el in elements]
        except:
            continue
    return []

# 使用多级选择器
prices = safe_extract(driver, [
    '.product-price .current',
    '.price-wrapper .price',
    '[class*="price"] .value',
    '.product-info strong:nth-child(2)'
])

# XPath 容错提取
def xpath_fallback_extract(driver, xpath_list):
    for xpath in xpath_list:
        try:
            element = driver.find_element(By.XPATH, xpath)
            if element and element.text:
                return element.text.strip()
        except:
            continue
    return ''

正则表达式辅助提取

python

import re

def extract_price(text):
    """从文本中提取价格"""
    patterns = [
        r'￥(\d+\.?\d*)',
        r'¥(\d+\.?\d*)',
        r'(\d+\.\d{2})',
        r'price[:\s]*(\d+)'
    ]
    for pattern in patterns:
        match = re.search(pattern, text)
        if match:
            return float(match.group(1))
    return None

def extract_numbers(text):
    """提取数字"""
    numbers = re.findall(r'\d+', text)
    return [int(n) for n in numbers] if numbers else []

多级选择器与解析失败兜底

python

from bs4 import BeautifulSoup

def extract_product_data(driver):
    """多级选择器 + 兜底策略"""
    product_data = {}

    # 主选择器
    try:
        name_el = driver.find_element(By.CSS_SELECTOR, '.product-name')
        product_data['name'] = name_el.text.strip()
    except:
        # 备选选择器
        try:
            name_el = driver.find_element(By.CSS_SELECTOR, '[class*="title"] h1')
            product_data['name'] = name_el.text.strip()
        except:
            # 终极兜底：获取页面标题
            product_data['name'] = driver.title

    # 价格提取兜底
    try:
        price = driver.find_element(By.CSS_SELECTOR, '.price').text
        product_data['price'] = float(re.search(r'[\d.]+', price).group())
    except:
        # 从整个页面内容中搜索价格
        body_text = driver.find_element(By.TAG_NAME, 'body').text
        price_match = re.search(r'￥?(\d+\.\d{2})', body_text)
        if price_match:
            product_data['price'] = float(price_match.group(1))
        else:
            product_data['price'] = None

    return product_data

性能优化

无头模式

python

from selenium import webdriver

options = webdriver.ChromeOptions()
options.add_argument('--headless')  # 不显示浏览器窗口
options.add_argument('--disable-gpu')

driver = webdriver.Chrome(options=options)

typescript

import puppeteer from 'puppeteer';

const browser = await puppeteer.launch({ headless: true });

禁用资源加载

python

from selenium import webdriver

options = webdriver.ChromeOptions()

# 禁用图片加载
prefs = {
    'profile.default_content_setting_values': {
        'images': 2,         # 禁用图片
        'javascript': 1,     # 启用 JS（必须，否则无法渲染）
        'stylesheets': 2,    # 禁用 CSS（谨慎，可能影响页面结构）
    }
}
options.add_experimental_option('prefs', prefs)
driver = webdriver.Chrome(options=options)

typescript

import puppeteer from 'puppeteer';

const browser = await puppeteer.launch();
const page = await browser.newPage();

// 拦截并禁用不必要的资源
await page.setRequestInterception(true);
page.on('request', (request) => {
  const type = request.resourceType();
  if (type === 'image' || type === 'font' || type === 'media') {
    request.abort();
  } else {
    request.continue();
  }
});

浏览器复用

避免频繁启动/关闭浏览器，采用连接池复用机制：

python

from selenium import webdriver
from queue import Queue
import threading

class BrowserPool:
    """浏览器实例连接池"""

    def __init__(self, max_browsers=3):
        self.max_browsers = max_browsers
        self.pool = Queue(maxsize=max_browsers)
        self._init_pool()

    def _init_pool(self):
        for _ in range(self.max_browsers):
            driver = self._create_driver()
            self.pool.put(driver)

    def _create_driver(self):
        options = webdriver.ChromeOptions()
        options.add_argument('--headless')
        return webdriver.Chrome(options=options)

    def acquire(self):
        return self.pool.get()

    def release(self, driver):
        self.pool.put(driver)

    def close_all(self):
        while not self.pool.empty():
            driver = self.pool.get()
            driver.quit()

# 使用连接池
pool = BrowserPool(max_browsers=5)
driver = pool.acquire()
try:
    driver.get('https://example.com')
    # 执行爬取操作
finally:
    pool.release(driver)

并发与连接池

python

from concurrent.futures import ThreadPoolExecutor, as_completed
from selenium import webdriver
import threading

local_storage = threading.local()

def get_driver():
    """线程级浏览器实例"""
    if not hasattr(local_storage, 'driver'):
        options = webdriver.ChromeOptions()
        options.add_argument('--headless')
        options.add_argument('--disable-gpu')
        local_storage.driver = webdriver.Chrome(options=options)
    return local_storage.driver

def scrape_page(url):
    driver = get_driver()
    driver.get(url)
    # 执行爬取逻辑
    items = driver.find_elements(By.CSS_SELECTOR, '.item')
    return [item.text for item in items]

# 并发爬取
urls = [f'https://example.com/page/{i}' for i in range(1, 11)]

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(scrape_page, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            print(f'{url}: 获取到 {len(data)} 条数据')
        except Exception as e:
            print(f'{url} 失败: {e}')

反检测综合策略

综合使用多种技术降低被识别为爬虫的概率：

python

from selenium import webdriver
from selenium.webdriver.common.by import By
import time
import random

def create_stealth_driver():
    """创建具备反检测能力的浏览器实例"""
    options = webdriver.ChromeOptions()

    # 基础设置
    options.add_argument('--disable-blink-features=AutomationControlled')
    options.add_experimental_option('excludeSwitches', ['enable-automation'])
    options.add_experimental_option('useAutomationExtension', False)

    # 窗口设置
    options.add_argument('--window-size=1920,1080')
    options.add_argument('--start-maximized')

    # 禁用无用功能
    options.add_argument('--disable-gpu')
    options.add_argument('--no-sandbox')

    # 随机延迟范围
    min_delay = 1.0
    max_delay = 3.0

    driver = webdriver.Chrome(options=options)

    # 执行反检测脚本
    driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
        'source': f'''
            // 修改 navigator.webdriver
            Object.defineProperty(navigator, 'webdriver', {{
                get: () => undefined
            }});

            // 设置 User-Agent
            Object.defineProperty(navigator, 'userAgent', {{
                get: () => navigator.userAgent.replace('Headless', '')
            }});

            // 设置语言
            Object.defineProperty(navigator, 'languages', {{
                get: () => ['zh-CN', 'zh']
            }});

            // 设置平台
            Object.defineProperty(navigator, 'platform', {{
                get: () => 'Win32'
            }});

            // 修改 WebGL 指纹
            const getParameter = WebGLRenderingContext.prototype.getParameter;
            WebGLRenderingContext.prototype.getParameter = function(p) {{
                if (p === 37445) return 'Intel Inc.';
                if (p === 37446) return 'Intel(R) UHD Graphics';
                return getParameter(p);
            }};

            // 添加 Chrome 运行时对象
            window.chrome = {{
                runtime: {{}},
                loadTimes: function() {{}},
                csi: function() {{}},
                app: {{}}
            }};
        '''
    })

    return driver

def human_like_delay():
    """模拟人类操作延迟"""
    time.sleep(random.uniform(0.5, 2.0))

def human_like_scroll(driver, target_y=None):
    """模拟人类滚动行为"""
    if target_y is None:
        target_y = random.randint(200, 800)

    current = 0
    while current < target_y:
        step = random.randint(50, 150)
        current += step
        if current > target_y:
            current = target_y
        driver.execute_script(f'window.scrollTo(0, {current})')
        time.sleep(random.uniform(0.1, 0.3))

typescript

import puppeteer from 'puppeteer';

async function createStealthBrowser() {
  const browser = await puppeteer.launch({
    headless: true,
    args: [
      '--no-sandbox',
      '--disable-setuid-sandbox',
      '--disable-dev-shm-usage',
      '--disable-accelerated-2d-canvas',
      '--disable-gpu',
      '--window-size=1920,1080'
    ]
  });

  const page = await browser.newPage();

  // 设置 Viewport
  await page.setViewport({
    width: 1920,
    height: 1080,
    deviceScaleFactor: 1
  });

  // 设置 User-Agent
  await page.setUserAgent(
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
  );

  // 设置 Geolocation
  await page.setGeolocation({
    latitude: 39.9042,
    longitude: 116.4074
  });

  // 设置 Timezone
  await page.emulateTimezone('Asia/Shanghai');

  // 设置额外请求头
  await page.setExtraHTTPHeaders({
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    'Accept-Encoding': 'gzip, deflate, br',
    'Accept': 'text/html,application/json,*/*',
    'Connection': 'keep-alive'
  });

  // 注入反检测脚本
  await page.evaluateOnNewDocument(() => {
    // 修改 WebDriver
    Object.defineProperty(navigator, 'webdriver', {
      get: () => undefined
    });

    // 修改 Languages
    Object.defineProperty(navigator, 'languages', {
      get: () => ['zh-CN', 'zh', 'en']
    });

    // 修改 Platform
    Object.defineProperty(navigator, 'platform', {
      get: () => 'Win32'
    });

    // 修改 Hardware Concurrency
    Object.defineProperty(navigator, 'hardwareConcurrency', {
      get: () => 8
    });

    // 修改 Device Memory
    Object.defineProperty(navigator, 'deviceMemory', {
      get: () => 8
    });

    // 模拟 Chrome
    (window as any).chrome = {
      runtime: {},
      loadTimes: () => {},
      csi: () => {},
      app: {}
    };
  });

  return { browser, page };
}

监控与异常处理

爬虫稳定性

超时设置

python

from selenium import webdriver
from selenium.common.exceptions import TimeoutException

driver = webdriver.Chrome()

# 设置页面加载超时
driver.set_page_load_timeout(30)

# 设置脚本执行超时
driver.set_script_timeout(30)

# 处理超时异常
try:
    driver.get('https://example.com')
except TimeoutException:
    print('页面加载超时，执行兜底处理')
    driver.execute_script('window.stop()')  # 停止页面加载

typescript

import puppeteer from 'puppeteer';

const browser = await puppeteer.launch();
const page = await browser.newPage();

// 设置导航超时
page.setDefaultNavigationTimeout(30000);

// 设置等待选择器超时
page.setDefaultTimeout(10000);

// 处理超时
try {
  await page.goto('https://example.com', { timeout: 30000 });
} catch (error) {
  console.error('导航超时:', error);
  // 执行兜底处理
}

页面崩溃检测

python

from selenium import webdriver
from selenium.webdriver.common.by import By
import logging

driver = webdriver.Chrome()

def check_page_crashed():
    """检测页面是否崩溃"""
    try:
        driver.title  # 尝试简单操作检测页面状态
        return False
    except Exception:
        return True

def safe_get(url, retries=3):
    """带崩溃检测的页面加载"""
    for attempt in range(retries):
        try:
            driver.get(url)
            if check_page_crashed():
                raise Exception('页面崩溃')
            return True
        except Exception as e:
            logging.warning(f'第 {attempt + 1} 次访问 {url} 失败: {e}')
            time.sleep(2)
    return False

浏览器进程重启

python

import psutil
import os

def restart_browser_if_crashed(driver):
    """检测浏览器进程并重启"""
    try:
        driver.title
        return driver
    except:
        print('浏览器进程异常，准备重启')
        # 清理残留进程
        for proc in psutil.process_iter(['pid', 'name']):
            if proc.info['name'] and 'chrome' in proc.info['name'].lower():
                proc.kill()

        # 重新创建 driver
        options = webdriver.ChromeOptions()
        options.add_argument('--headless')
        driver = webdriver.Chrome(options=options)
        return driver

重试机制

python

import time
from functools import wraps

def retry(max_retries=3, delay=2, backoff=2):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_delay = delay
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    print(f'重试 {attempt + 1}/{max_retries}: {e}')
                    time.sleep(current_delay)
                    current_delay *= backoff
            return None
        return wrapper
    return decorator

@retry(max_retries=5, delay=1, backoff=2)
def scrape_page(url):
    driver = webdriver.Chrome()
    driver.get(url)
    data = driver.find_element(By.ID, 'data').text
    driver.quit()
    return data

代理切换

python

import requests
from itertools import cycle

class ProxyManager:
    """代理管理器"""

    def __init__(self, proxy_list):
        self.proxy_pool = cycle(proxy_list)
        self.current_proxy = None
        self.failed_count = 0

    def get_proxy(self):
        return next(self.proxy_pool)

    def test_proxy(self, proxy):
        """测试代理是否可用"""
        try:
            response = requests.get(
                'https://httpbin.org/ip',
                proxies={'http': proxy, 'https': proxy},
                timeout=5
            )
            return response.status_code == 200
        except:
            return False

    def switch_proxy(self):
        """切换到下一个可用的代理"""
        for _ in range(len(self.proxy_pool)):
            proxy = self.get_proxy()
            if self.test_proxy(proxy):
                self.current_proxy = proxy
                self.failed_count = 0
                return proxy
        return None

# 使用代理切换
proxy_manager = ProxyManager([
    'http://proxy1:8080',
    'http://proxy2:8080',
    'http://proxy3:8080'
])

def get_driver_with_proxy():
    proxy = proxy_manager.switch_proxy()
    options = webdriver.ChromeOptions()
    options.add_argument(f'--proxy-server={proxy}')
    return webdriver.Chrome(options=options)

异常场景

元素未找到

python

from selenium.common.exceptions import NoSuchElementException, StaleElementReferenceException

def safe_find_element(driver, by, value, timeout=10):
    """安全的元素查找，包含多种异常处理"""
    from selenium.webdriver.support.ui import WebDriverWait
    from selenium.webdriver.support import expected_conditions as EC

    try:
        element = WebDriverWait(driver, timeout).until(
            EC.presence_of_element_located((by, value))
        )
        return element
    except TimeoutException:
        logging.warning(f'元素未找到: {by}={value}')
        return None
    except StaleElementReferenceException:
        logging.warning(f'元素已过期: {by}={value}')
        # 重新查找
        return safe_find_element(driver, by, value, timeout)

页面加载超时

python

from selenium.common.exceptions import TimeoutException

def scrape_with_timeout(driver, url, timeout=15):
    """带超时控制的页面爬取"""
    driver.set_page_load_timeout(timeout)
    try:
        driver.get(url)
        return True
    except TimeoutException:
        # 超时后页面可能已部分加载，尝试获取已加载的内容
        logging.warning(f'页面 {url} 加载超时，尝试获取已加载内容')
        driver.execute_script('window.stop()')
        return True  # 继续处理已加载的部分
    except Exception as e:
        logging.error(f'页面 {url} 加载失败: {e}')
        return False

验证码出现

python

def detect_captcha(driver):
    """检测验证码"""
    captcha_keywords = ['captcha', '验证码', 'recaptcha', 'verify', 'security-check']
    page_source = driver.page_source.lower()
    for keyword in captcha_keywords:
        if keyword in page_source:
            return True
    return False

def handle_captcha(driver):
    """验证码处理策略"""
    if detect_captcha(driver):
        logging.warning('检测到验证码')

        # 策略1: 切换代理 IP
        proxy_manager.switch_proxy()

        # 策略2: 等待一段时间后重试
        time.sleep(60)

        # 策略3: 使用第三方打码服务（如 2captcha）
        # captcha_solver = TwoCaptchaSolver('api_key')
        # captcha_element = driver.find_element(By.CSS_SELECTOR, '.captcha-image')
        # result = captcha_solver.solve(captcha_element)

        return False
    return True

IP 封锁

python

def check_ip_blocked(driver):
    """检测是否被封锁"""
    blocked_indicators = [
        'access denied',
        'blocked',
        'too many requests',
        '429',
        'captcha',
        'please wait',
        'access limited'
    ]
    page_text = driver.find_element(By.TAG_NAME, 'body').text.lower()
    for indicator in blocked_indicators:
        if indicator in page_text:
            return True
    return False

def scrape_with_proxy_rotation(driver, url, proxy_manager):
    """带代理轮换的爬取"""
    max_retries = 5
    for attempt in range(max_retries):
        driver.get(url)
        if check_ip_blocked(driver):
            logging.warning(f'IP 被封锁，尝试切换代理 (第 {attempt + 1} 次)')
            new_proxy = proxy_manager.switch_proxy()
            if new_proxy:
                # 重建带新代理的 driver
                driver.quit()
                options = webdriver.ChromeOptions()
                options.add_argument(f'--proxy-server={new_proxy}')
                driver = webdriver.Chrome(options=options)
            else:
                logging.error('无可用代理')
                return None
        else:
            return driver
    return None

浏览器崩溃与内存泄漏

python

import psutil
import os

def monitor_memory(threshold_mb=1024):
    """监控浏览器内存使用"""
    for proc in psutil.process_iter(['pid', 'name', 'memory_info']):
        try:
            if 'chrome' in proc.info['name'].lower():
                memory_mb = proc.info['memory_info'].rss / 1024 / 1024
                if memory_mb > threshold_mb:
                    logging.warning(f'Chrome 进程 {proc.info["pid"]} 内存占用 {memory_mb:.0f}MB，超过阈值')
                    return True
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return False

def restart_driver_if_needed(driver):
    """内存超限时重启浏览器"""
    if monitor_memory():
        logging.info('内存超限，重启浏览器')
        driver.quit()
        options = webdriver.ChromeOptions()
        options.add_argument('--headless')
        # 添加 --disable-dev-shm-usage 减少内存泄漏
        options.add_argument('--disable-dev-shm-usage')
        options.add_argument('--memory-pressure-off')
        driver = webdriver.Chrome(options=options)
    return driver

日志和监控

请求日志

python

import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('scraper.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

class ScrapeLogger:
    """爬虫日志记录器"""

    def __init__(self):
        self.stats = {
            'total_requests': 0,
            'successful': 0,
            'failed': 0,
            'start_time': datetime.now()
        }

    def log_request(self, url, status, duration, error=None):
        self.stats['total_requests'] += 1
        if status == 'success':
            self.stats['successful'] += 1
            logging.info(f'成功 [{duration:.2f}s] {url}')
        else:
            self.stats['failed'] += 1
            logging.error(f'失败 [{duration:.2f}s] {url} - {error}')

    def print_summary(self):
        elapsed = (datetime.now() - self.stats['start_time']).total_seconds()
        success_rate = self.stats['successful'] / max(self.stats['total_requests'], 1) * 100
        logging.info(f'=== 爬取统计 ===')
        logging.info(f'总请求: {self.stats["total_requests"]}')
        logging.info(f'成功: {self.stats["successful"]}')
        logging.info(f'失败: {self.stats["failed"]}')
        logging.info(f'成功率: {success_rate:.1f}%')
        logging.info(f'耗时: {elapsed:.0f}s')
        logging.info(f'================')

成功率统计

python

import json
from collections import defaultdict

class SuccessRateMonitor:
    """成功率监控"""

    def __init__(self, window_size=100):
        self.window_size = window_size
        self.results = []
        self.domain_stats = defaultdict(lambda: {'success': 0, 'failed': 0})

    def record(self, url, success, error=None):
        self.results.append({
            'url': url,
            'success': success,
            'error': error,
            'timestamp': datetime.now().isoformat()
        })

        # 按域名统计
        from urllib.parse import urlparse
        domain = urlparse(url).netloc
        if success:
            self.domain_stats[domain]['success'] += 1
        else:
            self.domain_stats[domain]['failed'] += 1

        # 只保留最近 window_size 条
        if len(self.results) > self.window_size:
            self.results.pop(0)

    def get_success_rate(self):
        if not self.results:
            return 1.0
        return sum(1 for r in self.results if r['success']) / len(self.results)

    def should_continue(self, threshold=0.8):
        """根据成功率判断是否继续爬取"""
        rate = self.get_success_rate()
        if rate < threshold:
            logging.warning(f'成功率 {rate:.1%} 低于阈值 {threshold:.0%}')
            return False
        return True

    def get_domain_health(self):
        """获取各域名的健康状况"""
        health = {}
        for domain, stats in self.domain_stats.items():
            total = stats['success'] + stats['failed']
            health[domain] = {
                'success_rate': stats['success'] / total if total > 0 else 1.0,
                'total': total
            }
        return health

数据完整性校验

python

class DataValidator:
    """数据完整性校验"""

    def __init__(self, schema):
        self.schema = schema  # 期望的数据结构
        self.errors = []

    def validate_item(self, item):
        """校验单个数据项"""
        missing_fields = []
        type_mismatches = []

        for field, field_type in self.schema.items():
            if field not in item:
                missing_fields.append(field)
            elif item[field] is not None and not isinstance(item[field], field_type):
                type_mismatches.append(f'{field}: 期望 {field_type.__name__}, 实际 {type(item[field]).__name__}')

        if missing_fields or type_mismatches:
            self.errors.append({
                'item': item,
                'missing_fields': missing_fields,
                'type_mismatches': type_mismatches
            })
            return False
        return True

    def validate_batch(self, items):
        """校验批量数据"""
        valid_count = 0
        invalid_count = 0

        for item in items:
            if self.validate_item(item):
                valid_count += 1
            else:
                invalid_count += 1

        return {
            'total': len(items),
            'valid': valid_count,
            'invalid': invalid_count,
            'integrity_rate': valid_count / max(len(items), 1) * 100,
            'errors': self.errors[-10:]  # 保留最近 10 条错误
        }

告警通知

python

import smtplib
import requests
from email.mime.text import MIMEText

class AlertNotifier:
    """告警通知模块"""

    def __init__(self, config):
        self.config = config

    def send_email(self, subject, body):
        """发送邮件告警"""
        msg = MIMEText(body, 'plain', 'utf-8')
        msg['Subject'] = subject
        msg['From'] = self.config['email_from']
        msg['To'] = self.config['email_to']

        with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
            server.login(self.config['smtp_user'], self.config['smtp_password'])
            server.send_message(msg)

    def send_dingtalk(self, message):
        """发送钉钉群机器人通知"""
        webhook_url = self.config['dingtalk_webhook']
        payload = {
            'msgtype': 'text',
            'text': {
                'content': message
            }
        }
        requests.post(webhook_url, json=payload)

    def send_webhook(self, message):
        """发送通用 Webhook 通知"""
        webhook_url = self.config.get('webhook_url')
        if webhook_url:
            requests.post(webhook_url, json={'text': message})

    def alert_on_failure(self, url, error, success_rate):
        """根据失败情况触发告警"""
        subject = f'爬虫告警: {url}'
        body = f'请求失败\nURL: {url}\n错误: {error}\n当前成功率: {success_rate:.1%}'

        if success_rate < 0.5:
            self.send_email(subject, body)
            self.send_dingtalk(f'爬虫严重告警\n{body}')

        self.send_webhook({'subject': subject, 'body': body})
        logging.error(f'已触发告警: {subject}')

动态页面爬取（Selenium/Puppeteer） ​

动态页面爬取原理 ​

动态页面 vs 静态页面 ​

渲染机制 ​

Selenium 爬虫 ​

Selenium 架构 ​

元素定位 ​

等待策略 ​

Selenium 实例 ​

Selenium Stealth 反检测 ​

Puppeteer/Playwright 爬虫 ​

Puppeteer 核心 API ​

Playwright ​

Puppeteer Stealth ​

Playwright vs Puppeteer 对比 ​

动态页面数据处理 ​

API 接口直接抓取 ​

提取数据清洗 ​

性能优化 ​

反检测综合策略 ​

监控与异常处理 ​

爬虫稳定性 ​

异常场景 ​

日志和监控 ​

动态页面爬取（Selenium/Puppeteer）

动态页面爬取原理

动态页面 vs 静态页面

渲染机制

Selenium 爬虫

Selenium 架构

元素定位

等待策略

Selenium 实例

Selenium Stealth 反检测

Puppeteer/Playwright 爬虫

Puppeteer 核心 API

Playwright

Puppeteer Stealth

Playwright vs Puppeteer 对比

动态页面数据处理

API 接口直接抓取

提取数据清洗

性能优化

反检测综合策略

监控与异常处理

爬虫稳定性

异常场景

日志和监控