基于Playwright与AI的新闻聚合与智能摘要系统实战

1. 项目概述：为什么选择Playwright与AI做新闻聚合

最近在折腾一个能自动抓新闻、自动总结的“信息助理”，核心就两件事：一是让程序像人一样去网站上点开新闻、翻页、甚至登录，把内容扒下来；二是让AI快速消化这些内容，给我一个简洁的摘要。听起来挺复杂，但用对工具，这事儿对新手也友好。我最终选定的技术栈是Playwright搭配大模型API，原因很简单：Playwright是目前对现代网页支持最好、最稳定的浏览器自动化工具，而大模型（比如GPT、文心一言、通义千问的API）处理文本总结是降维打击。

这个项目的价值在于，它能帮你从信息洪流中解放出来。想象一下，你每天需要关注十几个不同网站的行业动态，手动一个个点开看，效率极低。这个脚本可以定时运行，自动收集所有你关心的新闻，并生成一份结构化的简报。无论是用于个人学习、市场调研，还是内容创作素材收集，都非常实用。整个流程完全自动化，从打开浏览器、模拟点击、抓取内容，到调用AI进行总结，一气呵成。下面，我就把这个“开箱即用”的保姆级方案拆解给你，代码可以直接复制修改。

2. 环境准备与核心工具选型

2.1 Python环境与Playwright安装

首先，确保你的电脑上安装了Python（建议3.8及以上版本）。打开命令行（Windows的CMD/PowerShell，Mac/Linux的Terminal），我们来安装必要的包。

# 安装Playwright的Python库 pip install playwright # 安装Playwright所需的浏览器（Chromium, Firefox, WebKit） playwright install

这里有个关键点：playwright install这个命令会下载浏览器内核。因为Playwright需要真实的浏览器环境来运行，而不是简单的HTTP请求。这确保了它能处理JavaScript渲染的现代网页（比如很多新闻网站用的Vue或React），也能完美模拟点击、滚动、填写表单等真人操作。如果下载速度慢，可以尝试设置环境变量PLAYWRIGHT_DOWNLOAD_HOST为国内镜像源，但请注意，这需要你自行搜索可靠的镜像地址并评估其安全性。

2.2 大模型API的选择与配置

我们需要一个能处理文本总结的大模型接口。这里有几个主流选择：

1. OpenAI GPT系列：效果稳定，但需要海外网络环境及付费。
2. 国内大厂API：如百度文心一言、阿里通义千问、讯飞星火等。这些对国内用户更友好，访问速度快，通常也提供免费额度供测试。
3. 本地部署大模型：如使用Ollama运行Llama 3等模型。这完全离线，数据隐私性最好，但对电脑硬件（尤其是GPU）有一定要求，且总结效果可能略逊于顶级商用API。

对于本教程，我们以获取便捷的国内大模型API为例。你需要去对应平台的官网注册账号，并创建一个API Key。通常，平台会提供详细的接入文档和计费说明（很多有免费额度）。将获得的API Key和API Base URL（接口地址）保存好，我们稍后会用到。

注意：保管好你的API Key，不要直接上传到公开的代码仓库（如GitHub），以免被他人盗用产生费用。推荐使用环境变量或配置文件来管理。

2.3 项目目录结构规划

在开始写代码前，规划好目录能让逻辑更清晰。建议创建一个项目文件夹，结构如下：

news_auto_summarizer/ ├── config.py # 配置文件，存放API密钥、目标网址等 ├── crawler.py # 爬虫核心模块，使用Playwright抓取 ├── summarizer.py # 总结模块，调用大模型API ├── main.py # 主程序，调度整个流程 ├── requirements.txt # 项目依赖列表 └── logs/ # （可选）日志目录

我们先在config.py里配置基础信息：

