企业官网建设流程全解析

1. 项目概述：为什么我们需要现代浏览器自动化

如果你还在用传统的requests+BeautifulSoup或者Selenium写爬虫，可能会觉得越来越吃力。现在的网站，尤其是那些重度依赖 JavaScript 渲染的单页应用（SPA），数据加载逻辑复杂，反爬机制层出不穷。我之前接手一个电商价格监控项目，用老方法抓取动态加载的商品列表和价格，不仅代码臃肿，而且稳定性极差，动不动就因为页面元素没加载出来而报错。直到我开始系统性地使用Playwright配合Python 异步（asyncio）技术栈，才真正体会到什么叫“降维打击”。

这个组合解决的不仅仅是“能爬”的问题，更是“高效、稳定、易维护”地爬取。Playwright 由微软开发，它不像 Selenium 那样只是一个浏览器驱动桥接器，而是直接通过DevTools Protocol与浏览器内核对话，能精准模拟几乎所有用户操作：点击、输入、滚动、拖拽，甚至能拦截和修改网络请求。而 Python 的异步编程，则让我们能同时控制多个浏览器实例或标签页，将传统的“单线程排队”式爬取，升级为“多车道并发”作业，效率提升不是一点半点。

简单来说，这个实战项目适合所有希望爬虫技术更上一层楼的开发者。无论你是想自动化处理一些日常的网页操作（比如自动填报、数据收集），还是需要构建一个高并发的商业数据采集系统，基于 Playwright 与异步技术的方案，都能提供一个坚实、现代的起点。接下来，我会带你从原理到实战，一步步拆解如何搭建这样一个高效能的爬虫引擎。

2. 核心工具选型：Playwright 与异步框架深度解析

2.1 为什么是 Playwright，而不是 Selenium 或 Puppeteer？

在浏览器自动化领域，Selenium 是老兵，Puppeteer 是 Chrome 的亲儿子，而 Playwright 可以看作是集大成者的“新生代”。我的选型主要基于下面几个实战中痛点的解决：

第一，多浏览器内核的无缝支持。Playwright 为 Chromium、Firefox 和 WebKit（Safari 的引擎）都提供了高度一致的 API。这意味着你写一份脚本，可以几乎不加修改地在三种主流浏览器引擎上运行。对于需要验证跨浏览器兼容性的爬取任务（比如确保某个交互动作在 Safari 上也能正确触发数据加载），这一点至关重要。Selenium 虽然也支持多浏览器，但不同驱动的行为差异和配置复杂度要高得多。

第二，自动等待与稳健的选择器。这是让我放弃 Selenium 的关键原因之一。Playwright 的操作（如click,fill）内置了智能等待：它会等待元素可操作（可见、启用、稳定）后再执行，并设置合理的超时。你很少需要再写WebDriverWait和一堆expected_conditions。它的选择器引擎也非常强大，支持 CSS、XPath，还有专属的text=、has=等语义化选择器，定位元素更直观准确。

第三，强大的网络请求拦截与模拟能力。Playwright 允许你在页面加载前后，监听、修改或阻断任何网络请求。这对于爬虫来说是个“神器”。比如，你可以直接拦截页面加载时发出的 Ajax API 请求，拿到结构清晰的 JSON 数据，完全绕过渲染环节，效率极高。你也可以修改请求头来模拟不同设备，或者阻断图片、样式表请求以加快爬取速度。

第四，丰富的设备与上下文模拟。通过browser.new_context()可以轻松创建独立的浏览器上下文，每个上下文拥有独立的 cookies、本地存储，相当于一个全新的会话。你可以很方便地模拟手机设备（包括屏幕尺寸、UA），这对于爬取移动端适配的网站非常有用。

注意：虽然 Puppeteer 在 Chrome 自动化上同样优秀，但 Playwright 的多浏览器支持和对标 Puppeteer 的 API 设计（甚至更友好），使其在通用性和未来生态上更具优势。对于 Python 开发者，Playwright 的 Python 绑定 (playwright-python) 由官方维护，更新及时，文档完善。

2.2 异步（asyncio）在爬虫中的核心价值

