Python Playwright爬虫实战:高效采集动态网页数据

1. 项目概述:为什么是Playwright?

最近在社区里看到不少朋友在讨论数据采集,尤其是用Python做爬虫。传统的requests+BeautifulSoup或者Scrapy框架依然是主力,但越来越多的人开始抱怨:现在的网页越来越“花哨”了,大量数据通过JavaScript动态加载,甚至藏在需要交互(比如点击按钮、滚动页面)之后才会出现的元素里。对付这些“现代”网页,老办法要么力不从心,要么写出来的代码又臭又长,维护起来头疼。

这正是我这次想分享的切入点。标题里的“Playwright爬虫数据采集代码示例”,核心价值就在于它提供了一个应对现代Web应用的、更优雅高效的解决方案。Playwright不是一个单纯的爬虫库,它是一个由微软开源的浏览器自动化测试框架。但正因为它能“驱动”真实的浏览器(如Chromium, Firefox, WebKit),所以它天然具备了处理动态内容、执行复杂交互的能力,这让它在数据采集领域大放异彩。

简单来说,如果你要采集的网站:

  1. 数据是异步加载的(AJAX)。
  2. 需要登录后才能访问。
  3. 页面结构复杂,依赖大量JavaScript渲染。
  4. 有反爬机制,对简单的HTTP请求拦截严格。

那么,Playwright很可能就是你的“瑞士军刀”。它模拟的是真实用户行为,从浏览器指纹到网络请求都高度仿真,极大地降低了被反爬策略识别和封锁的风险。接下来的内容,我会用一个完整的、可运行的代码示例,带你从零开始,深入理解如何用Playwright构建一个健壮的数据采集脚本,并分享我踩过的一些坑和实战技巧。

2. 核心思路与方案选型

在动手写代码之前,我们先理清思路。一个数据采集项目,无外乎几个核心环节:发起请求、解析页面、提取数据、处理异常、存储结果。Playwright在这些环节中扮演了什么角色?我们又该如何搭配其他工具?

2.1 Playwright vs. 传统方案的优劣对比

为什么选择Playwright而不是Selenium或纯HTTP请求?我们可以从几个维度来看:

特性/方案requests+BeautifulSoup/lxmlSeleniumPlaywright
核心原理发送HTTP请求,解析静态HTML。通过WebDriver协议驱动浏览器。通过DevTools协议直接与浏览器内核通信。
处理动态内容无法直接处理,需手动分析API或使用Selenium辅助。优秀,可等待元素、执行JS。极佳,内置智能等待,执行JS更高效。
执行速度极快,无浏览器开销。较慢,浏览器启动和驱动通信有开销。较快,协议更高效,支持无头模式且轻量。
资源消耗极低高,每个实例一个完整浏览器进程。中等,但优于Selenium,上下文隔离好。
安装与配置简单,pip install即可。复杂,需下载对应浏览器驱动并匹配版本。简单,playwright install一键安装浏览器。
跨浏览器支持无。支持,但需各自驱动。原生支持Chromium, Firefox, WebKit,API统一。
网络拦截与模拟需借助mitmproxy等,较复杂。有限支持。强大,可轻松拦截修改请求/响应,模拟地理、语言等。
移动端模拟无。有限支持。支持良好,可模拟特定设备(如iPhone)。

选择Playwright的关键理由

  1. 对现代Web的完美支持:单页应用(SPA)是常态,Playwright的等待策略(wait_for_selector,wait_for_load_state)能优雅地处理这些异步加载。
  2. 开发体验友好:自动等待机制减少了大量time.sleep的猜测代码。Playwright Codegen可以录制操作生成代码,对快速原型开发帮助巨大。
  3. 强大的调试能力playwright debug模式、追踪录制(trace viewer)功能,让定位页面问题变得直观。
  4. 统一的API:一套代码稍作调整即可运行在三大浏览器引擎上,方便测试兼容性或应对不同站点的限制。

2.2 项目整体架构设计