# config.py # 大模型API配置（以某国内平台为例，实际需替换为你的信息） LLM_API_KEY = "your_api_key_here" LLM_API_BASE = "https://api.example.com/v1" # API基础地址 LLM_MODEL = "ernie-bot-4" # 或 "qwen-max", "spark-v3.5" 等 # 目标新闻网站列表（示例） NEWS_SITES = [ { "name": "科技新闻网示例", "url": "https://example-news.com/tech", "article_selector": ".article-list a.title", # 文章链接选择器 "content_selector": ".article-content", # 正文内容选择器 "need_login": False, # 是否需要登录 "login_url": None, # 如果需要登录的登录页面地址 }, # 可以添加更多网站配置... ] # 爬虫配置 HEADLESS = True # 是否使用无头模式（不显示浏览器界面），True为后台运行 SLOW_MO = 100 # 操作延迟（毫秒），模拟真人操作，有助于避免被反爬 TIMEOUT = 30000 # 页面加载超时时间（毫秒）

3. 核心模块一：使用Playwright实现智能爬虫

Playwright的强大之处在于它能处理复杂的用户交互。我们的爬虫模块需要完成：启动浏览器、导航到页面、可能处理登录、列表页翻页抓取文章链接、进入详情页提取内容。

3.1 浏览器启动与页面导航基础

我们创建一个BrowserManager类来管理浏览器生命周期。

# crawler.py from playwright.sync_api import sync_playwright import logging from config import HEADLESS, SLOW_MO, TIMEOUT logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class BrowserManager: def __init__(self): self.playwright = None self.browser = None self.context = None def start(self): """启动浏览器和上下文""" self.playwright = sync_playwright().start() # 选择Chromium，兼容性最好。也可选firefox或webkit。 self.browser = self.playwright.chromium.launch(headless=HEADLESS, slow_mo=SLOW_MO) # 创建上下文，可以设置视窗大小、User-Agent等 self.context = self.browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...' ) logger.info("浏览器启动成功。") return self.context def stop(self): """关闭浏览器，释放资源""" if self.context: self.context.close() if self.browser: self.browser.close() if self.playwright: self.playwright.stop() logger.info("浏览器资源已释放。")

3.2 自动填写表单与登录处理

很多新闻网站需要登录才能查看全文或避免频繁访问限制。Playwright处理表单非常直观。

# crawler.py (续) class NewsCrawler: def __init__(self, browser_manager): self.browser_manager = browser_manager self.context = browser_manager.context def handle_login(self, page, login_url, login_info): """ 处理网站登录 :param page: playwright页面对象 :param login_url: 登录页面地址 :param login_info: 字典，包含如 {'username_selector': '#user', 'password_selector': '#pass', 'submit_selector': 'button[type=submit]', 'username': 'your_user', 'password': 'your_pass'} """ logger.info(f"正在导航至登录页: {login_url}") page.goto(login_url, timeout=TIMEOUT) page.wait_for_load_state('networkidle') # 等待网络空闲 # 填写用户名 page.fill(login_info['username_selector'], login_info['username']) # 填写密码 page.fill(login_info['password_selector'], login_info['password']) # 点击提交按钮 page.click(login_info['submit_selector']) # 等待登录成功后的跳转或页面变化，这里根据实际情况调整等待条件 page.wait_for_url('**/home**') # 示例：等待跳转到包含/home的URL logger.info("登录成功。") # 可选：保存登录状态cookies，避免同一会话重复登录 # self.context.storage_state(path="state.json") def crawl_article_list(self, page, list_url, article_selector): """ 抓取列表页的文章链接 :return: 文章链接列表 """ logger.info(f"正在抓取列表页: {list_url}") page.goto(list_url, timeout=TIMEOUT) # 等待文章列表元素加载 page.wait_for_selector(article_selector, timeout=TIMEOUT) # 获取所有文章链接元素 link_elements = page.query_selector_all(article_selector) article_links = [] for elem in link_elements: href = elem.get_attribute('href') # 处理相对链接 if href and not href.startswith('http'): href = urljoin(list_url, href) if href: article_links.append(href) logger.info(f"从列表页提取到 {len(article_links)} 篇文章链接。") return article_links