同步爬虫是“做完一件事，再做下一件”。当遇到网络 I/O 等待（如下载页面）时，整个程序就卡住了，CPU 闲置。异步爬虫则是在等待一个任务时，立刻去执行其他可执行的任务。

在浏览器自动化爬虫中，异步的价值被放大：

1. 并行控制多个浏览器页面：一个异步事件循环可以同时驱动几十个甚至上百个Page对象进行数据抓取，而每个页面在等待元素加载或网络响应时，不会阻塞其他页面。
2. 高效处理非阻塞操作：Playwright 的绝大多数 API 都是异步的（如page.goto(),page.click()）。使用asyncio可以自然地调用这些 API，让等待时间被充分利用。
3. 资源利用率高：相比用threading实现并发，asyncio是单线程内进行任务调度，避免了线程切换的开销和复杂的锁机制，更轻量，更适合 I/O 密集型任务。

简单类比：同步爬虫像只有一个收银台的超市，顾客排长队；而异步爬虫像有多个收银台，一个顾客在装袋时，收银员可以给下一个顾客扫码，整体吞吐量大幅提升。

3. 环境搭建与核心 API 初探

3.1 一步到位的环境安装

别被复杂的安装吓到，Playwright 的安装已经非常简化。我推荐使用pip并在虚拟环境中进行。

# 1. 创建并进入虚拟环境（可选但强烈推荐） python -m venv playwright-env source playwright-env/bin/activate # Linux/macOS # playwright-env\Scripts\activate # Windows # 2. 安装 playwright 的 Python 包 pip install playwright # 3. 安装 Playwright 所需的浏览器内核 playwright install

第三步playwright install会下载 Chromium、Firefox 和 WebKit 的可执行文件到本地缓存。这些浏览器是专门为自动化定制的版本，不会干扰你系统上已安装的浏览器。

实操心得：在国内网络环境下，直接安装浏览器可能会很慢或失败。有两个解决办法：一是使用镜像源，设置环境变量PLAYWRIGHT_DOWNLOAD_HOST=https://npmmirror.com/mirrors/playwright后再执行安装命令；二是手动下载浏览器包，具体方法可查阅 Playwright 中文社区文档。我通常选择在晚上网络好的时候跑安装命令。

3.2 你的第一个异步 Playwright 脚本

让我们从一个最简单的例子开始，感受一下异步上下文管理器和 Playwright API 的简洁。

import asyncio from playwright.async_api import async_playwright async def main(): # 使用 async_playwright 上下文管理器启动 Playwright async with async_playwright() as p: # 启动一个 Chromium 浏览器实例，headless=False 表示显示界面 browser = await p.chromium.launch(headless=False) # 创建一个新的浏览器上下文（独立会话） context = await browser.new_context() # 在新上下文中打开一个页面 page = await context.new_page() # 导航到目标网站 await page.goto('https://books.toscrape.com/') # 等待页面中某个关键元素出现，确保加载完成 await page.wait_for_selector('.product_pod') # 执行一个简单的操作：点击第一本书的链接 await page.click('.product_pod h3 a') # 等待新页面加载 await page.wait_for_selector('.product_main') # 获取新页面的标题 title = await page.text_content('.product_main h1') print(f'书名: {title}') # 获取价格 price = await page.text_content('.price_color') print(f'价格: {price}') # 截图保存，用于调试 await page.screenshot(path='book_detail.png') # 关闭浏览器 await browser.close() # 运行异步主函数 asyncio.run(main())

这段代码做了以下几件事：

1. 以非无头模式（显示浏览器界面）启动 Chromium。
2. 访问一个经典的爬虫练习网站。
3. 等待商品列表加载，然后点击第一本书。
4. 跳转到详情页后，提取书名和价格。
5. 截图并关闭浏览器。

运行它，你会看到一个浏览器窗口自动打开并执行所有操作。这就是自动化最直观的体现。

3.3 核心 API 对象关系与生命周期

理解下面三个核心对象的关系，是写出健壮脚本的基础：

1. Browser：代表一个浏览器实例。通过await playwright.chromium.launch()创建。一个Browser可以包含多个独立的BrowserContext。
2. BrowserContext：浏览器上下文。这相当于一个独立的“隐身会话”，拥有独立的 cookies、本地存储和缓存。通过await browser.new_context()创建。这是实现多账号隔离或并行任务的关键。一个Context可以包含多个Page。
3. Page：代表一个标签页。我们绝大部分操作都在Page对象上完成，如导航、点击、提取数据。通过await context.new_page()创建。