我们的示例项目将采集一个模拟的图书网站信息。这个网站有分页、有动态加载、可能需要处理登录状态(为简化,示例中跳过登录,但会给出思路)。整体架构如下:

  1. 初始化与配置:启动Playwright,创建浏览器上下文(Context),设置代理、用户代理(UA)、视口大小等,模拟真实环境。
  2. 导航与等待:访问目标URL,使用Playwright的智能等待确保目标内容加载完成。
  3. 数据定位与提取:使用CSS选择器或XPath定位到包含数据的HTML元素。
  4. 数据处理与清洗:从提取的原始文本中,通过正则表达式或字符串方法清洗出结构化数据(如书名、价格、评分)。
  5. 分页与循环:定位“下一页”按钮或构造分页URL,实现自动翻页采集。
  6. 异常处理与持久化:网络超时、元素缺失等异常情况的捕获与处理,并将最终数据保存为JSON或CSV文件。
  7. 资源清理:关闭浏览器上下文和浏览器实例,释放资源。

我们将使用asyncio异步API,因为Playwright的异步API性能更高,更适合I/O密集型的爬虫任务。同时,我们会搭配pandas进行简单的数据整理和导出。

3. 环境准备与核心依赖安装

工欲善其事,必先利其器。让我们先把环境搭建起来。

3.1 创建虚拟环境与安装包

强烈建议使用虚拟环境来管理项目依赖,避免包冲突。

# 1. 创建项目目录并进入 mkdir playwright-spider-demo && cd playwright-spider-demo # 2. 创建虚拟环境(这里使用venv,你也可以用conda) python -m venv venv # 3. 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 4. 安装Playwright pip install playwright # 5. 安装Playwright所需的浏览器(Chromium, Firefox, WebKit) # 这一步会下载浏览器,时间稍长,请保持网络通畅 playwright install chromium # 如果你需要测试其他浏览器,也可以安装 firefox 或 webkit # 6. 安装数据处理和存储需要的库 pip install pandas

注意playwright install命令会下载浏览器到用户目录下的缓存中。如果遇到网络问题导致下载失败,可以尝试设置镜像源,或者使用playwright install --help查看离线安装选项。

3.2 验证安装与快速测试

安装完成后,写一个最简单的脚本验证一切是否正常。

# test_install.py import asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: # 启动Chromium浏览器,headless=True表示无头模式(不显示GUI) browser = await p.chromium.launch(headless=True) # 创建一个新的浏览器上下文(类似于一个独立的隐身会话) context = await browser.new_context() # 在新标签页中打开页面 page = await context.new_page() # 导航到百度 await page.goto('https://www.baidu.com') # 获取页面标题并打印 title = await page.title() print(f'页面标题: {title}') # 截图保存(可选) await page.screenshot(path='baidu.png') # 关闭资源 await context.close() await browser.close() asyncio.run(main())

运行这个脚本:

python test_install.py

如果输出页面标题: 百度一下,你就知道,并且当前目录下生成了baidu.png图片,那么恭喜你,Playwright环境已经成功配置。

4. 实战:构建一个完整的图书信息采集器

现在,我们进入核心部分。假设我们要采集一个模拟网站http://books.toscrape.com(这是一个著名的爬虫练习网站)上的图书信息。虽然它是静态网站,但我们将用Playwright的方式来处理,并模拟一些动态场景。

4.1 目标分析与页面结构探查