实操心得：page.wait_for_selector和page.wait_for_load_state是保证爬虫稳定性的关键。现代网页大量使用异步加载，直接抓取可能拿到空页面。networkidle状态表示页面在至少500ms内没有新的网络请求，通常意味着主要内容已加载完成。对于动态加载更多的网站（如滚动翻页），可能需要监听滚动事件或使用page.evaluate执行JavaScript来模拟滚动。

3.3 内容提取与翻页逻辑

抓取到链接后，我们需要进入详情页提取标题和正文。

# crawler.py (续) from urllib.parse import urljoin class NewsCrawler: # ... 接上文 ... def extract_article_content(self, page, article_url, content_selector): """ 进入文章详情页并提取内容 :return: 字典，包含标题和正文 """ logger.info(f"正在提取文章内容: {article_url}") try: page.goto(article_url, timeout=TIMEOUT) # 等待正文内容加载 page.wait_for_selector(content_selector, timeout=TIMEOUT) # 提取标题 - 通常可以从页面title或特定标题选择器获取 title = page.title() # 简单方法，取页面标题 # 更精确的方法： title = page.text_content('h1.article-title') # 提取正文 content_element = page.query_selector(content_selector) if content_element: # 获取元素内的所有文本，并清理多余空白 content = content_element.inner_text().strip() # 简单的文本清理：合并多个换行和空格 import re content = re.sub(r'\n\s*\n', '\n\n', content) content = re.sub(r'[ \t]+', ' ', content) else: content = "" logger.warning(f"未找到内容选择器 {content_selector} 对应的元素。") return { 'url': article_url, 'title': title, 'content': content[:2000] # 限制内容长度，避免API token超限 } except Exception as e: logger.error(f"提取文章 {article_url} 内容时出错: {e}") return None def crawl_site(self, site_config): """ 对一个新闻站点进行完整抓取 :param site_config: 来自config.py的站点配置字典 :return: 抓取到的文章内容列表 """ all_articles = [] context = self.context page = context.new_page() # 为每个站点创建一个新页面 try: # 1. 如果需要登录，先处理登录 if site_config.get('need_login') and site_config.get('login_url'): # 这里需要你在config里补充具体的登录信息选择器 login_info = { 'username_selector': '#username_input', 'password_selector': '#password_input', 'submit_selector': 'button.login-btn', 'username': 'your_username', # 应从安全的地方读取，如环境变量 'password': 'your_password' } self.handle_login(page, site_config['login_url'], login_info) # 2. 抓取文章列表 article_links = self.crawl_article_list(page, site_config['url'], site_config['article_selector']) # 3. 遍历链接，提取每篇文章内容 for link in article_links[:5]: # 示例：只抓取前5篇，避免请求过多 article_data = self.extract_article_content(page, link, site_config['content_selector']) if article_data and article_data['content']: # 确保有内容 all_articles.append(article_data) # 礼貌性延迟，避免对服务器造成压力 page.wait_for_timeout(2000) # 4. （可选）处理翻页 # next_page_selector = site_config.get('next_page_selector') # if next_page_selector: # while page.is_visible(next_page_selector): # page.click(next_page_selector) # page.wait_for_timeout(3000) # 等待新内容加载 # new_links = self.crawl_article_list(page, page.url, site_config['article_selector']) # article_links.extend(new_links) # # 继续提取新链接的内容... finally: page.close() # 关闭页面 logger.info(f"站点 [{site_config['name']}] 抓取完成，共获得 {len(all_articles)} 篇文章。") return all_articles

4. 核心模块二：连接大模型进行智能总结

抓取到原始文本后，我们需要调用大模型API来生成摘要。这里的关键是构造一个清晰的提示词（Prompt），让AI理解我们的需求。

4.1 构建高效的总结提示词（Prompt）

提示词的质量直接决定总结效果。一个好的新闻总结Prompt应该包含：

1. 角色定义：让AI明确自己的任务。
2. 输入格式说明：告诉AI你给它的是什么。
3. 输出格式要求：明确需要AI返回什么，最好是结构化数据（如JSON）。
4. 具体任务指令：总结的重点是什么（如核心事件、观点、数据、后续影响）。