它们的关系是：Playwright->Browser->BrowserContext->Page。 一个典型的生命周期是：启动 Playwright -> 启动 Browser -> （可选）创建多个 Context -> 在每个 Context 中创建 Page -> 在 Page 上操作 -> 关闭 Page -> 关闭 Context -> 关闭 Browser -> 停止 Playwright。

使用async with语句可以自动管理这些资源的关闭，是更推荐的写法。

4. 实战进阶：构建高效异步爬虫引擎

4.1 实现可靠的页面导航与等待策略

直接page.goto()然后立刻抓取，十有八九会失败，因为页面还没加载完。Playwright 提供了多种等待机制，正确使用它们是稳定的前提。

page.goto()的wait_until参数：

# 默认是 'load'，等待 load 事件触发，但此时动态内容可能还没加载 await page.goto(url) # 更推荐使用 'domcontentloaded'，DOM 解析完成即继续，更快 await page.goto(url, wait_until='domcontentloaded') # 或者使用 'networkidle'，等待网络基本空闲（500ms内无超过2个网络请求），适合SPA await page.goto(url, wait_until='networkidle')

对于现代网站，我通常先用'domcontentloaded'快速获取基础 HTML，再用显式等待定位动态加载的元素。

显式等待：page.wait_for_selector()及其家族： 这是最常用的等待方式，会等待直到指定的选择器匹配的元素出现在 DOM 中。

# 等待一个具有特定类名的元素出现 await page.wait_for_selector('.product-list', state='visible') # state可以是 'attached', 'visible', 'hidden'等 # 等待一个包含特定文本的元素出现 await page.wait_for_selector('text=下一页') # 等待导航完成（例如点击链接后） async with page.expect_navigation(): await page.click('a#next-page') # 或者 await page.click('a#next-page') await page.wait_for_url('**/page-2') # 等待 URL 变成特定模式

超时设置与重试： 所有等待方法都有timeout参数（默认 30 秒）。对于不稳定的网站，可以适当延长，并配合重试逻辑。

import asyncio from playwright.async_api import TimeoutError as PlaywrightTimeoutError async def safe_goto(page, url, retries=3): for attempt in range(retries): try: await page.goto(url, wait_until='networkidle', timeout=60000) return True except PlaywrightTimeoutError: print(f'第 {attempt + 1} 次导航超时，重试...') await asyncio.sleep(2 ** attempt) # 指数退避 print(f'导航失败: {url}') return False

4.2 高级交互：应对复杂页面与反爬

处理下拉框、文件上传：

# 选择下拉框选项 await page.select_option('select#country', value='CN') # 通过 value await page.select_option('select#country', label='中国') # 通过显示文本 # 文件上传 await page.set_input_files('input[type="file"]', 'path/to/my/file.pdf')

执行 JavaScript：page.evaluate()是在页面上下文中执行 JavaScript 的利器，可以获取 Playwright API 不方便直接拿到的东西，或者进行复杂操作。

# 获取 window 对象里的数据 data = await page.evaluate('() => window.__INITIAL_STATE__') print(data) # 滚动到页面底部 await page.evaluate('window.scrollTo(0, document.body.scrollHeight)') # 修改页面样式（比如隐藏干扰元素） await page.evaluate('''() => { document.querySelector('.ad-banner').style.display = 'none'; }''')

模拟人类行为：降低被检测风险： 过于精准和快速的点击容易被识别为机器人。可以引入随机延迟和更自然的移动轨迹。

import random from playwright.async_api import Page async def human_like_click(page: Page, selector: str): element = await page.wait_for_selector(selector) box = await element.bounding_box() # 模拟鼠标移动：先移动到大致区域，再移动到精确位置 await page.mouse.move( box['x'] + box['width'] * random.uniform(0.2, 0.8), box['y'] + box['height'] * random.uniform(0.2, 0.8), steps=random.randint(10, 30) # 移动步数，模拟曲线 ) await page.wait_for_timeout(random.randint(50, 300)) # 随机停顿 await element.click()

