基于Playwright网络监听的高效数据采集方案：告别DOM解析，直击API源头

1. 项目概述：为什么我们需要更“聪明”的网络层采集？

如果你写过爬虫，大概率经历过这种痛苦：页面元素千变万化，今天用BeautifulSoup写好的XPath，明天网站改个class名或者加个div嵌套，整个脚本就报废了。更别提那些动态加载的内容，你得像个侦探一样，在开发者工具的Network面板里大海捞针，找到那个关键的XHR请求，然后费劲地模拟它的参数和Headers。这种“页面解析”为主的爬虫，就像在沙地上建城堡，基础太不稳固。

最近我在做一个电商价格监控的项目，目标网站大量使用了前端渲染，商品列表、详情、评论都是通过异步接口（XHR/JSON）加载的。传统的Selenium或requests+BeautifulSoup组合在这里显得力不从心：要么速度慢得令人发指，要么根本无法获取到核心数据。直到我把目光投向了网络层——直接监听浏览器发出的每一个请求和响应。这就像从“在沙滩上捡贝壳”变成了“直接拦截运送贝壳的货车”，效率和数据完整性是天壤之别。

这次要分享的，就是基于Playwright的Response监听技术，构建一个稳定、高效的网络层采集方案。它的核心思想是：我们不和变幻莫测的DOM结构较劲，而是直击数据源头——网络请求。无论是XHR、Fetch还是JSONP，只要数据是通过网络传输的，我们就能在它抵达浏览器渲染引擎之前将其捕获、解析并落地存储。方案还集成了CSV导出和SQLite持久化，确保从采集到存储的链路完整、可靠。对于需要处理大量动态内容、追求数据稳定性和采集效率的开发者来说，这无疑是一把利器。

2. 核心思路与方案选型：为何是Playwright + Response监听？

在动手之前，我们需要理清思路：实现网络层采集有哪些路可走？为什么最终选择了Playwright的Response事件监听？

2.1 常见网络采集方案对比

1. 浏览器开发者工具手动分析：这是最原始的方法。打开F12，记录网络活动，找到数据接口，手动复制cURL命令，再用requests或httpx重放。优点是直接，缺点是极度低效、无法自动化、无法应对反爬（如签名参数）。
2. Mitmproxy等中间人代理：在客户端和服务器之间插入一个代理，所有流量都经过它，因此可以查看和修改任意请求/响应。功能强大，可以处理HTTPS。但配置相对复杂，需要安装证书，且对于纯浏览器环境的一些复杂交互（如WebSocket）支持不如直接集成在浏览器内的方案直观。
3. Selenium + Browsermob-Proxy：Selenium控制浏览器，Browsermob-Proxy作为代理拦截流量。这是一个经典组合，但架构稍显笨重，需要同时管理浏览器和代理服务，稳定性调试起来比较麻烦。
4. CDP (Chrome DevTools Protocol) 直接调用：通过WebSocket连接浏览器，直接发送CDP命令来获取网络请求、控制浏览器行为。功能最底层、最强大，但API较为复杂，需要自己处理很多底层细节。
5. Playwright / Puppeteer 的Request/Response事件监听：这是本文选择的方案。Playwright（以及Puppeteer）在控制浏览器的同时，提供了非常简洁的API来监听网络活动。你只需要几行代码，就能为页面绑定一个回调函数，每当有网络响应返回时，这个函数就会被触发，你可以直接访问到完整的Response对象。

2.2 为什么选择Playwright？

在Puppeteer（主要驱动Chrome）和Playwright之间，我选择了后者，原因如下：

• 多浏览器支持：Playwright原生支持Chromium、Firefox和WebKit（Safari内核）。这意味着你可以用同一套脚本测试网站在不同浏览器下的行为，对于需要绕过某些基于浏览器指纹的反爬策略很有帮助。
• 更现代的API与更好的性能：Playwright的API设计被认为更优雅，并且在许多场景下（如等待元素、处理弹窗）提供了更便捷的方法。其底层通信效率也较高。
• 强大的自动化能力：除了网络监听，Playwright在模拟用户操作（点击、输入、滚动、拖拽）方面非常出色，能够轻松处理登录、验证码滑块（通过坐标模拟）等复杂交互，为触发数据接口创造了条件。
• 活跃的社区与微软背书：作为微软的开源项目，Playwright发展迅速，社区活跃，遇到问题更容易找到解决方案。