# summarizer.py import requests import json from config import LLM_API_KEY, LLM_API_BASE, LLM_MODEL class LLMSummarizer: def __init__(self): self.api_key = LLM_API_KEY self.api_base = LLM_API_BASE self.model = LLM_MODEL self.headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } def build_summary_prompt(self, title, content): """ 构建总结新闻的Prompt """ prompt = f"""你是一个专业的新闻编辑助理。请根据以下新闻标题和正文，生成一份简洁的摘要。 新闻标题：{title} 新闻正文： {content} 请按照以下JSON格式回复，不要包含任何其他解释性文字： {{ "summary": "新闻的核心内容摘要，不超过200字。", "keywords": ["关键词1", "关键词2", "关键词3", "关键词4", "关键词5"], "sentiment": "对新闻情感倾向的判断，可选：积极/消极/中性", "category": "新闻所属类别，如：科技/财经/体育/娱乐/政治等" }} """ return prompt def call_llm_api(self, prompt): """ 调用大模型API """ # 注意：不同平台的API请求体格式可能不同，请根据官方文档调整 data = { "model": self.model, "messages": [ {"role": "user", "content": prompt} ], "temperature": 0.2, # 低温度使输出更确定、更聚焦 "max_tokens": 500 # 限制回复长度 } try: response = requests.post( f"{self.api_base}/chat/completions", # 常见端点，需根据平台调整 headers=self.headers, data=json.dumps(data), timeout=60 ) response.raise_for_status() # 如果状态码不是200，抛出异常 result = response.json() # 提取AI返回的文本内容 ai_response = result['choices'][0]['message']['content'].strip() return ai_response except requests.exceptions.RequestException as e: print(f"API请求失败: {e}") if hasattr(e, 'response') and e.response: print(f"错误响应: {e.response.text}") return None except KeyError as e: print(f"解析API响应时出错，响应结构可能已改变: {e}") print(f"原始响应: {result}") return None

4.2 解析AI回复与数据清洗

AI返回的是文本，我们需要将其解析成结构化的Python字典，以便后续存储或展示。

# summarizer.py (续) import re class LLMSummarizer: # ... 接上文 ... def parse_ai_response(self, ai_response_text): """ 尝试从AI的回复中解析出JSON结构。 由于AI可能不会严格返回纯JSON，我们需要一些容错处理。 """ if not ai_response_text: return None # 方法1：尝试直接查找JSON块 json_match = re.search(r'\{.*\}', ai_response_text, re.DOTALL) if json_match: json_str = json_match.group() try: return json.loads(json_str) except json.JSONDecodeError: # 如果直接解析失败，可能是JSON内部有轻微格式问题，尝试简单清理 pass # 方法2：如果找不到标准JSON，尝试按行提取关键信息（降级方案） result = { "summary": "", "keywords": [], "sentiment": "中性", "category": "其他" } lines = ai_response_text.split('\n') for line in lines: line_lower = line.lower() if 'summary:' in line_lower or '摘要:' in line_lower: result['summary'] = line.split(':', 1)[-1].strip() elif 'keyword' in line_lower: # 尝试提取类似 "keywords: a, b, c" 的内容 parts = line.split(':', 1) if len(parts) > 1: kw_str = parts[1].strip() result['keywords'] = [k.strip() for k in kw_str.split(',')[:5]] # 可以添加更多规则提取 sentiment 和 category... # 如果摘要还是空的，使用前200个字符作为fallback if not result['summary']: result['summary'] = ai_response_text[:200].strip() return result def summarize_article(self, title, content): """ 总结单篇文章的完整流程 """ if not content: print("文章内容为空，跳过总结。") return None prompt = self.build_summary_prompt(title, content) print(f"正在为《{title[:30]}...》生成摘要...") ai_response = self.call_llm_api(prompt) if ai_response: structured_data = self.parse_ai_response(ai_response) return structured_data else: print("调用大模型API失败。") return None

