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

📅 2026/7/10 1:45:41
基于Playwright与AI的新闻聚合与智能摘要系统实战
1. 项目概述为什么选择Playwright与AI做新闻聚合最近在折腾一个能自动抓新闻、自动总结的“信息助理”核心就两件事一是让程序像人一样去网站上点开新闻、翻页、甚至登录把内容扒下来二是让AI快速消化这些内容给我一个简洁的摘要。听起来挺复杂但用对工具这事儿对新手也友好。我最终选定的技术栈是Playwright搭配大模型API原因很简单Playwright是目前对现代网页支持最好、最稳定的浏览器自动化工具而大模型比如GPT、文心一言、通义千问的API处理文本总结是降维打击。这个项目的价值在于它能帮你从信息洪流中解放出来。想象一下你每天需要关注十几个不同网站的行业动态手动一个个点开看效率极低。这个脚本可以定时运行自动收集所有你关心的新闻并生成一份结构化的简报。无论是用于个人学习、市场调研还是内容创作素材收集都非常实用。整个流程完全自动化从打开浏览器、模拟点击、抓取内容到调用AI进行总结一气呵成。下面我就把这个“开箱即用”的保姆级方案拆解给你代码可以直接复制修改。2. 环境准备与核心工具选型2.1 Python环境与Playwright安装首先确保你的电脑上安装了Python建议3.8及以上版本。打开命令行Windows的CMD/PowerShellMac/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的选择与配置我们需要一个能处理文本总结的大模型接口。这里有几个主流选择OpenAI GPT系列效果稳定但需要海外网络环境及付费。国内大厂API如百度文心一言、阿里通义千问、讯飞星火等。这些对国内用户更友好访问速度快通常也提供免费额度供测试。本地部署大模型如使用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(levellogging.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(headlessHEADLESS, slow_moSLOW_MO) # 创建上下文可以设置视窗大小、User-Agent等 self.context self.browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/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[typesubmit], username: your_user, password: your_pass} logger.info(f正在导航至登录页: {login_url}) page.goto(login_url, timeoutTIMEOUT) 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(pathstate.json) def crawl_article_list(self, page, list_url, article_selector): 抓取列表页的文章链接 :return: 文章链接列表 logger.info(f正在抓取列表页: {list_url}) page.goto(list_url, timeoutTIMEOUT) # 等待文章列表元素加载 page.wait_for_selector(article_selector, timeoutTIMEOUT) # 获取所有文章链接元素 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, timeoutTIMEOUT) # 等待正文内容加载 page.wait_for_selector(content_selector, timeoutTIMEOUT) # 提取标题 - 通常可以从页面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_articles4. 核心模块二连接大模型进行智能总结抓取到原始文本后我们需要调用大模型API来生成摘要。这里的关键是构造一个清晰的提示词Prompt让AI理解我们的需求。4.1 构建高效的总结提示词Prompt提示词的质量直接决定总结效果。一个好的新闻总结Prompt应该包含角色定义让AI明确自己的任务。输入格式说明告诉AI你给它的是什么。输出格式要求明确需要AI返回什么最好是结构化数据如JSON。具体任务指令总结的重点是什么如核心事件、观点、数据、后续影响。# 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: fBearer {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, # 常见端点需根据平台调整 headersself.headers, datajson.dumps(data), timeout60 ) 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(fAPI请求失败: {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 None4.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 fnews_summary_{datetime.now().strftime(%Y%m%d_%H%M%S)}.json with open(output_file, w, encodingutf-8) as f: json.dump(all_results, f, ensure_asciiFalse, indent2) 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, filenamenews_report.md): 将结果生成为Markdown格式的报告 with open(filename, w, encodingutf-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(fMarkdown报告已生成: {filename})6. 常见问题与排查技巧实录在实际运行中你肯定会遇到各种问题。这里记录了几个最常见的坑和解决办法。6.1 Playwright爬虫常见问题问题1元素找不到Selector失效现象page.wait_for_selector超时或page.query_selector返回None。原因选择器写错了或者网页结构变了。元素在iframe内。页面是动态渲染的需要更长的等待时间或等待特定事件。排查与解决手动验证在浏览器的开发者工具F12中使用$$(你的选择器)测试选择器是否有效。使用更稳定的选择器优先使用id、>frame page.frame_locator(iframe[namecontent]).first # 定位iframe element frame.locator(.target-element).first # 在iframe内定位元素调整等待策略尝试page.wait_for_function等待某个JavaScript条件成立或者使用page.wait_for_timeout增加固定等待时间不推荐为首选。开启非无头模式调试设置HEADLESSFalse亲眼看看页面加载到了哪一步。问题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调用常见问题问题1API返回错误如429 401429 Too Many Requests请求频率超限。必须在代码中增加请求间隔time.sleep并考虑实现一个简单的令牌桶算法来控制速率。401 UnauthorizedAPI Key错误或过期。检查config.py中的密钥是否正确是否有空格。确认该API Key是否有调用该模型的权限。400 Bad Request请求格式错误或参数不对。仔细对照官方API文档检查请求的URL、Headers和Body格式。特别是messages的数组结构。问题2总结效果不佳现象摘要偏离主题、遗漏关键信息、或格式不符合要求。优化方向精炼Prompt在Prompt中更明确地指示。例如“请用中文总结重点提取人物、事件、时间、地点、原因这五个要素。”或者“摘要风格请模仿专业财经记者。”提供示例Few-Shot在Prompt中给出一两个输入输出的例子让AI模仿。调整模型参数降低temperature如0.1使输出更稳定提高max_tokens给AI更多发挥空间。分步总结对于长文章可以先让AI提取关键段落再对关键段落进行总结。6.3 性能与稳定性优化建议异步并发如果抓取多个独立网站可以使用Playwright的异步APIasync_playwright和asyncio.gather并发执行大幅缩短总耗时。但要注意目标网站的承受能力避免过于激进。断点续爬将已成功抓取的文章URL记录到一个文件中或数据库。每次启动时先加载这个记录跳过已处理过的URL。这对于定期运行的脚本非常有用。日志系统使用Python的logging模块将不同级别的信息INFO, WARNING, ERROR输出到文件和控制台方便后期排查问题。配置分离将网站选择器、登录信息等易变的内容完全放在config.py或更外部的配置文件中。当网站改版时只需修改配置无需改动核心代码。这个项目从环境搭建到最终运行覆盖了从浏览器自动化到AI应用的关键环节。代码块中的内容可以直接复制到你的编辑器中替换掉配置信息后就能运行。最开始的几次运行建议把HEADLESS设为False看着浏览器自己操作既能确认脚本行为是否符合预期也相当有成就感。遇到问题别慌多看看Playwright官方文档和对应大模型平台的API文档大部分问题都有解决方案。