核心优势总结：Playwright的Response监听方案，将浏览器自动化与网络监听无缝集成。你不需要额外配置代理，所有逻辑都在一个脚本内完成。它既能像真实用户一样操作页面、触发数据请求，又能像狙击手一样精准捕获返回的数据包，实现了“行为模拟”与“数据抓取”的完美统一。

3. 环境搭建与核心API解析

工欲善其事，必先利其器。我们先来把环境和核心工具搞清楚。

3.1 环境准备与Playwright安装

首先确保你有一个Python环境（3.7+）。建议使用虚拟环境来管理依赖。

# 创建并激活虚拟环境（可选，但推荐） python -m venv playwright-env # Windows: playwright-env\Scripts\activate # macOS/Linux: source playwright-env/bin/activate # 安装Playwright的Python库 pip install playwright # 安装Playwright所需的浏览器内核（Chromium, Firefox, WebKit） playwright install

注意：playwright install这一步会下载浏览器二进制文件，体积较大（几百MB），请确保网络通畅。如果只想安装Chromium，可以使用playwright install chromium。

3.2 核心API：page.on(“response”)

这是整个方案的灵魂。page.on(“response”, callback)方法允许你为页面对象注册一个事件监听器。每当页面收到任何一个网络响应（包括文档、样式表、脚本、图片、XHR/Fetch请求等），你注册的回调函数就会被调用。

回调函数会接收一个Response对象作为参数，这个对象包含了关于这个响应的所有信息：

• response.url: 请求的URL。这是我们过滤目标接口的关键。
• response.status: HTTP状态码（如200, 404, 500）。
• response.headers: 响应头信息。
• response.request: 对应的Request对象，可以获取请求方法、请求头、POST数据等。
• response.ok: 布尔值，表示请求是否成功（状态码200-299）。
• response.text(): 异步方法，获取响应体的文本内容（如HTML, JSON字符串）。
• response.json(): 异步方法，如果响应内容是JSON，直接解析为Python字典或列表。
• response.body(): 异步方法，获取原始的二进制响应体。

一个最简单的监听示例如下：

import asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: browser = await p.chromium.launch(headless=False) # 非无头模式，方便观察 page = await browser.new_page() # 定义回调函数 async def on_response(response): # 打印所有响应的URL print(f”URL: {response.url}, Status: {response.status}“) # 注册监听器 page.on(“response”, on_response) # 导航到目标页面，触发网络请求 await page.goto(“https://httpbin.org/json”) await page.wait_for_timeout(3000) # 等待3秒，确保所有异步请求完成 await browser.close() asyncio.run(main())

运行这段代码，你会看到控制台打印出访问https://httpbin.org/json时产生的所有请求（主文档、可能有的favicon.ico等）及其状态码。但这显然太“吵”了，我们需要进行过滤。

3.3 精准过滤：只捕获我们关心的数据接口

在实际项目中，一个页面可能会产生数十甚至上百个请求。我们只关心那些携带业务数据（通常是JSON格式）的XHR或Fetch请求。这就需要我们在回调函数里添加过滤逻辑。

过滤策略通常基于URL和响应内容类型：

1. URL关键词过滤：如果目标网站的API有规律可循，比如都包含/api/、/graphql、/data等路径，这是最直接高效的过滤方式。
2. 响应头Content-Type过滤：检查response.headers.get(‘content-type’)是否包含application/json。这是判断响应体是否为JSON的可靠方法。
3. 请求方法过滤：通常获取数据的API使用GET或POST方法，可以通过response.request.method来判断。
4. 综合判断：最稳妥的方式是结合以上几点。

优化后的回调函数示例：

async def on_response(response): url = response.url content_type = response.headers.get(‘content-type’, ‘’) # 过滤条件：URL包含‘api’，且响应内容是JSON if ‘api’ in url and ‘application/json’ in content_type: print(f”捕获到API: {url}“) try: # 尝试解析JSON json_data = await response.json() print(f”数据样例: {json.dumps(json_data, indent=2)[:200]}…“) # 只打印前200字符 # 这里可以调用处理函数，将json_data保存起来 await process_data(json_data, url) except Exception as e: # 可能不是有效的JSON，或者网络错误 print(f”解析JSON失败 {url}: {e}“)

实操心得：过滤条件不要一开始就写得太死。建议在开发阶段，先宽松地打印出所有application/json的响应，观察目标网站API的URL模式和数据结构，然后再逐步收紧过滤条件，避免漏掉关键数据。

4. 项目实战：构建一个完整的网络层采集系统