首先,手动打开目标网站,用浏览器的开发者工具(F12)查看页面结构。

  1. 列表页:http://books.toscrape.com/catalogue/page-1.html
  2. 每本书在一个<article class="product_pod">元素内。
  3. 我们需要的信息:书名(<h3> <a>title属性)、价格(<p class="price_color">)、库存状态(<p class="instock availability">)、评分(<p class="star-rating的class,如Three代表3星)。
  4. 分页:页面底部有next按钮,点击可进入下一页。

实操心得:在写爬虫前,花10分钟仔细分析页面结构,能节省后面几小时的调试时间。重点关注数据的容器元素其唯一标识(如class、id)。善用开发者工具的Copy selectorCopy XPath功能,但不要完全依赖,自动生成的路径可能很脆弱。

4.2 核心代码实现与逐行解析

下面是我们完整的采集脚本,我将结合代码详细解释每个部分。

# book_scraper_async.py import asyncio import pandas as pd from playwright.async_api import async_playwright import logging import json from typing import List, Dict, Any # 配置日志,方便调试 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class BookScraper: def __init__(self, headless: bool = True): """ 初始化采集器 :param headless: 是否使用无头模式。调试时可设为False看浏览器操作。 """ self.headless = headless self.all_books: List[Dict[str, Any]] = [] # 存储所有图书数据 async def scrape_page(self, page, page_num: int) -> bool: """ 采集单个列表页的数据 :param page: Playwright页面对象 :param page_num: 当前页码,用于日志 :return: 是否成功采集到数据 """ logger.info(f'正在采集第 {page_num} 页...') try: # 核心:等待图书列表的容器元素加载完成 # 这里使用 `wait_for_selector` 确保动态内容已渲染 await page.wait_for_selector('article.product_pod', timeout=10000) # 使用 `query_selector_all` 获取所有图书元素 book_elements = await page.query_selector_all('article.product_pod') logger.info(f'第 {page_num} 页找到 {len(book_elements)} 本书。') for book_element in book_elements: book_info = {} # 1. 提取书名 (从h3 > a标签的title属性) title_elem = await book_element.query_selector('h3 > a') if title_elem: book_info['title'] = await title_elem.get_attribute('title') else: book_info['title'] = None # 2. 提取价格 price_elem = await book_element.query_selector('p.price_color') if price_elem: price_text = await price_elem.text_content() # 清洗价格字符串,例如 "£51.77" -> 51.77 book_info['price'] = float(price_text.replace('£', '')) if price_text else None else: book_info['price'] = None # 3. 提取库存状态 stock_elem = await book_element.query_selector('p.instock.availability') if stock_elem: stock_text = await stock_elem.text_content() book_info['in_stock'] = 'In stock' in stock_text.strip() if stock_text else False else: book_info['in_stock'] = False # 4. 提取评分 (从star-rating类的第二个单词,如'Three') rating_elem = await book_element.query_selector('p.star-rating') if rating_elem: class_name = await rating_elem.get_attribute('class') # class可能是 "star-rating Three" rating_map = {'One': 1, 'Two': 2, 'Three': 3, 'Four': 4, 'Five': 5} rating_word = class_name.split()[-1] if class_name else None book_info['rating'] = rating_map.get(rating_word, None) else: book_info['rating'] = None # 5. 提取详情页链接 (可选,用于后续深度采集) link_elem = await book_element.query_selector('h3 > a') if link_elem: book_info['detail_url'] = await link_elem.get_attribute('href') else: book_info['detail_url'] = None self.all_books.append(book_info) return True except Exception as e: logger.error(f'采集第 {page_num} 页时发生错误: {e}') # 可以在这里保存错误页面截图,辅助调试 # await page.screenshot(path=f'error_page_{page_num}.png') return False async def run(self, start_url: str, max_pages: int = 5): """ 主运行函数 :param start_url: 起始列表页URL :param max_pages: 最大采集页数,防止无限循环 """ async with async_playwright() as p: # 启动浏览器,推荐chromium,兼容性最好 browser = await p.chromium.launch(headless=self.headless) # 创建上下文,可以在这里统一设置用户代理、视口、语言等 context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, 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' ) page = await context.new_page() current_url = start_url current_page_num = 1 while current_url and current_page_num <= max_pages: # 导航到当前页 try: await page.goto(current_url, wait_until='networkidle', timeout=30000) # wait_until='networkidle' 等待页面网络活动基本停止,对SPA友好 except Exception as e: logger.error(f'导航到 {current_url} 失败: {e}') break # 采集当前页数据 success = await self.scrape_page(page, current_page_num) if not success: logger.warning(f'第 {current_page_num} 页采集失败,停止翻页。') break # 寻找“下一页”按钮 next_button = await page.query_selector('li.next > a') if next_button: # 获取下一页的链接 next_url = await next_button.get_attribute('href') # 处理相对路径 from urllib.parse import urljoin current_url = urljoin(current_url, next_url) current_page_num += 1 logger.info(f'找到下一页,跳转到: {current_url}') else: logger.info('已到达最后一页。') break # 关闭浏览器 await context.close() await browser.close() logger.info(f'采集结束,共采集到 {len(self.all_books)} 条图书数据。') def save_to_file(self, format: str = 'json'): """将采集到的数据保存到文件""" if not self.all_books: logger.warning('没有数据可保存。') return filename = f'books_data.{format}' if format == 'json': with open(filename, 'w', encoding='utf-8') as f: json.dump(self.all_books, f, ensure_ascii=False, indent=2) elif format == 'csv': df = pd.DataFrame(self.all_books) df.to_csv(filename, index=False, encoding='utf-8-sig') else: logger.error(f'不支持的格式: {format}') return logger.info(f'数据已保存至: {filename}') async def main(): scraper = BookScraper(headless=True) # 生产环境用True,调试可改为False start_url = 'http://books.toscrape.com/catalogue/page-1.html' await scraper.run(start_url, max_pages=3) # 先采集3页作为演示 scraper.save_to_file('json') # 保存为JSON scraper.save_to_file('csv') # 同时保存一份CSV if __name__ == '__main__': asyncio.run(main())

4.3 代码关键点深度解析

  1. 异步上下文管理器 (async with):async_playwright()browser.launch()都支持异步上下文管理,确保资源(如浏览器进程)在使用后被正确关闭,即使发生异常也是如此。这是编写健壮代码的好习惯。

  2. 智能等待 (wait_for_selector,wait_until): 这是Playwright的精华。

    • page.goto(url, wait_until='networkidle')networkidle会等待页面在至少500毫秒内没有超过2个网络连接时才算加载完成,这对加载了大量异步资源的现代网页非常有效。其他选项还有load(DOMContentLoaded事件),domcontentloaded,commit
    • page.wait_for_selector('article.product_pod'):显式等待我们关心的数据容器出现。这比固定sleep或等待整个页面加载完更精准、更高效。
  3. 元素查询与数据提取:

    • page.query_selector_all('css selector'):返回一个元素列表(ElementHandle)。
    • element.query_selector('css selector'):在某个元素内继续查找子元素。
    • text_content()vsinner_text()text_content获取所有文本,包括<script><style>里的(通常不用),inner_text更接近用户看到的,会考虑CSS样式(如display: none的元素不会被包含)。爬虫中通常用text_content()
    • get_attribute('attr_name'):获取元素属性值,如href,title,src
  4. 错误处理与日志:脚本中使用了try...except包裹核心采集逻辑,并记录了详细的日志。在实际项目中,你还需要考虑更细粒度的重试机制(例如,某个元素找不到时重试几次)。

  5. 分页逻辑:我们通过查找具有特定CSS选择器(li.next > a)的“下一页”按钮来实现自动翻页。这是一种通用模式。有些网站分页是通过URL参数(如?page=2)控制的,那时就需要解析当前URL并构造下一页URL。

5. 高级技巧与性能优化

基础的采集器已经能工作,但要用于生产环境,还需要考虑更多。

5.1 请求拦截与资源过滤

很多网页加载了图片、字体、CSS、广告脚本等资源,对于爬虫来说这些是不必要的,只会拖慢速度并消耗流量。Playwright可以轻松拦截和过滤请求。

async def set_up_context(context): """设置上下文,并拦截不必要的请求""" # 路由拦截 await context.route('**/*', lambda route: handle_route(route)) async def handle_route(route): """处理单个路由请求""" # 获取请求资源类型 resource_type = route.request.resource_type # 只允许文档、XHR(Ajax)、脚本、fetch请求通过 if resource_type in ['document', 'xhr', 'script', 'fetch']: await route.continue_() else: # 中止图片、样式表、字体等请求 await route.abort() # 在主函数中创建context后调用 context = await browser.new_context() await set_up_context(context) page = await context.new_page()

注意:过度拦截可能导致页面渲染异常,因为有些脚本是页面功能所必需的。建议先观察页面网络请求,确定哪些是数据请求(通常是XHR或Fetch),再针对性拦截。对于目标网站,可以先不禁用任何请求,稳定后再逐步优化。

5.2 并发控制与速率限制

同时打开太多页面会消耗大量内存和CPU,也容易被目标网站封禁。我们需要控制并发度,并添加请求间隔。

import asyncio from asyncio import Semaphore async def scrape_with_concurrency(start_urls, max_concurrent=3, delay=1.0): """ 并发采集多个起始URL :param start_urls: 起始URL列表 :param max_concurrent: 最大并发页面数 :param delay: 每次请求之间的延迟(秒) """ semaphore = Semaphore(max_concurrent) # 控制并发信号量 tasks = [] for i, url in enumerate(start_urls): # 为每个URL创建一个任务,并传递信号量 task = asyncio.create_task(scrape_single_url(url, semaphore, delay * i)) tasks.append(task) await asyncio.gather(*tasks) async def scrape_single_url(url, semaphore, initial_delay): """采集单个URL,受信号量控制""" async with semaphore: await asyncio.sleep(initial_delay) # 初始延迟,错开启动 # ... 实际的页面打开和数据采集逻辑 ... logger.info(f'开始采集: {url}') # 模拟采集时间 await asyncio.sleep(2) logger.info(f'完成采集: {url}')

在实际项目中,更精细的控制可以结合asyncio.Queue和生产者-消费者模型。

5.3 处理复杂交互:登录、下拉、弹窗

Playwright的强大之处在于能模拟几乎所有用户交互。

模拟登录(以表单为例):

await page.goto('https://example.com/login') # 等待输入框出现 await page.wait_for_selector('#username') # 填充表单 await page.fill('#username', 'your_username') await page.fill('#password', 'your_password') # 点击登录按钮 await page.click('button[type="submit"]') # 等待登录成功后的跳转或某个元素出现 await page.wait_for_selector('#user-profile', timeout=10000) # 登录后的状态(cookies, localStorage)会被自动保存在context中

处理下拉加载(滚动):

# 方法1:模拟滚动到底部 await page.evaluate('window.scrollTo(0, document.body.scrollHeight)') # 等待新内容加载(需要观察页面行为) await page.wait_for_timeout(2000) # 简单等待,不稳定 # 更好的方法是等待某个新出现的元素 # await page.wait_for_selector('.new-item-class', timeout=5000) # 方法2:自动滚动直到没有新内容 prev_height = None while True: curr_height = await page.evaluate('document.body.scrollHeight') if prev_height == curr_height: break await page.evaluate(f'window.scrollTo(0, {curr_height})') await page.wait_for_timeout(1000) # 等待加载 prev_height = curr_height

处理弹窗/对话框:

# 监听对话框(alert, confirm, prompt)并接受 page.on('dialog', lambda dialog: dialog.accept()) # 或者更精确地处理 def handle_dialog(dialog): print(f'对话框消息: {dialog.message}') if dialog.type == 'alert': dialog.accept() elif dialog.type == 'confirm': dialog.dismiss() # 或 dialog.accept() page.on('dialog', handle_dialog)

6. 反爬策略应对与道德规范

使用Playwright这类浏览器自动化工具,虽然绕过了很多基于请求头或简单JS的反爬,但并非万能。高价值的网站可能有更复杂的检测机制。

6.1 常见反爬手段与Playwright应对

  1. 浏览器指纹检测:网站通过检测navigator,screen,WebGL等属性来判断是否为真实浏览器。

    • 应对:Playwright创建的浏览器上下文本身具有完整的指纹。但你可以进一步随机化或自定义这些属性,不过要小心过度修改导致指纹异常。
    context = await browser.new_context( viewport={'width': 1366, 'height': 768}, user_agent='随机UA字符串', locale='zh-CN', timezone_id='Asia/Shanghai', # 可以覆盖更多属性,但需谨慎 # permissions=['geolocation'] # 模拟地理位置授权 )
  2. 行为模式检测:过于规律的请求间隔、鼠标移动轨迹过于机械等。

    • 应对:引入随机延迟、使用page.mouse.move()模拟人类鼠标移动轨迹。
    import random await asyncio.sleep(random.uniform(1.0, 3.0)) # 随机延迟1-3秒 # 模拟鼠标移动到某个元素上 element = await page.query_selector('button') box = await element.bounding_box() await page.mouse.move(box['x'] + box['width']/2, box['y'] + box['height']/2)
  3. 验证码:这是最棘手的。

    • 应对
      • 规避:尝试降低请求频率,避免触发验证码。
      • 识别服务:对接第三方打码平台(如超级鹰、图鉴等),但会增加成本和复杂度。
      • 手动处理:在调试模式(headless=False)下弹出验证码时,手动输入。
      • Cookie复用:登录一次后,保存context的cookies,下次直接加载使用,避免重复登录触发验证码。
      # 保存cookies cookies = await context.cookies() with open('cookies.json', 'w') as f: json.dump(cookies, f) # 加载cookies with open('cookies.json', 'r') as f: cookies = json.load(f) await context.add_cookies(cookies)

6.2 遵守Robots协议与道德规范

这是每个爬虫开发者必须牢记的底线。

  • 尊重robots.txt:在爬取前,访问目标网站/robots.txt,查看是否允许爬取你目标目录,以及规定的爬取延迟(Crawl-delay)。虽然Playwright不直接解析robots.txt,但你应该手动遵守。
  • 控制爬取速度:设置合理的请求间隔(如2-5秒),避免对目标网站服务器造成压力。我们的示例中使用了delay参数。
  • 识别并遵守User-Agent声明:有些网站在robots.txt里声明了友好的爬虫User-Agent,你可以使用它。
  • 只爬取公开数据:不要尝试爬取需要付费登录或明确声明版权保护的数据。
  • 明确数据用途:爬取的数据应用于个人学习、研究或公益目的,切勿用于商业牟利或侵犯他人权益。

重要提示:网络上的公开言论如“写爬虫要限制下,压力太大,把正规爬虫挤得都没带宽了”反映了部分网站管理员的无奈。作为开发者,我们有责任维护良好的网络生态。一个负责任的爬虫应该是“礼貌”的、可识别的(通过合理的User-Agent)、且对服务器影响最小的。

7. 调试、问题排查与性能监控

即使代码写得再好,运行时也难免遇到各种问题。Playwright提供了一系列强大的调试工具。

7.1 常用调试技巧

  1. 慢动作与无头模式:调试时,关闭无头模式,并启用慢动作,可以看清每一步操作。

    browser = await p.chromium.launch(headless=False, slow_mo=500) # 每个操作延迟500毫秒
  2. 录制操作:使用playwright codegen命令,可以打开一个浏览器和代码生成器,你的操作会被实时转换成Playwright代码。这是学习API和快速生成脚本原型的利器。

    playwright codegen http://books.toscrape.com
  3. 截图与录屏:在关键步骤或出错时截图。

    await page.screenshot(path='debug.png', full_page=True) # 或者录制整个会话的追踪信息(更强大) await context.tracing.start(screenshots=True, snapshots=True) # ... 执行操作 ... await context.tracing.stop(path='trace.zip') # 用 `playwright show-trace trace.zip` 查看
  4. 控制台输出:执行页面中的JavaScript并打印结果。

    console_log = await page.evaluate('() => { return document.title; }') print(console_log) # 或者监听页面的console事件 page.on('console', lambda msg: print(f'页面日志: {msg.text}'))

7.2 常见问题排查表

问题现象可能原因排查步骤与解决方案
**元素找不到 (TimeoutError) **1. 选择器错误或过时。
2. 页面未完全加载/元素是动态生成的。
3. 元素在iframe内。
4. 页面有弹窗遮挡。
1. 用开发者工具重新检查元素,使用更稳定的选择器(如>页面卡住/加载慢1. 网络问题或目标服务器慢。
2. 页面有无限循环的JS或未完成的资源请求。
3. 拦截了关键资源。
1. 增加page.gototimeout
2. 使用wait_until='domcontentloaded'而非'networkidle',或设置超时后强制继续。
3. 检查路由拦截规则,是否误杀了必要资源(如API请求)。
被网站屏蔽1. IP被识别为爬虫并封禁。
2. 浏览器指纹被检测。
3. 行为模式异常。
1. 使用代理IP池轮换。
2. 确保使用完整的浏览器上下文,避免使用--disable-blink-features等容易被检测的标志。
3. 增加随机延迟、模拟鼠标移动、使用更“人类”的浏览模式。
内存泄漏/浏览器崩溃1. 未正确关闭页面或上下文。
2. 打开的页面过多。
3. 页面内有内存泄漏的JS。
1. 始终使用async with或确保在finally块中调用close()
2. 控制并发页面数,及时关闭不再需要的页面(await page.close())。
3. 定期重启浏览器实例。

7.3 性能监控与优化建议

对于长时间运行的大规模采集任务,监控是必要的。

  • 日志分级:使用logging模块设置DEBUG,INFO,WARNING,ERROR等级别,生产环境只记录ERROR和关键INFO
  • 统计指标:记录已采集页面数、成功率、平均耗时、失败URL等。
  • 资源监控:监控脚本进程的内存和CPU使用情况。如果持续增长,可能存在内存泄漏。
  • 优化建议
    • 重用浏览器实例:避免为每个任务都启动/关闭浏览器,开销巨大。
    • 创建多个上下文(Context):每个上下文有独立的cookie、缓存,但共享浏览器进程,比启动多个浏览器更轻量。
    • 谨慎使用page.evaluate:频繁在页面和Python环境之间传递数据有性能成本,尽量批量操作。
    • 禁用不必要的功能:如对于纯数据采集,可以禁用图片加载、Service Worker等。
    browser = await p.chromium.launch(args=['--disable-images', '--disable-service-worker'])

8. 项目扩展与进阶方向

这个示例项目是一个起点。根据实际需求,你可以从以下几个方向进行扩展:

  1. 分布式爬虫:使用Celery+RedisScrapy+Scrapy-Redis架构,将URL分发到多台机器或多个进程,用Playwright作为下载器中间件来处理动态页面。
  2. 集成Scrapy:虽然Scrapy本身是异步框架,但直接集成Playwright的异步API需要一些技巧。可以使用scrapy-playwright这样的第三方中间件,它很好地桥接了两者。
  3. 数据管道:采集后的数据可以实时推送到消息队列(如Kafka、RabbitMQ),然后由下游的数据清洗、入库服务消费。
  4. 容器化部署:将爬虫脚本、Playwright浏览器打包进Docker镜像,便于在云服务器或K8s集群上调度运行。注意在Docker中运行需要安装必要的系统依赖。
  5. 更复杂的解析:对于极其复杂的页面,可以结合BeautifulSoupparsel进行HTML解析,或者直接使用Playwright获取渲染后的HTML字符串再用这些库解析,有时比纯Playwright选择器更灵活。
  6. API逆向工程:如果网站的数据是通过API接口(XHR/Fetch)获取的,直接用Playwright拦截这些请求,直接获取结构化数据(通常是JSON),效率远高于解析HTML。使用page.on('request')page.on('response')事件监听器可以轻松做到。

我个人在长期使用Playwright进行数据采集后,最大的体会是:它模糊了“爬虫”和“自动化测试”的边界。你用来保证网页功能正确的工具,同样可以高效地获取数据。这种思路的转变,往往能帮你打开一扇新的大门。最后一个小技巧,善用playwright inspector(通过设置PWDEBUG=1环境变量启动),它能让你以可交互的方式单步调试你的脚本,对于理解页面加载顺序和定位元素找不到的问题有奇效。