1. 项目概述：当金融数据遇上JS反爬

做金融数据分析的朋友，最近是不是感觉数据越来越难“拿”了？以前用requests加个User-Agent就能轻松抓取的股票列表、财报摘要、实时行情，现在打开页面一片空白，或者数据变成了需要执行复杂JavaScript才能渲染出来的动态内容。这背后，正是网站为了应对自动化爬取而部署的JavaScript反爬虫技术，尤其是在数据价值高、实时性要求强的金融领域，这种对抗尤为激烈。

面对这种局面，传统的静态爬虫工具已经力不从心。我们需要一个能“像真人一样”操作浏览器的工具，去执行页面里的JS代码，等待数据渲染完成，然后再进行提取。这就是“无头浏览器”的用武之地。而在众多无头浏览器方案中，Playwright以其现代化、高性能和对多浏览器（Chromium, Firefox, WebKit）的原生支持脱颖而出。它由微软开发，API设计清晰，异步支持优秀，特别适合处理现代Web应用。

这个项目，我们就聚焦于使用Python版的Playwright，来实战攻克一个典型的金融数据网站（我们以模拟场景为例，核心思路通用），绕过其JS反爬机制，稳定、高效地获取结构化数据。整个过程，我会带你从环境搭建、核心原理剖析，到一步步编写抗封禁的健壮爬虫，并分享我在金融数据抓取中积累的独家避坑经验。

2. 核心思路与工具选型解析

2.1 为什么是Playwright？而不仅仅是Selenium

很多朋友的第一反应可能是Selenium。确实，Selenium是老牌且强大的浏览器自动化工具。但在应对现代反爬，特别是需要快速执行、资源控制精细的场景下，Playwright展现出了更明显的优势：

1. 自动等待机制：Playwright的API设计是“等待感知”的。例如，page.click(selector)这个操作，内部会自动等待该元素可点击、可见后再执行。这极大地简化了代码，我们不需要再写一堆time.sleep或显式的WebDriverWait，减少了因等待时间不当导致的抓取失败。
2. 网络拦截与模拟：Playwright可以非常方便地监听和修改网络请求。这对于反爬至关重要：我们可以直接拦截页面加载后发出的Ajax/Fetch数据请求（这些请求往往返回干净的JSON数据），而无需等待整个页面渲染完成，速度更快，资源消耗更少。
3. 更真实的浏览器环境：Playwright启动的浏览器实例，默认携带的“自动化测试”特征比老版本Selenium控制的浏览器更少。虽然网站依然能检测，但配合一些上下文（Context）级别的参数设置，可以模拟得更像普通浏览器。
4. 性能与资源：Playwright的架构更现代，通常比同等情况下的Selenium WebDriver消耗更少的内存和CPU资源。对于需要长时间运行或并发抓取的任务，这一点很重要。

对于金融数据抓取，我们经常需要处理大量由前端框架（如React, Vue）渲染的表格、图表。Playwright的“自动等待”和强大的选择器（支持文本匹配、CSS、XPath）能让我们更稳定地定位到这些动态加载的元素。

2.2 无头模式 vs. 有头模式：策略选择

Playwright可以以“无头”模式运行，即不显示浏览器图形界面。这在服务器环境或追求效率时是首选。但有头模式在开发和调试阶段无可替代。

实操心得：在开发爬虫脚本时，强烈建议先使用有头模式（headless=False）。这样你可以亲眼看到浏览器每一步在做什么：页面是否加载完全？元素是否如预期般出现？弹出的登录框或验证码是否挡住了目标？亲眼所见能帮你快速定位问题。等脚本稳定后，再切换到无头模式进行批量任务。

2.3 整体爬取策略设计

我们的目标网站假设是一个“金融数据平台”，其股票详情页的数据（如市盈率、市值、涨跌幅）是通过前端JS异步加载的。直接请求HTML得不到这些数据。我们的策略分三层：

1. 基础绕过：使用Playwright模拟真实浏览器访问，执行JS，渲染完整页面后抓取。
2. 效率优化：通过Playwright监听网络请求，直接拦截并获取数据接口（API）返回的JSON，跳过渲染开销。
3. 抗封禁增强：随机化浏览器指纹（User-Agent, Viewport, Timezone等），使用代理IP池，模拟人类操作间隔，避免触发频率限制。

下面，我们就从零开始，搭建环境并实现这一策略。

3. 环境准备与核心配置实战

3.1 Python环境与Playwright安装

首先确保你安装了Python（3.7+）。建议使用虚拟环境管理项目依赖。