理论讲完了，我们来搭建一个完整的、可复用的采集系统。这个系统要完成以下任务：

1. 监听并过滤目标数据接口。
2. 解析JSON数据，并提取出结构化的字段。
3. 将数据同时保存到CSV文件（便于快速查看和Excel分析）和SQLite数据库（便于持久化存储和复杂查询）。
4. 具备良好的扩展性，方便适配不同的网站。

4.1 系统架构与核心类设计

我们将代码组织得清晰一些，创建一个名为NetworkSpider的类。

import asyncio import json import csv import sqlite3 from datetime import datetime from typing import Dict, List, Any, Optional, Callable from playwright.async_api import Page, Response import aiofiles # 用于异步文件写入，需安装：pip install aiofiles class NetworkSpider: “”“基于Playwright Response监听的网络爬虫”“” def __init__(self, db_path: str = “spider_data.db”, csv_path: str = “data.csv”): self.db_path = db_path self.csv_path = csv_path self._init_database() # 初始化数据库表 self._init_csv_file() # 初始化CSV文件表头 self.data_buffer: List[Dict] = [] # 数据缓冲区，积累一定量后批量写入 self.buffer_size = 50 # 缓冲区大小 def _init_database(self): “”“创建SQLite数据库和表”“” conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 创建一个通用的数据表，包含原始JSON、URL、采集时间等。 # 实际项目中，你可能需要根据数据结构创建更具体的表。 cursor.execute(‘‘‘ CREATE TABLE IF NOT EXISTS crawled_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT NOT NULL, raw_data TEXT, -- 存储原始的JSON字符串 timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ‘‘’) # 可以再创建一个表来存储解析后的结构化数据 cursor.execute(‘‘‘ CREATE TABLE IF NOT EXISTS parsed_products ( -- 示例：产品数据表 id INTEGER PRIMARY KEY AUTOINCREMENT, product_id TEXT, name TEXT, price REAL, url TEXT, source_url TEXT, crawled_at DATETIME ) ‘‘’) conn.commit() conn.close() def _init_csv_file(self): “”“初始化CSV文件，写入表头”“” # 这里定义CSV的列。你需要根据实际要保存的数据结构来定义。 csv_headers = [“product_id”, “name”, “price”, “url”, “source_url”, “crawled_at”] try: with open(self.csv_path, mode=‘x’, newline=‘’, encoding=‘utf-8-sig’) as f: # ‘x’模式表示文件不存在才创建 writer = csv.DictWriter(f, fieldnames=csv_headers) writer.writeheader() except FileExistsError: # 文件已存在，跳过 pass async def _on_response_handler(self, response: Response): “”“响应事件的核心处理函数”“” url = response.url content_type = response.headers.get(‘content-type’, ‘’).lower() # 1. 定义你的过滤规则 is_target_api = ( ‘/api/products’ in url or # 示例：URL包含特定路径 ‘/graphql’ in url # 示例：GraphQL接口 ) and ‘application/json’ in content_type if not is_target_api: return print(f”[捕获] {url}“) try: # 2. 获取JSON数据 json_data = await response.json() # 3. 处理数据（解析、清洗、存储） await self._process_captured_data(json_data, url) except Exception as e: print(f”[错误] 处理响应失败 {url}: {e}“) async def _process_captured_data(self, raw_data: Any, source_url: str): “”“处理捕获到的原始数据。 这里需要你根据目标网站返回的JSON结构，编写具体的解析逻辑。 这是一个示例，假设raw_data是一个包含产品列表的字典。 ”“” # 示例：假设接口返回 {“products”: [{…}, {…}]} if isinstance(raw_data, dict) and ‘products’ in raw_data: products = raw_data[‘products’] elif isinstance(raw_data, list): products = raw_data else: # 如果不是预期的结构，可以选择将原始数据存入数据库 await self._save_raw_data(raw_data, source_url) return parsed_items = [] for item in products: # 提取字段，这里需要你根据实际数据结构调整 parsed_item = { “product_id”: item.get(“id”), “name”: item.get(“name”, “”).strip(), “price”: float(item.get(“price”, 0)) if item.get(“price”) else None, “url”: item.get(“detailUrl”), “source_url”: source_url, “crawled_at”: datetime.now().isoformat() } # 过滤掉完全无效的数据（例如没有ID和名称） if parsed_item[“product_id”] and parsed_item[“name”]: parsed_items.append(parsed_item) # 将解析后的数据加入缓冲区 self.data_buffer.extend(parsed_items) print(f”[解析] 从 {source_url} 解析出 {len(parsed_items)} 条有效产品数据。缓冲区累计: {len(self.data_buffer)} 条”) # 如果缓冲区满了，则批量写入 if len(self.data_buffer) >= self.buffer_size: await self._flush_buffer_to_storage() async def _save_raw_data(self, raw_data: Any, source_url: str): “”“将原始JSON数据保存到数据库的通用表中”“” conn = sqlite3.connect(self.db_path) cursor = conn.cursor() try: cursor.execute( “INSERT INTO crawled_data (url, raw_data) VALUES (?, ?)”, (source_url, json.dumps(raw_data, ensure_ascii=False)) ) conn.commit() except Exception as e: print(f”[错误] 保存原始数据失败: {e}“) conn.rollback() finally: conn.close() async def _flush_buffer_to_storage(self): “”“将缓冲区中的数据批量写入CSV和SQLite”“” if not self.data_buffer: return items_to_save = self.data_buffer[:] self.data_buffer = [] # 清空缓冲区 # 1. 批量写入CSV (异步写入，避免阻塞) async with aiofiles.open(self.csv_path, mode=‘a’, newline=‘’, encoding=‘utf-8-sig’) as f: writer = csv.DictWriter(f, fieldnames=[“product_id”, “name”, “price”, “url”, “source_url”, “crawled_at”]) # aiofiles不支持直接的DictWriter，需要手动构造行 for item in items_to_save: line = ‘,’.join(f’“{str(item[col]).replace(‘“’, ‘”“’)}”‘ for col in writer.fieldnames) + ‘\n’ await f.write(line) # 2. 批量写入SQLite conn = sqlite3.connect(self.db_path) cursor = conn.cursor() try: cursor.executemany(‘‘‘ INSERT INTO parsed_products (product_id, name, price, url, source_url, crawled_at) VALUES (:product_id, :name, :price, :url, :source_url, :crawled_at) ‘‘’, items_to_save) conn.commit() print(f”[存储] 成功批量写入 {len(items_to_save)} 条数据到数据库和CSV。”) except Exception as e: print(f”[错误] 批量写入数据库失败: {e}“) conn.rollback() # 写入失败，将数据放回缓冲区（简单处理，生产环境需更健壮） self.data_buffer.extend(items_to_save) finally: conn.close() async def crawl(self, start_url: str, page_actions: Optional[Callable[[Page], Any]] = None): “”“启动爬虫的主函数”“” async with async_playwright() as p: # 建议使用Chromium，稳定性较好 browser = await p.chromium.launch( headless=True, # 生产环境建议无头模式 args=[‘--disable-blink-features=AutomationControlled’] # 隐藏自动化特征 ) # 创建上下文，可以设置更仿真的User-Agent等 context = await browser.new_context( user_agent=‘Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 …’, viewport={‘width’: 1920, ‘height’: 1080} ) page = await context.new_page() # 注册响应监听器 page.on(“response”, self._on_response_handler) # 导航到起始页面 await page.goto(start_url, wait_until=“networkidle”) # 等待到网络空闲 # 执行自定义的页面操作（如点击“加载更多”、搜索、登录等） if page_actions: await page_actions(page) # 等待一段时间，确保所有异步请求完成。更优的做法是等待特定元素出现。 await page.wait_for_timeout(5000) # 等待5秒 # 最后，将缓冲区剩余数据写入存储 await self._flush_buffer_to_storage() await context.close() await browser.close()