处理弹窗与对话框：

# 监听并接受确认对话框 page.on('dialog', lambda dialog: dialog.accept()) # 监听并处理新窗口（标签页）打开 async with page.expect_popup() as popup_info: await page.click('a[target="_blank"]') new_page = await popup_info.value # 在新页面操作 await new_page.close()

4.3 数据提取的艺术：从元素到结构化数据

Playwright 提供了多种提取页面数据的方法：

text_content()vsinner_text()：

• element.text_content()：获取元素内所有文本节点的内容，包括被 CSS 隐藏的文本。
• element.inner_text()：获取渲染后的文本，会忽略隐藏元素的文本，并尊重 CSS 的换行。通常inner_text()更符合“所见即所得”。

批量提取与列表推导：

# 提取所有商品标题 book_titles = await page.locator('.product_pod h3 a').all_inner_texts() # 提取所有商品价格，并清洗数据 book_prices = [] price_elements = await page.locator('.price_color').all() for element in price_elements: text = await element.inner_text() # 清洗价格，例如去除货币符号并转为浮点数 price = float(text.replace('£', '').replace('$', '').strip()) book_prices.append(price) # 更复杂的结构化提取：使用 evaluate 一次性提取 books_data = await page.evaluate('''() => { const items = []; document.querySelectorAll('.product_pod').forEach(pod => { items.push({ title: pod.querySelector('h3 a').title, price: pod.querySelector('.price_color').innerText, link: pod.querySelector('h3 a').href }); }); return items; }''')

page.locator()：强大的定位器 API：locator是 Playwright 推荐的核心查询 API，它返回一个Locator对象，支持链式调用和更丰富的操作。

# 定位第一个匹配的元素 buy_button = page.locator('button:has-text("购买")').first # 定位第 N 个匹配的元素 third_item = page.locator('.list-item').nth(2) # 基于其他元素进行定位（父级、相邻） price = page.locator('.product').locator('.price') # 在.product内找.price next_sibling = page.locator('h1').locator('xpath=following-sibling::p')

4.4 网络请求拦截：爬虫的“捷径”

这是 Playwright 相对于传统爬虫工具的杀手级功能。很多网站的数据是通过 XHR/Fetch 请求加载的 JSON，直接拦截这些请求比解析渲染后的 HTML 更高效、更稳定。

监听所有请求与响应：

async def log_request(request): if 'api/data' in request.url: print(f'>> 请求: {request.method} {request.url}') # 可以在这里修改请求头 # headers = request.headers # headers['User-Agent'] = 'My Custom UA' # ... 但修改需要通过 route.continue_ 实现 async def log_response(response): if 'api/data' in response.url and response.status == 200: print(f'<< 响应: {response.status} {response.url}') # 可以直接获取响应体（JSON） try: json_data = await response.json() print(f' 数据: {json_data}') # 这里可以将 json_data 保存或处理 except: text_data = await response.text() print(f' 文本: {text_data[:200]}...') page.on('request', log_request) page.on('response', log_response) await page.goto('https://example.com')

拦截并修改请求，或直接返回模拟数据：

# 拦截特定模式的请求，并阻止其继续发送，直接返回自定义响应 await page.route('**/api/v1/ads/*', lambda route: route.abort()) # 阻断广告请求 # 拦截请求并修改请求头 async def modify_header(route, request): headers = request.headers headers['x-custom-token'] = 'my-secret-token' await route.continue_(headers=headers) await page.route('**/api/**', modify_header) # 拦截请求并返回本地模拟数据（Mock） async def mock_response(route, request): await route.fulfill( status=200, content_type='application/json', body=json.dumps({'mock': 'data'}) ) await page.route('**/api/user/profile', mock_response)

通过拦截网络请求，你可以在数据到达浏览器渲染引擎之前就将其捕获，极大减少了不必要的资源加载和解析开销。

5. 异步并发架构设计与性能优化

5.1 基础并发：使用 asyncio.gather 控制多个页面

最简单的并发模式是同时打开多个页面（Page）进行抓取。但要注意，所有页面共享同一个浏览器上下文（BrowserContext）的资源（如内存、CPU），数量过多会导致浏览器卡顿甚至崩溃。