# 创建并进入虚拟环境（可选但推荐） python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装Playwright的Python库 pip install playwright # 安装Playwright所需的浏览器内核（Chromium, Firefox, WebKit） playwright install chromium # 对于金融爬虫，Chromium兼容性最好，通常只装这个就够了

playwright install命令会下载浏览器二进制文件，可能需要一些时间，请保持网络通畅。

3.2 初始化浏览器与上下文：抗检测的关键一步

直接使用browser = await playwright.chromium.launch()是最简单的，但这样启动的浏览器实例容易被识别为自动化工具。我们需要创建一个配置更复杂的“上下文”来隐藏指纹。

import asyncio from playwright.async_api import async_playwright import random async def main(): async with async_playwright() as p: # 1. 启动浏览器，建议先有头调试 browser = await p.chromium.launch(headless=False, slow_mo=100) # slow_mo让动作变慢，方便观察 # 2. 创建上下文，这是模拟真实浏览器的核心 # 获取一组常见的用户代理字符串 user_agents = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 ...', # ... 可以准备更多 ] context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, # 随机化分辨率更安全 user_agent=random.choice(user_agents), locale='zh-CN', # 语言 timezone_id='Asia/Shanghai', # 时区 # 忽略HTTPS错误（某些测试环境可能需要） ignore_https_errors=False, # 设置更真实的权限，如地理位置（可选） permissions=['geolocation'] ) # 3. 创建页面 page = await context.new_page() # ... 后续爬取逻辑 await browser.close() asyncio.run(main())

注意事项：slow_mo参数在调试时非常有用，但它会显著减慢所有操作。在生产环境运行爬虫时，务必将其移除或设为0，否则你的抓取效率会极低。

3.3 核心配置：屏蔽无用资源与启用请求拦截

金融数据页面往往包含大量图片、样式表、字体和广告脚本。对于纯数据抓取，这些资源不仅拖慢速度，还消耗流量。我们可以在上下文中设置路由，拦截并中止这些请求。

async def create_anti_detect_context(browser): context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent=random.choice(USER_AGENTS), locale='zh-CN', ) # 启用请求拦截功能 await context.route('**/*', lambda route: route.abort() if route.request.resource_type in ['image', 'stylesheet', 'font', 'media'] else route.continue_() ) return context