注意事项：大模型API是收费服务，且通常有调用频率限制（RPM/TPM）。在循环中调用时，务必添加延迟（如time.sleep(1)），避免触发限流。同时，控制每次发送的文本长度，因为费用通常与输入的Token数量（可粗略理解为字数）正相关。上述代码中我们对文章内容做了截断content[:2000]就是出于这个考虑。

5. 主程序调度与数据持久化

现在我们把爬虫和总结器组装起来，并考虑如何保存结果。

5.1 主流程编排与错误处理

主程序main.py负责串联所有步骤，并加入必要的异常处理和日志。

# main.py import time import json from datetime import datetime from crawler import BrowserManager, NewsCrawler from summarizer import LLMSummarizer from config import NEWS_SITES def main(): """主执行函数""" # 初始化管理器 browser_manager = BrowserManager() summarizer = LLMSummarizer() all_results = [] try: context = browser_manager.start() crawler = NewsCrawler(browser_manager) for site in NEWS_SITES: print(f"\n{'='*50}") print(f"开始处理站点: {site['name']}") print(f"{'='*50}") try: # 1. 抓取文章 articles = crawler.crawl_site(site) if not articles: print(f"站点 {site['name']} 未抓取到文章。") continue # 2. 逐篇文章总结 for article in articles: print(f"\n处理文章: {article['title'][:50]}...") summary_data = summarizer.summarize_article(article['title'], article['content']) if summary_data: # 合并原始数据和总结数据 final_data = { **article, # 包含 url, title, content **summary_data, # 包含 summary, keywords, sentiment, category 'source': site['name'], 'crawl_time': datetime.now().isoformat() } all_results.append(final_data) print(f" 摘要生成成功。") else: print(f" 摘要生成失败。") # 请求间隔，避免API限流 time.sleep(2) except Exception as e: print(f"处理站点 {site['name']} 时发生错误: {e}") # 记录详细日志 import traceback traceback.print_exc() continue # 一个站点出错，继续下一个 finally: # 确保浏览器被关闭 browser_manager.stop() # 3. 保存结果 if all_results: output_file = f"news_summary_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(output_file, 'w', encoding='utf-8') as f: json.dump(all_results, f, ensure_ascii=False, indent=2) print(f"\n所有任务完成！结果已保存至: {output_file}") print(f"共处理 {len(all_results)} 篇文章。") else: print("\n未抓取到任何可总结的文章。") if __name__ == "__main__": main()

5.2 结果存储与展示优化

保存为JSON文件是最简单的方式，方便后续处理。你也可以将数据存入数据库（如SQLite、MySQL）或导出为Markdown、HTML格式的日报。

# 一个简单的Markdown报告生成函数示例 def generate_markdown_report(results, filename="news_report.md"): """将结果生成为Markdown格式的报告""" with open(filename, 'w', encoding='utf-8') as f: f.write(f"# 每日新闻摘要 {datetime.now().date()}\n\n") f.write(f"生成时间：{datetime.now()}\n") f.write(f"共收录 {len(results)} 条新闻。\n\n") # 按类别分组 from collections import defaultdict by_category = defaultdict(list) for item in results: by_category[item.get('category', '未分类')].append(item) for category, articles in by_category.items(): f.write(f"## {category}\n\n") for article in articles: f.write(f"### {article['title']}\n") f.write(f"**来源**：{article['source']} | **情感**：{article.get('sentiment', 'N/A')}\n") f.write(f"**关键词**：{', '.join(article.get('keywords', []))}\n") f.write(f"**摘要**：{article['summary']}\n") f.write(f"[原文链接]({article['url']})\n\n") f.write("---\n\n") print(f"Markdown报告已生成: {filename}")

6. 常见问题与排查技巧实录

在实际运行中，你肯定会遇到各种问题。这里记录了几个最常见的坑和解决办法。

6.1 Playwright爬虫常见问题

问题1：元素找不到（Selector失效）

• 现象：page.wait_for_selector超时，或page.query_selector返回None。
• 原因：
  1. 选择器写错了，或者网页结构变了。
  2. 元素在iframe内。
  3. 页面是动态渲染的，需要更长的等待时间或等待特定事件。