4.2 主程序与自定义操作

现在，我们写一个主程序来使用这个爬虫类。假设我们要爬取一个虚构的电商网站https://example-shop.com/products，这个页面通过滚动会不断加载更多产品。

import asyncio from playwright.async_api import Page async def scroll_to_load_more(page: Page): “”“自定义页面操作：模拟滚动以触发分页加载”“” print(“开始模拟滚动加载…”) for i in range(3): # 假设滚动3次 # 滚动到页面底部 await page.evaluate(“window.scrollTo(0, document.body.scrollHeight)”) # 等待新内容加载。更佳实践是等待某个加载指示器出现或消失。 await page.wait_for_timeout(2000) # 等待2秒 print(f”已完成第 {i+1} 次滚动。”) async def main(): spider = NetworkSpider( db_path=“example_shop.db”, csv_path=“example_shop_products.csv” ) start_url = “https://example-shop.com/products” await spider.crawl(start_url, page_actions=scroll_to_load_more) print(“采集任务完成！”) if __name__ == “__main__”: asyncio.run(main())

运行这个脚本，它会打开浏览器（无头模式），访问目标页面，自动滚动三次。在此期间，所有符合过滤条件的XHR/JSON响应都会被捕获、解析，并分批存储到SQLite数据库和CSV文件中。