import asyncio from playwright.async_api import async_playwright async def scrape_one_page(browser, url): """一个独立的抓取任务""" page = await browser.new_page() try: await page.goto(url, wait_until='domcontentloaded') # ... 你的抓取逻辑 ... title = await page.title() return {'url': url, 'title': title} finally: await page.close() # 务必关闭页面释放资源 async def main(): urls = ['https://example.com/1', 'https://example.com/2', 'https://example.com/3'] async with async_playwright() as p: browser = await p.chromium.launch(headless=True) # 创建任务列表 tasks = [scrape_one_page(browser, url) for url in urls] # 并发执行所有任务 results = await asyncio.gather(*tasks, return_exceptions=True) # 处理结果，注意处理异常 for result in results: if isinstance(result, Exception): print(f'任务失败: {result}') else: print(result) await browser.close() asyncio.run(main())

5.2 高级并发：使用信号量（Semaphore）控制并发度

直接使用gather启动大量任务可能会瞬间创建过多页面，导致资源耗尽。我们需要一个“阀门”来控制同时进行的任务数量，这就是asyncio.Semaphore。

import asyncio from playwright.async_api import async_playwright class ConcurrentScraper: def __init__(self, max_concurrent=5): self.semaphore = asyncio.Semaphore(max_concurrent) async def scrape_with_limit(self, browser, url): """使用信号量限制并发""" async with self.semaphore: # 只有拿到信号量才能进入 return await self._scrape_page(browser, url) async def _scrape_page(self, browser, url): page = await browser.new_page() try: await page.goto(url, timeout=60000) data = await page.evaluate('() => document.title') return {'url': url, 'data': data} except Exception as e: return {'url': url, 'error': str(e)} finally: await page.close() async def main(): urls = [f'https://example.com/{i}' for i in range(100)] # 100个URL scraper = ConcurrentScraper(max_concurrent=10) # 最大并发10个页面 async with async_playwright() as p: browser = await p.chromium.launch(headless=True) tasks = [scraper.scrape_with_limit(browser, url) for url in urls] # 使用 asyncio.as_completed 来实时获取完成的任务 for coro in asyncio.as_completed(tasks): result = await coro print(f'完成: {result["url"]}') await browser.close()

5.3 多浏览器上下文隔离与资源管理

当任务需要完全独立的会话（如不同的登录态）时，或者为了更好的稳定性，应该为每组任务创建独立的BrowserContext。

async def worker(context_id, urls, result_queue): """一个工作线程，拥有自己独立的浏览器上下文""" async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() # 独立上下文 for url in urls: page = await context.new_page() try: await page.goto(url) # ... 抓取逻辑 ... await result_queue.put({'context': context_id, 'url': url, 'success': True}) except Exception as e: await result_queue.put({'context': context_id, 'url': url, 'error': str(e)}) finally: await page.close() await context.close() await browser.close() async def main(): # 将URL分给多个worker all_urls = [...] # 大量URL chunk_size = len(all_urls) // 3 url_chunks = [all_urls[i:i+chunk_size] for i in range(0, len(all_urls), chunk_size)] result_queue = asyncio.Queue() tasks = [] for i, chunk in enumerate(url_chunks): task = asyncio.create_task(worker(i, chunk, result_queue)) tasks.append(task) # 收集结果 results = [] while not result_queue.empty() or any(not t.done() for t in tasks): try: result = await asyncio.wait_for(result_queue.get(), timeout=1.0) results.append(result) except asyncio.TimeoutError: pass await asyncio.gather(*tasks, return_exceptions=True) print(f'共完成 {len(results)} 个任务')

5.4 性能优化与稳定性技巧

1. 启用无头模式与禁用不必要的资源加载：生产环境务必使用headless=True。可以进一步通过上下文选项禁用图片、样式表等，大幅提升加载速度。

   browser = await p.chromium.launch(headless=True) context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/5.0 ...', # 忽略 HTTPS 错误（对某些测试环境有用） ignore_https_errors=True, # 拦截并阻断不必要的资源 bypass_csp=True # 有时需要绕过内容安全策略 ) # 或者通过路由全局拦截 await context.route("**/*.{png,jpg,jpeg,svg,gif,webp}", lambda route: route.abort()) await context.route("**/*.css", lambda route: route.abort())