这段代码会拦截所有类型为图片、样式表、字体和媒体的请求并中止，大幅提升页面加载速度。**/*是Playwright的URL模式匹配语法，表示匹配所有URL。

4. 爬取策略实现：从渲染抓取到API拦截

4.1 策略一：等待渲染后抓取

这是最直接的方法。我们让Playwright加载页面，等待特定元素出现，然后提取其文本内容。

假设目标页面是https://example-finance.com/stock/600519（茅台股票），市盈率数据在一个CSS类为pe-ratio的<span>标签里。

async def scrape_by_waiting(page, url): await page.goto(url) # 关键：等待目标元素出现。使用 `page.wait_for_selector` # 比 `page.wait_for_timeout` 更可靠。 try: await page.wait_for_selector('.pe-ratio', timeout=10000) # 等待10秒 pe_ratio_element = await page.query_selector('.pe-ratio') pe_ratio = await pe_ratio_element.text_content() print(f"市盈率（渲染后抓取）: {pe_ratio}") except Exception as e: print(f"等待元素超时或未找到: {e}") # 可以在这里截图，方便调试 await page.screenshot(path='debug_timeout.png')

为什么用wait_for_selector而不是time.sleep？因为网络速度和服务器响应时间不确定。固定sleep可能等不够（元素还没出来）或等太久（效率低下）。wait_for_selector是条件等待，元素一出现就继续，更智能、更健壮。

4.2 策略二：拦截API请求（高效核心）

现代网站大多采用前后端分离架构，页面骨架（HTML）先加载，然后浏览器执行JS，再通过Ajax或Fetch请求API获取数据，最后用JS把数据填充到页面上。如果我们能直接拿到这个API返回的JSON，就省去了渲染和解析DOM的步骤，效率极高，且数据更结构化。

步骤1：监听网络请求在页面跳转前，我们先设置一个请求监听器。

async def scrape_by_intercepting_api(page, url): # 用于存储拦截到的数据 intercepted_data = [] # 定义请求处理函数 def handle_response(response): # 检查响应URL是否包含我们关心的API特征 if '/api/stock/detail' in response.url: # 假设的API路径，需要实际分析 # 检查状态码和内容类型 if response.status == 200 and 'application/json' in response.headers.get('content-type', ''): # 异步获取响应体并处理 asyncio.create_task(process_response_body(response)) async def process_response_body(response): try: json_data = await response.json() intercepted_data.append(json_data) print(f"拦截到API数据: {json_data.get('symbol')}") except Exception as e: print(f"解析JSON失败: {e}") # 监听‘response’事件 page.on('response', handle_response) # 导航到目标页面 await page.goto(url) # 等待一段时间，确保所有异步请求完成 # 这里可以结合等待某个页面元素出现，作为请求完成的标志 await page.wait_for_selector('.data-loaded-indicator', timeout=15000, state='attached') # 处理收集到的数据 for data in intercepted_data: # 提取你需要的信息，例如： symbol = data.get('symbol') pe_ratio = data.get('pe_ratio') market_cap = data.get('market_cap') print(f"股票: {symbol}, 市盈率: {pe_ratio}, 市值: {market_cap}") return intercepted_data

如何找到API地址？这是关键。你需要使用浏览器开发者工具（F12）：

1. 打开目标网站页面。
2. 进入Network（网络）标签。
3. 刷新页面或进行触发数据加载的操作（如点击“加载更多”）。
4. 在请求列表中，筛选XHR或Fetch类型的请求。
5. 查看这些请求的URL和Preview，找到包含目标数据的那个请求。
6. 将URL中的特征部分（如/api/stock/detail）替换到上面的代码中。

4.3 策略三：处理复杂交互（登录、点击、滚动）

有些数据需要登录后才能查看，或者藏在“点击查看更多”、“滚动加载”的交互后面。

模拟登录：

async def login(page, login_url, username, password): await page.goto(login_url) # 等待登录表单出现 await page.wait_for_selector('input[name="username"]') # 填充表单 await page.fill('input[name="username"]', username) await page.fill('input[name="password"]', password) # 点击登录按钮 await page.click('button[type="submit"]') # 等待登录成功后的跳转或元素出现 await page.wait_for_selector('.user-avatar', timeout=10000) print("登录成功") # 重要：保存登录状态（cookies），以便后续页面使用同一个上下文 # 因为我们在同一个context里操作，cookies会自动保持

模拟点击与滚动加载：

async def scrape_paginated_data(page, base_url): all_items = [] page_num = 1 while True: url = f"{base_url}?page={page_num}" await page.goto(url) # 等待当前页数据加载 await page.wait_for_selector('.data-row') # 提取当前页数据 items = await page.eval_on_selector_all('.data-row', 'elements => elements.map(el => el.innerText)') all_items.extend(items) # 检查是否有“下一页”按钮 next_button = await page.query_selector('button:has-text("下一页")') if not next_button or await next_button.is_disabled(): break # 没有下一页了 # 模拟点击下一页 await next_button.click() # 等待新页面数据加载（确保旧数据已消失，新数据已出现） await page.wait_for_function('''() => { const oldFirstRow = document.querySelector('.data-row:first-child'); return oldFirstRow && !oldFirstRow.innerText.includes('%s'); }''' % items[0].split()[0] if items else '') page_num += 1 # 添加随机延迟，模拟人类操作，避免被封 await page.wait_for_timeout(random.randint(1000, 3000)) return all_items

5. 高级反反爬与稳定性保障

5.1 浏览器指纹伪装进阶

基础的user_agent和viewport伪装还不够。高级反爬系统会检测更多属性，如WebGL、Canvas、AudioContext、插件列表等。Playwright可以通过初始化参数注入JS代码来修改这些指纹。

context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent=random.choice(USER_AGENTS), # 注入JS代码，覆盖navigator属性 extra_http_headers={ 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', }, ) # 在页面加载前执行JS，修改或隐藏某些属性 await page.add_init_script(""" // 覆盖webdriver属性（常见检测点） Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); // 修改语言偏好 Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh', 'en'] }); // 修改插件数量，使其看起来更像普通浏览器 Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); """)

重要提示：指纹修改是一场“猫鼠游戏”。修改得越多、越彻底，越可能引入不一致性，反而被识别。对于大多数金融网站，做好基础伪装（UA、分辨率、时区）并配合合理的请求节奏，通常已足够。过度修改可能适得其反。

5.2 代理IP池集成

单一IP高频访问是封禁的主要原因。必须使用代理IP池。

# 假设你有一个函数 get_proxy() 从代理池获取一个代理字符串，格式如 'http://user:pass@host:port' PROXY_SERVER = get_proxy() browser = await p.chromium.launch( headless=True, proxy={ 'server': PROXY_SERVER, # 如果需要用户名密码认证 # 'username': 'your_username', # 'password': 'your_password' } )

代理使用心得：

1. 质量优先：金融网站对代理质量要求高，透明代理、数据中心代理容易被封。尽量使用高质量住宅代理或移动代理，虽然成本高，但成功率高。
2. 轮换策略：不要一个IP用到死。可以设置每抓取N个页面或遇到特定HTTP状态码（如429， 403）时，更换浏览器上下文（连带更换代理）。
3. 本地测试：先不用代理，确保脚本逻辑正确。再加代理，并准备好处理代理失效（超时、连接拒绝）的异常重试机制。

5.3 请求节奏控制与随机化

模拟人类浏览的不规律性。

import random import time async def human_like_delay(page): """在关键操作间添加随机延迟""" # 随机延迟1到5秒 delay = random.uniform(1, 5) # 使用page.wait_for_timeout，它是异步的，不会阻塞事件循环 await page.wait_for_timeout(delay * 1000) # 单位是毫秒 # 在爬取循环中使用 for item in items_to_process: await page.click(item_selector) await human_like_delay(page) # ... 处理数据 await human_like_delay(page)

此外，随机化操作顺序（如果逻辑允许）、模拟鼠标移动（page.mouse.move(x, y)）等，都能增加行为的真实性。

6. 实战案例：抓取模拟金融数据平台

假设我们要从一个模拟的金融平台https://demo-finance-data.com/market抓取所有股票的实时概览（代码、名称、最新价、涨跌幅）。该页面数据通过滚动加载，并且需要处理一个简单的滑块验证码。

6.1 页面分析与难点识别

1. 数据加载方式：打开页面只加载前20条，滚动到底部时，通过JS触发API加载下一页。
2. 反爬措施：
   • 频繁请求会弹出滑块验证码。
   • 请求头有简单校验。
   • 对无头浏览器有基础检测。

6.2 爬虫脚本编写

import asyncio import random import json from playwright.async_api import async_playwright USER_AGENTS = [...] DEMO_URL = "https://demo-finance-data.com/market" async def solve_slide_captcha(page): """处理简单的滑块验证码（假设滑块是横向移动）""" # 1. 定位滑块轨道和滑块按钮 slider_track = await page.wait_for_selector('.captcha-slider-track') slider_button = await page.wait_for_selector('.captcha-slider-button') # 2. 获取轨道和滑块的边界框 track_box = await slider_track.bounding_box() button_box = await slider_button.bounding_box() # 3. 计算需要拖动的距离（假设验证码需要拖到最右端） target_x = track_box['x'] + track_box['width'] - button_box['width'] / 2 start_x = button_box['x'] + button_box['width'] / 2 distance = target_x - start_x # 4. 模拟人类拖动：不是一次性拖到位，而是有加速、减速的过程 await slider_button.hover() await page.mouse.down() # 分段移动，模拟人手 steps = int(distance / 5) for i in range(steps): # 移动速度可以随机变化 step = distance / steps + random.uniform(-1, 1) await page.mouse.move(start_x + i * (distance/steps) + step, button_box['y'] + button_box['height']/2) await page.wait_for_timeout(random.randint(20, 50)) # 每步小延迟 # 最后移动到精确位置 await page.mouse.move(target_x, button_box['y'] + button_box['height']/2) await page.mouse.up() print("滑块验证码处理完成") # 等待验证通过，页面变化 await page.wait_for_timeout(2000) async def scrape_market_data(): async with async_playwright() as p: # 启动浏览器，配置代理（此处省略代理配置） browser = await p.chromium.launch(headless=False, slow_mo=200) # 调试阶段慢速 context = await browser.new_context( viewport={'width': 1366, 'height': 768}, user_agent=random.choice(USER_AGENTS), locale='zh-CN', # 设置一些常见的HTTP头 extra_http_headers={ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Cache-Control': 'no-cache', } ) page = await context.new_page() # 监听数据API stock_data_list = [] def on_api_response(response): if '/api/market/list' in response.url and response.status == 200: asyncio.create_task(parse_api_response(response)) async def parse_api_response(response): try: data = await response.json() if data.get('code') == 0: # 假设返回码0表示成功 stock_data_list.extend(data.get('data', [])) print(f"已拦截到 {len(data.get('data', []))} 条数据") except Exception as e: print(f"解析API响应失败: {e}") page.on('response', on_api_response) # 访问首页 await page.goto(DEMO_URL, wait_until='networkidle') # 等待网络空闲 # 检查是否有验证码 if await page.query_selector('.captcha-container'): print("检测到验证码，正在处理...") await solve_slide_captcha(page) # 模拟滚动加载，直到没有新数据 previous_data_count = 0 max_scroll_attempts = 10 attempts = 0 while attempts < max_scroll_attempts: # 滚动到底部 await page.evaluate('window.scrollTo(0, document.body.scrollHeight)') # 等待可能的新数据加载 await page.wait_for_timeout(3000 + random.randint(0, 2000)) # 检查数据是否增长 current_count = len(stock_data_list) if current_count > previous_data_count: print(f"滚动后数据增加: {previous_data_count} -> {current_count}") previous_data_count = current_count attempts = 0 # 重置尝试计数 else: attempts += 1 print(f"第 {attempts} 次滚动未发现新数据") # 可以尝试点击“加载更多”按钮（如果有） load_more_btn = await page.query_selector('button.load-more') if load_more_btn: await load_more_btn.click() await page.wait_for_timeout(2000) # 随机延迟，避免规律请求 await page.wait_for_timeout(random.randint(1000, 3000)) print(f"共抓取到 {len(stock_data_list)} 条股票数据") # 保存数据到JSON文件 with open('market_data.json', 'w', encoding='utf-8') as f: json.dump(stock_data_list, f, ensure_ascii=False, indent=2) await browser.close() if __name__ == '__main__': asyncio.run(scrape_market_data())

6.3 案例总结与关键点

1. 混合策略：本例结合了API拦截（主）和滚动模拟（辅）。优先拦截高效的数据接口，同时用滚动操作来触发接口调用。
2. 验证码处理：演示了如何处理一种简单的交互式验证码。对于更复杂的验证码（如点选、语序），可能需要引入专门的识别服务或手动处理。
3. 循环终止条件：设置了最大滚动尝试次数，防止因页面逻辑错误导致无限循环。
4. 数据去重：拦截到的数据可能在多次响应中有重复，实际应用中需要根据唯一标识（如股票代码）进行去重。

7. 常见问题排查与性能优化

7.1 典型错误与解决方案

7.2 性能优化技巧

1. 复用浏览器上下文：创建和启动浏览器的开销很大。如果有多项抓取任务，应复用同一个浏览器实例和上下文，只创建新的页面（Page）。

   async def task_runner(): async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context(...) # 配置一次 tasks = [] for url in url_list: # 每个任务使用独立的页面，但共享上下文（cookies、缓存等） task = asyncio.create_task(scrape_one_page(context, url)) tasks.append(task) await asyncio.gather(*tasks) # 并发执行 await browser.close()

2. 并发控制：虽然可以开很多个Page并发，但过多会耗尽内存和CPU。使用信号量（asyncio.Semaphore）控制最大并发数。

   semaphore = asyncio.Semaphore(5) # 最多5个页面并发 async def scrape_with_limit(context, url): async with semaphore: page = await context.new_page() await page.goto(url) # ... 抓取逻辑 await page.close()

3. 选择性加载资源：如3.3节所示，拦截不必要的资源（图片、字体、CSS）能极大提升加载速度。

4. 善用wait_for_selector的状态参数：除了等待元素出现（state=’visible’默认），还可以等待元素隐藏、从DOM分离等，更精确地控制流程。

   # 等待“加载中”提示消失 await page.wait_for_selector(‘.loading-spinner’, state=‘hidden’) # 等待某个元素被添加到DOM（但不一定可见） await page.wait_for_selector(‘.new-item’, state=‘attached’)

7.3 日志与监控

一个健壮的爬虫需要完善的日志记录，以便出错时回溯。

import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler('crawler.log'), logging.StreamHandler()]) async def scrape_something(page): logging.info(f“开始抓取页面: {page.url}”) try: # ... 抓取操作 logging.info(“抓取成功”) except Exception as e: logging.error(f“抓取失败: {e}”, exc_info=True) # 出错时截图 await page.screenshot(path=f“error_{int(time.time())}.png”)

将关键步骤、拦截到的数据量、遇到的错误都记录下来。结合定时任务（如cron或APScheduler），你可以构建一个长期运行的金融数据采集系统。

最后，请始终牢记合法合规与**尊重robots.txt**的原则。在开始任何爬取项目前，务必检查目标网站的robots.txt文件和服务条款，明确哪些数据可以抓取，频率限制是多少。对目标网站的资源保持友好，设置合理的请求间隔，避免对其正常运营造成压力。将抓取的数据用于个人学习、分析或法律允许的范围内，是每一位技术从业者应尽的责任。