5. 高级技巧与稳定性优化

一个能用于生产环境的爬虫，绝不能只满足于“跑通”。下面分享几个提升稳定性、效率和隐蔽性的关键技巧。

5.1 请求过滤与性能优化

• 避免内存泄漏：我们的监听器回调会捕获大量响应。如果每个响应都立即进行await response.json()这类异步操作，并在回调里进行复杂的处理，可能会阻塞事件循环或消耗过多内存。我们的方案通过“缓冲区+批量写入”来缓解这个问题。此外，对于不关心的请求（如图片、CSS），应尽早return，减少不必要的处理。
• 精确URL匹配：使用re模块进行正则表达式匹配，比简单的in操作更精确。例如，re.match(r‘https://api\.example\.com/v1/products/\d+’, url)可以精准匹配产品详情接口。
• 按需解析：不是所有JSON响应都需要立即解析。可以先根据URL判断其重要性，只解析核心的数据接口，其他如配置信息、用户状态等接口可以先只存储URL或忽略。

5.2 反反爬策略集成

Playwright本身提供了一些绕过基础检测的功能，但面对复杂的反爬系统，还需要组合拳。

1. 伪装浏览器指纹：

   context = await browser.new_context( user_agent=‘你的User-Agent字符串’, viewport={‘width’: 1920, ‘height’: 1080}, locale=‘zh-CN’, timezone_id=‘Asia/Shanghai’, # 可以覆盖更多设备属性 device_scale_factor=1, has_touch=False, is_mobile=False, )

2. 使用代理IP：这是应对IP封锁最有效的手段之一。

   browser = await p.chromium.launch( headless=True, proxy={ “server”: “http://your-proxy-server:port”, “username”: “username”, # 如果需要认证 “password”: “password” } )

   重要提示：务必使用合法合规的代理服务，并遵守目标网站的robots.txt协议和速率限制。过度频繁的请求会给对方服务器带来压力，也可能导致你的IP或代理IP被永久封禁。

3. 随机化操作间隔：在page.wait_for_timeout()中使用随机延迟，模拟人类操作的不确定性。

   import random await page.wait_for_timeout(random.uniform(1000, 3000)) # 随机等待1-3秒

4. 处理验证码：Playwright可以截图，对于简单的图形验证码，可以截图后调用第三方OCR服务识别。对于复杂的滑块验证，可以通过page.mouse.move()和page.mouse.down()等API模拟拖动，但这需要精确计算滑块轨迹，难度较高。

5.3 错误处理与重试机制

网络请求充满不确定性，健壮的错误处理必不可少。

async def robust_goto(page, url, max_retries=3): “”“一个带重试的导航函数”“” for attempt in range(max_retries): try: response = await page.goto(url, wait_until=“networkidle”, timeout=30000) if response and response.ok: return response else: print(f”导航失败，状态码: {response.status if response else ‘无响应’}，第{attempt+1}次重试…”) except Exception as e: print(f”导航异常: {e}，第{attempt+1}次重试…”) await page.wait_for_timeout(2000 * (attempt + 1)) # 重试间隔递增 raise Exception(f”导航到 {url} 失败，已达最大重试次数 {max_retries}“) # 在crawl方法中使用 response = await robust_goto(page, start_url)

对于数据接口的监听，也可以在_on_response_handler中增加重试逻辑，特别是对response.json()的调用。

6. 数据处理、存储与后续分析

数据抓下来只是第一步，如何高效地存储和利用它们同样重要。

6.1 双存储策略：CSV与SQLite的优劣与协同

• CSV文件：

  • 优点：人类可读，可以用Excel、Numbers等软件直接打开查看和简单分析。结构简单，易于分享。对于一次性或小规模数据，非常方便。
  • 缺点：不适合存储嵌套的JSON结构（需要展平）。查询效率低，尤其是数据量大时。不支持事务，并发写入可能出错。
  • 在我们的方案中角色：快速查看和校验数据。每次运行脚本，都能立即得到一个可以打开的CSV文件，检查数据格式和内容是否正确。

• SQLite数据库：

  • 优点：是一个完整的、轻量级的关系型数据库。支持SQL查询，可以轻松地进行数据筛选、聚合、连接等复杂操作。支持事务，保证数据一致性。可以存储原始JSON文本和解析后的结构化数据，非常灵活。
  • 缺点：需要SQL知识才能有效利用。文件是二进制的，不能直接文本编辑。
  • 在我们的方案中角色：主存储和持久化。所有历史数据都保存在这里，便于后续的数据分析、去重、监控等。

协同工作流：开发调试阶段，多看看CSV。数据积累和分析阶段，使用SQLite配合Python的pandas或sqlite3库进行深入处理。

6.2 使用SQLite进行数据分析示例

假设我们已经运行爬虫几天，积累了一些产品价格数据。我们可以轻松地分析价格变化。

import sqlite3 import pandas as pd import matplotlib.pyplot as plt conn = sqlite3.connect(‘example_shop.db’) # 1. 使用pandas直接读取，进行数据分析 df = pd.read_sql_query(“SELECT * FROM parsed_products”, conn) print(df.head()) print(f”共采集到 {df[‘product_id’].nunique()} 种独特商品。”) # 2. 分析某个商品的价格历史 product_id = ‘12345’ price_history_df = pd.read_sql_query( “SELECT price, crawled_at FROM parsed_products WHERE product_id = ? ORDER BY crawled_at”, conn, params=(product_id,) ) price_history_df[‘crawled_at’] = pd.to_datetime(price_history_df[‘crawled_at’]) price_history_df.set_index(‘crawled_at’, inplace=True) # 绘制价格走势图 plt.figure(figsize=(12, 6)) plt.plot(price_history_df.index, price_history_df[‘price’], marker=‘o’) plt.title(f”Product {product_id} Price History”) plt.xlabel(‘Date’) plt.ylabel(‘Price’) plt.grid(True) plt.xticks(rotation=45) plt.tight_layout() plt.savefig(‘price_history.png’) plt.show() conn.close()

6.3 数据去重与增量更新

网络监听可能会捕获到重复的请求（比如页面刷新）。我们需要在存储时进行去重。

• 基于业务逻辑去重：在_process_captured_data方法中，在插入数据库前先查询是否已存在。例如，对于产品数据，可以根据product_id和crawled_at的时间精度（如按天）来判断是否重复。

  # 在插入parsed_products前检查 cursor.execute( “SELECT 1 FROM parsed_products WHERE product_id = ? AND DATE(crawled_at) = DATE(?)”, (item[‘product_id’], item[‘crawled_at’]) ) if cursor.fetchone(): print(f”产品 {item[‘product_id’]} 今日数据已存在，跳过。”) continue

• 使用INSERT OR IGNORE/REPLACE：SQLite支持INSERT OR IGNORE INTO ...或INSERT OR REPLACE INTO ...语法，但需要表有唯一约束（UNIQUE constraint）。我们可以修改表结构，为(product_id, crawled_at_date)添加唯一约束，然后使用INSERT OR IGNORE。

7. 常见问题排查与实战心得

在实际开发中，你肯定会遇到各种各样的问题。这里记录了一些典型问题的排查思路和我踩过的坑。

7.1 问题排查清单

7.2 实战心得与技巧

1. 从“宽”到“严”：刚开始写过滤规则时，不妨宽松一些，先把所有application/json的响应都打印出来，研究清楚目标网站的所有数据接口，再逐步精确过滤。避免一开始就写死规则，导致漏掉重要数据。
2. 原始数据是黄金：无论你的解析逻辑多么完善，一定要保留一份原始的、未加工的响应数据（就像我们设计的crawled_data表）。网站随时可能改版，有了原始数据，你就有能力在任何时候修复和升级你的解析器。
3. 异步编程的坑：Playwright是异步库。确保你的回调函数（on_response）是async的，并且在里面调用await response.json()。如果在非async函数中调用，或者忘记了await，会导致程序挂起或报错。
4. 等待的艺术：page.wait_for_timeout()是简单的等待，但不够智能。更好的做法是使用page.wait_for_selector()、page.wait_for_response()或page.wait_for_function()来等待特定的内容出现或请求完成。这能让脚本更稳定、更快。
5. 日志是你的眼睛：在关键步骤（开始监听、捕获到API、解析成功、存储成功、发生错误）都加上详细的日志输出。这能让你在脚本无声无息地失败时，快速定位问题所在。
6. 尊重与节制：最后也是最重要的，网络爬虫行走在法律的灰色地带。务必尊重网站的robots.txt协议，控制请求速率，避免对目标网站造成显著负担。将爬虫用于正当的学习、研究和合规的数据聚合，才是长久之道。