2. 复用浏览器实例与上下文：避免为每个任务都启动/关闭浏览器，开销巨大。应该在程序开始时启动浏览器，结束时关闭。

3. 合理设置超时与重试：对goto,click,wait_for_selector等操作设置合理的timeout，并配合指数退避算法进行重试。

4. 监控与日志：使用page.on('console'),page.on('pageerror')监听页面错误和日志，便于调试。记录每个任务的开始、结束时间和状态，有助于分析性能瓶颈。

5. 资源泄漏防范：确保Page和BrowserContext在使用后被正确关闭（close()）。使用try...finally块或异步上下文管理器来保证。

6. 工程化实践：从脚本到可维护的项目

6.1 项目结构组织

一个可维护的爬虫项目不应该把所有代码堆在一个文件里。建议按功能模块拆分：

modern_scraper/ ├── config.py # 配置文件（并发数、超时、URL列表等） ├── main.py # 主程序入口 ├── core/ │ ├── browser.py # 浏览器池管理、启动/关闭逻辑 │ ├── scraper.py # 核心抓取逻辑类 │ └── utils.py # 工具函数（重试、日志、数据处理） ├── handlers/ │ ├── request_interceptor.py # 请求拦截处理器 │ └── data_parser.py # 数据解析器 ├── models/ │ └── item.py # 数据模型（Pydantic/Dataclass） ├── storage/ │ ├── json_writer.py # 数据存储（JSON, CSV, 数据库） │ └── db_client.py └── logs/ # 日志目录

6.2 配置管理与错误处理

使用配置文件（如config.yaml或config.py）来管理参数：

# config.py class Config: HEADLESS = True BROWSER_TYPE = "chromium" # or "firefox", "webkit" CONCURRENT_TASKS = 10 REQUEST_TIMEOUT = 60000 # ms RETRY_TIMES = 3 USER_AGENT = "Mozilla/5.0 ..." # ... 其他配置

实现一个健壮的错误处理与重试装饰器：

import asyncio import functools from loguru import logger # 推荐使用 loguru 进行日志管理 def async_retry(max_retries=3, delays=(1, 3, 5)): """异步重试装饰器""" def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_retries): try: return await func(*args, **kwargs) except Exception as e: last_exception = e if attempt < max_retries - 1: delay = delays[attempt] if attempt < len(delays) else delays[-1] logger.warning(f"函数 {func.__name__} 第{attempt+1}次失败: {e}, {delay}秒后重试") await asyncio.sleep(delay) else: logger.error(f"函数 {func.__name__} 重试{max_retries}次后仍失败") raise last_exception return wrapper return decorator # 使用示例 @async_retry(max_retries=3, delays=(1, 2, 4)) async def fetch_page_data(page, url): await page.goto(url, timeout=30000) # ... 抓取逻辑 ...

6.3 数据存储与任务队列

对于大规模爬虫，需要考虑如何持久化存储和任务调度。

轻量级方案：SQLite + 内存队列

import aiosqlite import asyncio from asyncio import Queue class TaskQueue: def __init__(self): self.queue = Queue() self.processed_urls = set() # 简单的去重集合，生产环境应用布隆过滤器或Redis async def add_url(self, url): if url not in self.processed_urls: await self.queue.put(url) self.processed_urls.add(url) async def get_url(self): return await self.queue.get() async def init_database(): async with aiosqlite.connect('scraped_data.db') as db: await db.execute(''' CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT UNIQUE, title TEXT, price REAL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') await db.commit() async def save_to_db(data): async with aiosqlite.connect('scraped_data.db') as db: await db.execute(''' INSERT OR IGNORE INTO products (url, title, price) VALUES (?, ?, ?) ''', (data['url'], data['title'], data['price'])) await db.commit()

分布式方案雏形：使用消息队列（如 Redis）对于需要分布式部署的爬虫，可以使用aioredis来管理任务队列和去重集合。

