1. 项目概述为什么我们需要更“聪明”的网络层采集如果你写过爬虫大概率经历过这种痛苦页面元素千变万化今天用BeautifulSoup写好的XPath明天网站改个class名或者加个div嵌套整个脚本就报废了。更别提那些动态加载的内容你得像个侦探一样在开发者工具的Network面板里大海捞针找到那个关键的XHR请求然后费劲地模拟它的参数和Headers。这种“页面解析”为主的爬虫就像在沙地上建城堡基础太不稳固。最近我在做一个电商价格监控的项目目标网站大量使用了前端渲染商品列表、详情、评论都是通过异步接口XHR/JSON加载的。传统的Selenium或requestsBeautifulSoup组合在这里显得力不从心要么速度慢得令人发指要么根本无法获取到核心数据。直到我把目光投向了网络层——直接监听浏览器发出的每一个请求和响应。这就像从“在沙滩上捡贝壳”变成了“直接拦截运送贝壳的货车”效率和数据完整性是天壤之别。这次要分享的就是基于Playwright的Response监听技术构建一个稳定、高效的网络层采集方案。它的核心思想是我们不和变幻莫测的DOM结构较劲而是直击数据源头——网络请求。无论是XHR、Fetch还是JSONP只要数据是通过网络传输的我们就能在它抵达浏览器渲染引擎之前将其捕获、解析并落地存储。方案还集成了CSV导出和SQLite持久化确保从采集到存储的链路完整、可靠。对于需要处理大量动态内容、追求数据稳定性和采集效率的开发者来说这无疑是一把利器。2. 核心思路与方案选型为何是Playwright Response监听在动手之前我们需要理清思路实现网络层采集有哪些路可走为什么最终选择了Playwright的Response事件监听2.1 常见网络采集方案对比浏览器开发者工具手动分析这是最原始的方法。打开F12记录网络活动找到数据接口手动复制cURL命令再用requests或httpx重放。优点是直接缺点是极度低效、无法自动化、无法应对反爬如签名参数。Mitmproxy等中间人代理在客户端和服务器之间插入一个代理所有流量都经过它因此可以查看和修改任意请求/响应。功能强大可以处理HTTPS。但配置相对复杂需要安装证书且对于纯浏览器环境的一些复杂交互如WebSocket支持不如直接集成在浏览器内的方案直观。Selenium Browsermob-ProxySelenium控制浏览器Browsermob-Proxy作为代理拦截流量。这是一个经典组合但架构稍显笨重需要同时管理浏览器和代理服务稳定性调试起来比较麻烦。CDP (Chrome DevTools Protocol) 直接调用通过WebSocket连接浏览器直接发送CDP命令来获取网络请求、控制浏览器行为。功能最底层、最强大但API较为复杂需要自己处理很多底层细节。Playwright / Puppeteer 的Request/Response事件监听这是本文选择的方案。Playwright以及Puppeteer在控制浏览器的同时提供了非常简洁的API来监听网络活动。你只需要几行代码就能为页面绑定一个回调函数每当有网络响应返回时这个函数就会被触发你可以直接访问到完整的Response对象。2.2 为什么选择Playwright在Puppeteer主要驱动Chrome和Playwright之间我选择了后者原因如下多浏览器支持Playwright原生支持Chromium、Firefox和WebKitSafari内核。这意味着你可以用同一套脚本测试网站在不同浏览器下的行为对于需要绕过某些基于浏览器指纹的反爬策略很有帮助。更现代的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 核心APIpage.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(headlessFalse) # 非无头模式方便观察 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和响应内容类型URL关键词过滤如果目标网站的API有规律可循比如都包含/api/、/graphql、/data等路径这是最直接高效的过滤方式。响应头Content-Type过滤检查response.headers.get(‘content-type’)是否包含application/json。这是判断响应体是否为JSON的可靠方法。请求方法过滤通常获取数据的API使用GET或POST方法可以通过response.request.method来判断。综合判断最稳妥的方式是结合以上几点。优化后的回调函数示例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, indent2)[: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. 项目实战构建一个完整的网络层采集系统理论讲完了我们来搭建一个完整的、可复用的采集系统。这个系统要完成以下任务监听并过滤目标数据接口。解析JSON数据并提取出结构化的字段。将数据同时保存到CSV文件便于快速查看和Excel分析和SQLite数据库便于持久化存储和复杂查询。具备良好的扩展性方便适配不同的网站。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, fieldnamescsv_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_asciiFalse)) ) 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( headlessTrue, # 生产环境建议无头模式 args[‘--disable-blink-featuresAutomationControlled’] # 隐藏自动化特征 ) # 创建上下文可以设置更仿真的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”已完成第 {i1} 次滚动。”) 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_actionsscroll_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本身提供了一些绕过基础检测的功能但面对复杂的反爬系统还需要组合拳。伪装浏览器指纹context await browser.new_context( user_agent‘你的User-Agent字符串’, viewport{‘width’: 1920, ‘height’: 1080}, locale‘zh-CN’, timezone_id‘Asia/Shanghai’, # 可以覆盖更多设备属性 device_scale_factor1, has_touchFalse, is_mobileFalse, )使用代理IP这是应对IP封锁最有效的手段之一。browser await p.chromium.launch( headlessTrue, proxy{ “server”: “http://your-proxy-server:port”, “username”: “username”, # 如果需要认证 “password”: “password” } )重要提示务必使用合法合规的代理服务并遵守目标网站的robots.txt协议和速率限制。过度频繁的请求会给对方服务器带来压力也可能导致你的IP或代理IP被永久封禁。随机化操作间隔在page.wait_for_timeout()中使用随机延迟模拟人类操作的不确定性。import random await page.wait_for_timeout(random.uniform(1000, 3000)) # 随机等待1-3秒处理验证码Playwright可以截图对于简单的图形验证码可以截图后调用第三方OCR服务识别。对于复杂的滑块验证可以通过page.mouse.move()和page.mouse.down()等API模拟拖动但这需要精确计算滑块轨迹难度较高。5.3 错误处理与重试机制网络请求充满不确定性健壮的错误处理必不可少。async def robust_goto(page, url, max_retries3): “”“一个带重试的导航函数”“” for attempt in range(max_retries): try: response await page.goto(url, wait_until“networkidle”, timeout30000) if response and response.ok: return response else: print(f”导航失败状态码: {response.status if response else ‘无响应’}第{attempt1}次重试…”) except Exception as e: print(f”导航异常: {e}第{attempt1}次重试…”) 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’, inplaceTrue) # 绘制价格走势图 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(rotation45) 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/REPLACESQLite支持INSERT OR IGNORE INTO ...或INSERT OR REPLACE INTO ...语法但需要表有唯一约束UNIQUE constraint。我们可以修改表结构为(product_id, crawled_at_date)添加唯一约束然后使用INSERT OR IGNORE。7. 常见问题排查与实战心得在实际开发中你肯定会遇到各种各样的问题。这里记录了一些典型问题的排查思路和我踩过的坑。7.1 问题排查清单问题现象可能原因排查步骤与解决方案监听不到任何响应1. 监听器注册时机不对在页面导航后注册。2. 页面是无头模式且请求太快监听器还没绑定。3. 请求是其他上下文如iframe发出的。1.确保在page.goto()和任何可能触发请求的操作之前注册监听器。最好在page await browser.new_page()之后立即注册。2. 在开发阶段使用headlessFalse观察浏览器行为。3. 检查请求是否来自page.frames可能需要为每个frame单独监听。捕获的响应内容是乱码或HTML过滤条件不准确捕获到了非目标响应如页面本身。1. 加强过滤条件结合URL、Content-Type、甚至response.request.method如只监听POST。2. 在回调函数开头打印response.url和content_type确认过滤逻辑。response.json()抛出异常响应体不是有效的JSON格式可能是空的、是HTML或JS。1. 先用response.status判断请求是否成功200。2. 用try…except包裹await response.json()。3. 可以先调用text await response.text()打印前几百字符看看内容到底是什么。爬虫运行一段时间后被封IP请求频率过高触发了网站的反爬机制。1.最重要的遵守robots.txt并显著降低请求频率。在操作间增加随机延迟。2. 使用代理IP池进行轮换。3. 模拟更真实的人类行为模式如随机滚动、鼠标移动。数据解析出错字段缺失网站接口数据结构发生变化。1. 在解析代码中大量使用.get(‘key’, default)方法提供默认值。2.将原始JSON响应完整地保存到crawled_data表。这样即使解析逻辑出错原始数据还在可以事后修复解析脚本重新处理。这是非常重要的数据备份策略。内存使用量不断增长1. 缓冲区data_buffer没有及时清空。2. 保留了过多的Response对象引用。1. 确保_flush_buffer_to_storage被定期调用如缓冲区满或爬虫结束时。2. 在回调函数中只提取需要的数据不要长期持有整个response对象。7.2 实战心得与技巧从“宽”到“严”刚开始写过滤规则时不妨宽松一些先把所有application/json的响应都打印出来研究清楚目标网站的所有数据接口再逐步精确过滤。避免一开始就写死规则导致漏掉重要数据。原始数据是黄金无论你的解析逻辑多么完善一定要保留一份原始的、未加工的响应数据就像我们设计的crawled_data表。网站随时可能改版有了原始数据你就有能力在任何时候修复和升级你的解析器。异步编程的坑Playwright是异步库。确保你的回调函数on_response是async的并且在里面调用await response.json()。如果在非async函数中调用或者忘记了await会导致程序挂起或报错。等待的艺术page.wait_for_timeout()是简单的等待但不够智能。更好的做法是使用page.wait_for_selector()、page.wait_for_response()或page.wait_for_function()来等待特定的内容出现或请求完成。这能让脚本更稳定、更快。日志是你的眼睛在关键步骤开始监听、捕获到API、解析成功、存储成功、发生错误都加上详细的日志输出。这能让你在脚本无声无息地失败时快速定位问题所在。尊重与节制最后也是最重要的网络爬虫行走在法律的灰色地带。务必尊重网站的robots.txt协议控制请求速率避免对目标网站造成显著负担。将爬虫用于正当的学习、研究和合规的数据聚合才是长久之道。