1. 项目概述为什么选择Playwright构建新闻采集系统最近在做一个新闻数据聚合的项目核心需求是从几十个主流新闻网站定时抓取最新的文章标题、正文、发布时间和来源。一开始用的是传统的requestsBeautifulSoup组合但很快就遇到了瓶颈超过一半的目标站点都重度依赖JavaScript渲染静态HTML里要么没内容要么只有个骨架。试过Selenium稳定性是个大问题页面加载时间一长就容易超时卡死多任务并发时资源占用也吓人。后来把目光投向了微软开源的Playwright经过一段时间的实战我可以说对于现代动态网页的数据采集尤其是像新闻网站这种交互复杂、反爬机制多样的场景Playwright配合asyncio构建的异步采集系统是目前Python生态里最趁手的工具之一。简单来说这个“基于Playwright的新闻数据异步采集系统”要解决的核心问题就是高效、稳定、规模化地从动态渲染的新闻网站中提取结构化数据。它适合有一定Python基础需要从现代Web应用中采集数据的数据分析师、爬虫工程师或者任何有数据获取需求的开发者。系统不仅能处理简单的静态页面更能轻松应对无限滚动、点击加载更多、弹窗登录等交互其内置的智能等待和丰富的选择器让定位元素变得异常简单。接下来我会从设计思路、核心实现到避坑经验完整拆解如何搭建这样一个系统。2. 系统核心设计与架构选型2.1 同步 vs. 异步为何异步是采集系统的必然选择在构建爬虫系统时第一个要抉择的就是使用同步还是异步模型。对于新闻采集这种典型的I/O密集型任务大部分时间在等待网络响应和页面渲染异步编程能带来质的飞跃。同步模式下你的代码顺序执行发起一个请求 - 等待页面完全加载 - 解析数据 - 存储然后才处理下一个请求。如果单个页面加载需要3秒采集100个页面就需要至少300秒这期间CPU大部分时间在空闲等待。而异步模型利用asyncio库可以在一个任务等待时立刻切换到另一个任务。在Playwright的异步API支持下你可以在一个事件循环中并发管理多个浏览器页面Page同时加载数十个新闻页面。理论上这能将总耗时从线性叠加降低到接近最慢的那个页面的加载时间。对于需要实时或准实时获取新闻数据的场景异步是保证效率的基石。注意异步编程需要改变思维避免在异步函数中调用阻塞式操作如time.sleep而应使用await asyncio.sleep。Playwright的异步APIplaywright.async_api与asyncio天然契合是构建本系统的技术基础。2.2 Playwright核心对象模型Browser, Context, Page理解Playwright的三层对象模型是写出高效、稳定爬虫的关键。这个模型设计得非常精妙兼顾了隔离性和资源复用。Browser代表一个浏览器进程实例。你可以启动Chromium、Firefox或WebKit。启动浏览器是重量级操作开销较大。BrowserContext这是Playwright中一个非常强大的概念可以理解为“独立的浏览器会话”。每个Context拥有独立的缓存、Cookie、本地存储和证书。在爬虫中你可以利用不同的Context来模拟不同的用户实现Cookie隔离防止因某个站点的登录态异常或封禁影响到其他任务。Page对应一个浏览器标签页。绝大部分操作如导航、点击、填写表单、提取数据都在Page对象上进行。一个高效的采集架构通常是启动一个Browser实例 - 创建多个Context - 在每个Context中创建多个Page进行并发采集。这样既能利用浏览器实例复用来减少开销又能通过Context实现必要的隔离。2.3 系统架构蓝图基于以上分析我设计的系统架构如下主调度程序 (Scheduler) | v 异步事件循环 (asyncio Event Loop) | |--- 任务队列 (Task Queue: 待采集的新闻列表页URL) | v 采集工作池 (Worker Pool) | |--- Worker 1: Browser - Context1 - [Page1, Page2...] |--- Worker 2: Browser - Context2 - [Page3, Page4...] | |--- 页面导航 (Goto) | |--- 智能等待 (Wait for Selector/Network) | |--- 数据解析 (Locator Extract) | |--- 反爬处理 (Stealth, Proxy) | v 数据管道 (Data Pipeline) | |--- 数据清洗 (Cleaning) |--- 去重处理 (Deduplication) |--- 格式化存储 (CSV/JSON/DB)这个架构中主调度程序负责从种子URL生成具体的采集任务如各个新闻频道的列表页。工作池中的每个“Worker”是一个异步函数它管理着一个Browser Context及其下的多个Page并发地执行采集任务。提取到的原始数据被放入数据管道进行后续处理。这种设计清晰地将调度、采集和处理解耦便于维护和扩展。3. 环境搭建与核心配置详解3.1 Python环境与Playwright安装首先确保你的Python版本在3.8以上推荐3.10或更高以获得更好的异步特性支持。使用虚拟环境是良好的实践可以避免包依赖冲突。# 创建并激活虚拟环境以Linux/macOS为例 python3 -m venv playwright-env source playwright-env/bin/activate # 安装Playwright Python包 pip install playwright # 安装Playwright所需的浏览器内核Chromium, Firefox, WebKit playwright install chromium # 对于新闻采集通常安装Chromium就够了它兼容性最好。 # playwright install --with-deps chromium # 如果系统缺少依赖可以用此命令playwright install命令会下载对应平台的浏览器二进制文件存放在用户目录下。这一步可能需要一些时间因为下载的浏览器体积较大。3.2 初始化一个异步Playwright实例所有异步操作都始于async_playwright这个异步上下文管理器。import asyncio from playwright.async_api import async_playwright async def main(): # 使用 async with 管理 playwright 实例的生命周期 async with async_playwright() as p: # 启动浏览器headlessTrue表示无头模式不显示UI适合服务器环境 browser await p.chromium.launch(headlessTrue) # 创建一个独立的上下文 context await browser.new_context() # 在上下文中创建一个新页面 page await context.new_page() # 导航到目标网址 await page.goto(https://news.example.com) # 进行你的采集操作... # ... # 关闭上下文和浏览器 await context.close() await browser.close() # 运行主函数 asyncio.run(main())3.3 关键启动参数配置浏览器启动时的参数配置直接影响爬虫的稳定性、性能和隐蔽性。async def launch_browser(): async with async_playwright() as p: browser await p.chromium.launch( headlessTrue, # 无头模式必备 slow_mo50, # 操作间延迟50毫秒模拟真人操作有助于绕过简单行为检测 args[ --disable-blink-featuresAutomationControlled, # 关键隐藏自动化特征 --no-sandbox, # 在Docker或某些Linux环境下可能需要 --disable-dev-shm-usage, # 解决共享内存问题避免崩溃 --start-maximized # 启动时最大化有时能避免响应式布局问题 ] ) # 创建上下文时也可以注入初始脚本进一步隐藏自动化痕迹 context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..., # 设置UA # 可以加载预先准备好的Cookie文件 # storage_statepath/to/cookies.json ) return browser, context实操心得--disable-blink-featuresAutomationControlled这个参数至关重要。很多网站会检测navigator.webdriver属性这个参数可以将其设置为undefined是成本最低的反反爬措施之一。另外slow_mo参数在调试时也非常有用可以让你看清爬虫的每一步操作。4. 核心采集流程与Playwright API实战4.1 导航与等待确保页面加载完成导航到页面并等待其“就绪”是爬虫稳定性的第一步。Playwright提供了多种等待策略。await page.goto(https://news.site.com/latest, wait_untilnetworkidle)wait_until参数有几种选项load等待load事件触发。domcontentloaded等待DOMContentLoaded事件触发。networkidle最常用等待网络活动基本停止大约500ms内没有超过2个网络请求。这对于等待AJAX加载内容的页面非常有效。commit等待接收到响应。对于新闻列表页通常有动态加载networkidle是个不错的默认选择。但有时还不够你可能需要等待某个特定元素出现。# 组合等待策略先导航然后等待某个关键元素出现 await page.goto(url, wait_untildomcontentloaded) try: # 等待新闻列表的容器元素出现最多等待10秒 await page.wait_for_selector(.news-list, timeout10000) except Exception as e: print(f等待列表超时: {url}) return None4.2 元素定位与数据提取强大的Locator APIPlaywright的LocatorAPI是其精华所在它代表一个随时可以执行操作的元素选择器且自带自动等待和重试机制。# 定位单个元素 title_element page.locator(h1.article-title) if await title_element.count() 0: title await title_element.inner_text() # 定位多个元素获取列表页所有文章链接 article_links page.locator(.news-item a.title) count await article_links.count() for i in range(count): link_element article_links.nth(i) href await link_element.get_attribute(href) text await link_element.inner_text() print(f{text}: {href}) # 更复杂的定位结合文本和CSS # 找到包含“独家”文本的新闻条目 exclusive_news page.locator(.news-item:has-text(独家))提取数据常用方法inner_text(): 获取元素及其子元素的可见文本。text_content(): 获取所有文本节点内容包括隐藏的。get_attribute(attr): 获取属性值如href、src。input_value(): 获取输入框的值。inner_html()/outer_html(): 获取HTML内容。注意事项inner_text()和text_content()有区别。inner_text会考虑CSS样式忽略隐藏元素的文本更接近用户所见。而text_content会获取所有文本可能包含脚本和样式内容。对于新闻正文通常使用inner_text或针对特定内容区域的text_content。4.3 处理交互点击、滚动、表单新闻网站常有“加载更多”按钮或无限滚动。# 案例1点击“加载更多”按钮 load_more_button page.locator(button:has-text(加载更多)) while await load_more_button.is_visible(): await load_more_button.click() # 点击后等待新内容加载 await page.wait_for_timeout(2000) # 简单等待2秒 # 或者更好等待新出现的某个元素 # await page.wait_for_selector(.news-item:nth-last-child(5), timeout5000) # 案例2模拟滚动触发无限加载 for _ in range(5): # 滚动5次 await page.evaluate(window.scrollTo(0, document.body.scrollHeight)) await page.wait_for_timeout(3000) # 等待新内容加载 # 可以检查内容高度是否增加 new_height await page.evaluate(document.body.scrollHeight) # ... 判断逻辑 # 案例3处理搜索或登录表单如果需要 await page.fill(input[nameq], 人工智能) await page.press(input[nameq], Enter) # 按回车键 # 或者点击搜索按钮 # await page.click(button[typesubmit])page.evaluate()用于在页面上下文中执行JavaScript这是处理复杂页面逻辑的终极武器。4.4 高级功能请求拦截与资源控制为了提高采集效率我们可以拦截不必要的请求如图片、样式表、广告、分析脚本。async def handle_route(route): 拦截请求的处理函数 resource_type route.request.resource_type # 只允许文档、XHRAJAX、fetch和脚本请求通过阻塞图片、样式、字体、媒体等 if resource_type in [image, stylesheet, font, media]: await route.abort() else: await route.continue_() # 在创建页面后设置路由拦截 await page.route(**/*, handle_route)这个技巧能显著提升页面加载速度减少带宽消耗。但需谨慎有时页面CSS或关键字体被拦截会导致布局错乱影响元素定位。建议根据目标网站情况调整拦截策略。5. 构建异步并发采集系统5.1 使用asyncio.gather实现并发控制简单的并发可以通过asyncio.gather来实现同时运行多个采集任务。import asyncio async def fetch_article(page, url): 采集单篇文章的协程函数 await page.goto(url, wait_untilnetworkidle) title await page.locator(h1).inner_text() content await page.locator(.article-content).inner_text() return {url: url, title: title, content: content} async def main(): async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) context await browser.new_context() article_urls [url1, url2, url3, url4, url5] # 文章链接列表 # 创建多个页面实例用于并发 pages [await context.new_page() for _ in range(3)] # 假设并发度为3 # 为每个页面分配任务简单的轮询分配 tasks [] for i, url in enumerate(article_urls): page pages[i % len(pages)] # 循环使用页面 task asyncio.create_task(fetch_article(page, url)) tasks.append(task) # 等待所有任务完成 results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果 for result in results: if isinstance(result, Exception): print(f任务失败: {result}) else: print(f采集成功: {result[title]}) await context.close() await browser.close() asyncio.run(main())5.2 更健壮的方案结合信号量Semaphore控制并发度直接使用gather会一次性启动所有任务如果任务数量成百上千可能导致内存激增或目标服务器压力过大。使用asyncio.Semaphore可以精确控制同时运行的任务数量。import asyncio from playwright.async_api import async_playwright class NewsCrawler: def __init__(self, max_concurrent5): self.semaphore asyncio.Semaphore(max_concurrent) # 控制最大并发数 async def crawl_article(self, context, url): 受信号量保护的采集函数 async with self.semaphore: # 确保同时只有max_concurrent个任务在执行 page await context.new_page() try: await page.goto(url, wait_untilnetworkidle, timeout30000) # ... 数据提取逻辑 ... data {url: url, title: 示例标题} return data except Exception as e: print(f采集失败 {url}: {e}) return None finally: await page.close() # 重要及时关闭页面释放资源 async def run(self, url_list): async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) # 可以为不同类别的网站创建不同的context实现隔离 context await browser.new_context() # 创建所有采集任务 tasks [self.crawl_article(context, url) for url in url_list] # 并发执行但受信号量限制 results await asyncio.gather(*tasks) # 过滤出成功的结果 successful_results [r for r in results if r] print(f共采集 {len(successful_results)}/{len(url_list)} 篇文章) await context.close() await browser.close() # 使用 crawler NewsCrawler(max_concurrent5) # 同时最多5个页面 asyncio.run(crawler.run([url1, url2, ...]))5.3 使用异步队列Queue进行任务调度对于更复杂的生产级系统结合asyncio.Queue实现生产者-消费者模型是更优的选择。主程序生产URL任务多个消费者协程从队列中取出任务并执行。import asyncio from asyncio import Queue from playwright.async_api import async_playwright async def worker(worker_id, queue, context, results): 消费者协程从队列取任务并执行 while True: url await queue.get() if url is None: # 终止信号 queue.task_done() break try: page await context.new_page() await page.goto(url) # ... 采集逻辑 ... results.append({url: url, worker: worker_id}) await page.close() except Exception as e: print(fWorker-{worker_id} 处理 {url} 失败: {e}) finally: queue.task_done() # 告知队列该任务已完成 async def main(): url_list [...] # 大量的URL列表 queue Queue() results [] # 将所有URL放入队列 for url in url_list: await queue.put(url) async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) context await browser.new_context() # 启动5个消费者协程 workers [] for i in range(5): task asyncio.create_task(worker(i, queue, context, results)) workers.append(task) # 等待所有任务被处理完 await queue.join() # 发送终止信号给所有消费者 for _ in range(5): await queue.put(None) # 等待所有消费者协程结束 await asyncio.gather(*workers) print(f采集完成共获得 {len(results)} 条结果) await context.close() await browser.close() asyncio.run(main())这种模式扩展性极强可以动态调整消费者数量优雅地处理任务失败和重试逻辑。6. 反爬策略与稳定性保障6.1 基础隐身与指纹修改尽管Playwright本身已经比Selenium隐蔽但一些高级反爬系统仍能检测。我们需要主动修改一些浏览器指纹。async def create_stealth_context(browser): context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..., localezh-CN, timezone_idAsia/Shanghai, # 覆盖WebDriver属性 bypass_cspTrue, # 绕过内容安全策略方便注入脚本 ) # 注入JavaScript代码覆盖navigator.webdriver等属性 await context.add_init_script( Object.defineProperty(navigator, webdriver, { get: () undefined }); // 覆盖plugins属性 Object.defineProperty(navigator, plugins, { get: () [1, 2, 3, 4, 5] }); // 覆盖languages属性 Object.defineProperty(navigator, languages, { get: () [zh-CN, zh, en] }); // 模拟Chrome的window.chrome对象 window.chrome { runtime: {}, loadTimes: function() {}, csi: function() {}, app: {} }; ) return context6.2 代理IP池集成使用代理是分散请求压力、绕过IP封锁的基本手段。最好搭配一个可靠的代理IP池服务。import random class ProxyManager: def __init__(self, proxy_list): self.proxy_list proxy_list def get_random_proxy(self): return random.choice(self.proxy_list) async def create_browser_with_proxy(p, proxy_manager): proxy proxy_manager.get_random_proxy() browser await p.chromium.launch( headlessTrue, proxy{ server: proxy, # 例如 http://123.45.67.89:8080 # 如果需要用户名密码认证 # username: user, # password: pass }, args[--disable-blink-featuresAutomationControlled] ) return browser重要提示代理IP的质量至关重要。免费代理往往不稳定、速度慢且容易被目标网站封禁。生产环境建议使用付费的优质代理服务并实现自动检测代理可用性的机制。6.3 请求频率控制与随机延迟即使使用代理过于频繁的请求也可能触发风控。需要在代码中植入随机延迟模拟人类浏览行为。import asyncio import random async def human_like_delay(min_seconds1, max_seconds3): 随机延迟模拟人类操作间隔 delay random.uniform(min_seconds, max_seconds) await asyncio.sleep(delay) async def crawl_with_delay(page, url): await page.goto(url) # ... 提取数据 ... await human_like_delay(2, 5) # 每次操作后延迟2-5秒更精细的控制可以基于域名设计请求间隔确保对同一域名的请求不会过于密集。6.4 错误处理与重试机制网络爬虫必须健壮。要预见各种错误超时、元素未找到、网络断开等并实现重试。import asyncio from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from playwright.async_api import TimeoutError as PlaywrightTimeoutError class CrawlerWithRetry: retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min2, max10), # 指数退避等待 retryretry_if_exception_type((PlaywrightTimeoutError, asyncio.TimeoutError)) ) async def fetch_page_with_retry(self, page, url): 带重试的页面获取函数 try: await page.goto(url, wait_untilnetworkidle, timeout30000) return await page.content() except Exception as e: print(f请求失败准备重试: {e}) # 重试前可以做一些清理比如关闭当前页面开一个新的 await page.close() raise e # 重新抛出异常让tenacity捕获并重试 # 使用tenacity库可以优雅地实现复杂的重试逻辑包括对不同异常的不同处理策略。7. 数据存储、去重与系统优化7.1 数据存储方案采集到的数据需要持久化。对于新闻数据常见的存储方式有文件存储JSON Lines, CSV适合中小规模或临时存储。import json import aiofiles # 异步文件操作库 async def save_to_jsonl(data, filename): async with aiofiles.open(filename, a, encodingutf-8) as f: await f.write(json.dumps(data, ensure_asciiFalse) \n)数据库存储MySQL, PostgreSQL, MongoDB适合大规模、结构化、需要复杂查询的数据。关系型数据库如PostgreSQL如果新闻数据字段固定标题、正文、时间、来源等且需要做关联分析这是好选择。文档数据库如MongoDB如果不同网站的文章结构差异大或者正文内容很长MongoDB的灵活性更有优势。搜索引擎Elasticsearch如果后续需要做全文检索、相关性排序可以直接存入ES。在异步环境中记得使用对应的异步数据库驱动如asyncpgPostgreSQL、aiomysqlMySQL、motorMongoDB避免阻塞事件循环。7.2 基于布隆过滤器的URL去重在采集新闻列表页时防止重复采集同一篇文章是必须的。布隆过滤器Bloom Filter是一种内存效率极高的概率数据结构非常适合判断一个URL是否已存在。import mmh3 # MurmurHash库 from bitarray import bitarray import math class SimpleBloomFilter: def __init__(self, size, hash_num): self.size size # 位数组大小 self.hash_num hash_num # 哈希函数个数 self.bit_array bitarray(size) self.bit_array.setall(0) def add(self, url): for seed in range(self.hash_num): index mmh3.hash(url, seed) % self.size self.bit_array[index] 1 def exists(self, url): for seed in range(self.hash_num): index mmh3.hash(url, seed) % self.size if self.bit_array[index] 0: return False return True # 可能存在有误判率 # 估算布隆过滤器参数 def estimate_bloom_params(n, p): n: 预期元素数量 p: 可接受的误判率 返回: (位数组大小 m, 哈希函数数量 k) m - (n * math.log(p)) / (math.log(2) ** 2) k (m / n) * math.log(2) return int(m), int(k) # 使用示例 expected_urls 1000000 # 预期100万个URL false_positive_rate 0.001 # 0.1%的误判率 m, k estimate_bloom_params(expected_urls, false_positive_rate) bloom SimpleBloomFilter(m, k) # 在采集到新URL时 new_url https://news.example.com/article/123 if not bloom.exists(new_url): bloom.add(new_url) # 加入采集队列 else: # 很可能已采集过跳过 pass对于绝对精确的去重可以将布隆过滤器与数据库存储结合先用布隆过滤器快速判断如果返回“可能存在”再去数据库做一次精确查询。7.3 性能监控与日志记录一个健壮的系统需要可观测性。集成日志记录和简单的性能监控。import logging import time from contextlib import contextmanager # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) contextmanager def log_execution_time(task_name): 记录任务执行时间的上下文管理器 start_time time.time() try: yield finally: elapsed time.time() - start_time logger.info(f{task_name} 执行耗时: {elapsed:.2f}秒) # 使用 async def crawl_site(): with log_execution_time(采集XX新闻站): # ... 采集逻辑 ... logger.info(f成功采集到 {count} 条新闻)可以定期输出一些指标如成功率成功数/总数、平均耗时、各网站状态等便于发现问题。8. 常见问题排查与实战技巧8.1 元素定位失败问题这是Playwright爬虫最常见的问题。页面结构变化、动态加载、iframe嵌套都可能导致定位失败。排查步骤确认页面已加载完成在定位前增加page.wait_for_selector或page.wait_for_load_state(networkidle)。使用Playwright Inspector调试在启动浏览器时设置headlessFalse和devtoolsTrue手动操作并录制代码或者使用playwright codegen命令生成定位器。playwright codegen https://news.example.com尝试更宽松的选择器如果CSS选择器太精确可以尝试用文本定位、XPath或page.locator()的其他变体。# 文本定位 await page.locator(text阅读全文).click() # 包含某属性的元素 await page.locator([data-rolearticle]).first.inner_text() # 使用XPath谨慎易脆 await page.locator(xpath//div[classcontent]/p[1]).inner_text()检查是否在iframe内有些网站的评论或广告区域是iframe。# 切换到iframe frame page.frame(namecomment-frame) # 通过name # 或者 frame page.frame(urlre.compile(r.*comments.*)) if frame: content await frame.locator(.comment).inner_text()8.2 页面超时与卡死处理网络不稳定或页面脚本问题可能导致page.goto或page.wait_for_selector超时。解决方案设置合理的超时时间全局设置或单独设置。# 全局设置 page.set_default_timeout(60000) # 60秒 # 单独设置 await page.goto(url, timeout60000) await page.wait_for_selector(.content, timeout30000)实现超时后的恢复逻辑使用asyncio.wait_for包装可能超时的操作并捕获asyncio.TimeoutError。try: await asyncio.wait_for(page.goto(url), timeout45.0) except asyncio.TimeoutError: logger.warning(f页面 {url} 加载超时尝试刷新或跳过) await page.reload() # 尝试刷新 # 或者直接跳过记录失败定期检查与重启对于长时间运行的爬虫可以设置一个监控任务定期检查浏览器或页面的健康状态必要时重启浏览器实例。8.3 内存泄漏与资源管理异步爬虫长时间运行可能出现内存缓慢增长的问题。关键预防措施及时关闭Page和Context每个任务完成后务必调用await page.close()。使用try...finally块确保资源释放。限制并发Page数量如前面所述使用信号量Semaphore严格控制同时打开的页面数量。定期重启Browser实例对于需要7x24小时运行的爬虫可以设置一个运行时长或任务数阈值达到后优雅地关闭当前Browser重启一个新的。class ManagedBrowser: def __init__(self, max_tasks1000): self.task_count 0 self.max_tasks max_tasks self.browser None async def get_page(self): if self.browser is None or self.task_count self.max_tasks: await self.restart_browser() self.task_count 1 context await self.browser.new_context() return await context.new_page() async def restart_browser(self): if self.browser: await self.browser.close() self.browser await p.chromium.launch() self.task_count 08.4 应对网站反爬升级反爬策略是动态的。今天有效的技巧明天可能就失效了。应对策略多样化User-Agent和浏览器指纹维护一个UA池每次创建Context时随机选取。定期更新add_init_script中注入的隐藏脚本。模拟更真实的行为除了随机延迟还可以模拟鼠标移动、随机滚动等。import random async def random_mouse_move(page): width, height 1920, 1080 x random.randint(0, width) y random.randint(0, height) await page.mouse.move(x, y)准备降级方案当Playwright被严重封锁时可以考虑降级到requests-html或httpxparsel只抓取静态HTML。虽然可能丢失动态内容但能保证基础数据流不断。人工介入与规则更新建立报警机制当某个网站的采集成功率持续下降时触发报警通知人工检查页面结构或反爬策略是否变化并更新对应的解析规则。构建一个基于Playwright的异步新闻采集系统技术核心在于深刻理解异步编程模型、Playwright的对象生命周期以及稳健的错误处理机制。从简单的单页采集到复杂的分布式并发每一步都需要在效率、稳定性和隐蔽性之间取得平衡。这套系统搭建起来后不仅仅是新闻任何依赖JavaScript渲染的现代Web应用数据都可以成为你的采集目标。关键在于始终对目标网站保持尊重合理控制采集频率遵守robots.txt协议将技术用在正当的数据获取与分析上这才是可持续的爬虫之道。