• 排查与解决：
  • 手动验证：在浏览器的开发者工具（F12）中，使用$$('你的选择器')测试选择器是否有效。
  • 使用更稳定的选择器：优先使用id、>frame = page.frame_locator('iframe[name="content"]').first # 定位iframe element = frame.locator('.target-element').first # 在iframe内定位元素
  • 调整等待策略：尝试page.wait_for_function等待某个JavaScript条件成立，或者使用page.wait_for_timeout增加固定等待时间（不推荐为首选）。
  • 开启非无头模式调试：设置HEADLESS=False，亲眼看看页面加载到了哪一步。

问题2：被网站检测为机器人

• 现象：弹出验证码、访问被拒绝、返回异常内容。
• 原因：Playwright的浏览器指纹虽然接近真人，但一些高级反爬系统仍能通过行为模式（如无间隔快速点击）进行识别。
• 解决：
  • 增加操作延迟：slow_mo参数非常有用，它让每个Playwright操作之间都有延迟，模拟人类的不确定性。
  • 随机化行为：在点击、滚动等操作前，加入随机的page.wait_for_timeout(random.randint(1000, 3000))。
  • 使用更真实的上下文：在browser.new_context()时，设置viewport、user_agent，甚至可以加载真实的浏览器用户数据目录（user_data_dir）来使用你的真实cookies和历史记录（需谨慎处理隐私）。
  • 代理IP：对于高频率抓取，考虑使用代理IP池轮换IP。Playwright启动浏览器时可以配置代理服务器。

6.2 大模型API调用常见问题

问题1：API返回错误（如429， 401）

• 429 Too Many Requests：请求频率超限。必须在代码中增加请求间隔（time.sleep），并考虑实现一个简单的令牌桶算法来控制速率。
• 401 Unauthorized：API Key错误或过期。检查config.py中的密钥是否正确，是否有空格。确认该API Key是否有调用该模型的权限。
• 400 Bad Request：请求格式错误或参数不对。仔细对照官方API文档，检查请求的URL、Headers和Body格式。特别是messages的数组结构。

问题2：总结效果不佳

• 现象：摘要偏离主题、遗漏关键信息、或格式不符合要求。
• 优化方向：
  1. 精炼Prompt：在Prompt中更明确地指示。例如：“请用中文总结，重点提取人物、事件、时间、地点、原因这五个要素。”或者“摘要风格请模仿专业财经记者。”
  2. 提供示例（Few-Shot）：在Prompt中给出一两个输入输出的例子，让AI模仿。
  3. 调整模型参数：降低temperature（如0.1）使输出更稳定；提高max_tokens给AI更多发挥空间。
  4. 分步总结：对于长文章，可以先让AI提取关键段落，再对关键段落进行总结。

6.3 性能与稳定性优化建议

1. 异步并发：如果抓取多个独立网站，可以使用Playwright的异步API（async_playwright）和asyncio.gather并发执行，大幅缩短总耗时。但要注意目标网站的承受能力，避免过于激进。
2. 断点续爬：将已成功抓取的文章URL记录到一个文件中或数据库。每次启动时先加载这个记录，跳过已处理过的URL。这对于定期运行的脚本非常有用。
3. 日志系统：使用Python的logging模块将不同级别的信息（INFO, WARNING, ERROR）输出到文件和控制台，方便后期排查问题。
4. 配置分离：将网站选择器、登录信息等易变的内容完全放在config.py或更外部的配置文件中。当网站改版时，只需修改配置，无需改动核心代码。

这个项目从环境搭建到最终运行，覆盖了从浏览器自动化到AI应用的关键环节。代码块中的内容可以直接复制到你的编辑器中，替换掉配置信息后就能运行。最开始的几次运行建议把HEADLESS设为False，看着浏览器自己操作，既能确认脚本行为是否符合预期，也相当有成就感。遇到问题别慌，多看看Playwright官方文档和对应大模型平台的API文档，大部分问题都有解决方案。