# 示例：使用 Redis 作为任务队列 import aioredis async def push_task_to_redis(redis, queue_name, url): await redis.lpush(queue_name, url) async def pop_task_from_redis(redis, queue_name): url = await redis.rpop(queue_name) return url # 使用 Redis Set 进行去重 async def is_duplicate(redis, set_name, url): return await redis.sismember(set_name, url) async def mark_as_processed(redis, set_name, url): await redis.sadd(set_name, url)

7. 常见问题排查与调试技巧实录

即使方案再完善，爬虫在实际运行中也会遇到各种稀奇古怪的问题。这里记录了我踩过的一些坑和解决方法。

7.1 元素定位失败：Selector 总是找不到

这是最常见的问题。

可能原因及排查：

1. 页面未加载完成：增加等待时间，或使用更可靠的等待条件（如wait_for_selector的state='visible'）。
2. 元素在 iframe 内：需要先切换到 iframe 上下文。

   frame = page.frame('frame-name') # 通过 name 或 URL 定位 # 或者 frame = page.frame_locator('iframe#id').content_frame await frame.click('button') # 在 frame 内操作

3. 选择器写错了/动态生成：使用 Playwright 的录制工具来生成可靠的选择器。在终端运行playwright codegen https://example.com，会打开一个浏览器和代码生成器，你的操作会被实时转换成代码，其中包含它推荐的选择器。
4. 页面有 Shadow DOM：需要使用::shadow选择器或page.evaluate穿透 Shadow Root。

   # 假设有一个自定义元素 <my-element> shadow_text = await page.locator('my-element::shadow .inner-element').text_content()

7.2 页面卡死或无响应

可能原因及排查：

1. JavaScript 无限循环或错误：监听pageerror事件。

   page.on('pageerror', lambda err: print(f'页面JS错误: {err}'))

2. 内存泄漏：检查是否没有及时关闭Page和BrowserContext。使用browser.contexts()查看当前存在的上下文数量。
3. 并发过高：减少Semaphore的并发数。监控系统内存和 CPU 使用情况。
4. 被网站检测并反制：尝试添加更真实的user_agent，启用viewport，在上下文中添加locale、timezone_id等属性，让浏览器指纹更像真人。也可以尝试使用playwright-stealth等插件来规避检测。

7.3 异步任务意外结束或报错不明确

调试技巧：

1. 启用 Playwright 调试日志：在启动浏览器时设置环境变量DEBUG=pw:api，可以看到所有 Playwright API 的调用日志。

   DEBUG=pw:api python your_script.py

2. 使用asyncio.run()的正确姿势：确保主函数被正确包装。对于需要清理资源的复杂程序，建议使用asyncio.run(main())的变体。

   async def main(): # ... 你的逻辑 ... pass if __name__ == '__main__': asyncio.run(main())

3. 捕获并记录所有异常：使用asyncio.gather(*tasks, return_exceptions=True)可以防止一个任务崩溃导致整个程序退出，并收集所有异常信息。

7.4 性能瓶颈分析

如果爬虫速度不如预期，可以按以下步骤排查：

1. 网络是否是瓶颈？检查目标网站的响应速度，考虑使用代理池。
2. 浏览器渲染是否是瓶颈？尝试拦截并阻断不必要的资源（如图片、CSS、字体），或者直接拦截 API 请求获取数据，跳过渲染。
3. CPU/内存是否是瓶颈？使用系统监控工具（如htop,任务管理器）观察。减少并发数，或使用多台机器分布式爬取。
4. 代码逻辑是否有阻塞操作？确保没有在异步函数中使用同步的、耗时的 I/O 操作（如同步文件读写、网络请求）。如果必须使用，用asyncio.to_thread将其放到线程池中执行。

7.5 一份快速自查表

构建一个基于 Playwright 和异步技术的爬虫，就像给传统的爬虫脚本装上了“自动驾驶”和“多线程处理器”。它不仅能处理最复杂的现代网页，还能以极高的效率并发作业。从简单的页面操作到复杂的网络请求拦截，从单脚本到可工程化的项目结构，这套技术栈为你提供了全方位的解决方案。在实际项目中，最关键的是根据目标网站的特点，灵活组合使用这些工具和模式，并建立起完善的监控、日志和错误处理机制，这样才能保证爬虫长期稳定、高效地运行。