Python新闻追踪器:基于网络爬虫与关键词过滤的个性化信息聚合工具
1. 项目概述:一个新闻追踪器的诞生与价值
最近在整理自己的信息流时,发现了一个挺普遍的问题:每天被各种App推送的新闻轰炸,但真正想持续跟进的某个特定事件、某个公司动态或者某个技术领域的最新进展,却总是淹没在信息洪流里。手动去各个网站搜索,效率低不说,还容易遗漏。于是,一个想法就冒了出来——能不能自己写个工具,让它自动帮我追踪我关心的新闻,并且把结果整理好推送给我?这就是newstracer这个项目最初的由来。
简单来说,newstracer是一个基于Python的新闻追踪与聚合工具。它的核心功能是,你可以给它一个或多个你关心的“关键词”或“主题”,它会定期(比如每天或每小时)自动去指定的新闻源(如主流新闻网站、科技博客、RSS订阅源)进行抓取和扫描,然后通过智能过滤和去重,将包含这些关键词的最新报道或文章,通过邮件、即时通讯工具(如Telegram Bot)或者直接生成一个摘要页面推送给你。它解决的核心痛点就是信息过载下的精准获取,让你从被动的信息接收者,变为主动的信息管理者。
这个项目特别适合几类朋友:一是对某个垂直领域(如人工智能、金融市场、特定行业政策)需要保持高度敏感度的从业者或研究者;二是关注特定公司或产品动态的市场、产品人员;三是像我一样的普通用户,只是想更高效地跟进几个长期兴趣点,避免在无关信息上浪费时间。从技术上看,它涉及网络爬虫、文本处理、任务调度和消息推送等多个环节,是一个典型的“胶水”项目,能把多个实用的技术点串联起来解决一个实际需求。
2. 核心设计思路与技术选型
2.1 需求拆解与架构设计
动手之前,我先仔细拆解了一下需求。一个新闻追踪器,听起来简单,但细想下来有几个关键模块必须处理好:
- 数据获取(Crawler/ Fetcher):从哪里获取新闻?频率如何?如何应对网站反爬?
- 内容解析与过滤(Parser & Filter):抓回来的HTML页面或RSS源,如何提取出干净的标题、正文、发布时间?如何根据我设定的关键词进行过滤?
- 去重与存储(Deduplicator & Storage):同一条新闻可能被多个源报道,如何识别并去重?历史数据存哪里,方便后续查询或分析趋势?
- 通知推送(Notifier):过滤后的新内容,通过什么渠道、以什么格式通知我?
- 任务调度(Scheduler):如何让整个流程自动化、周期性地运行?
基于这些模块,我设计了一个简单清晰的数据流架构:调度器触发抓取任务 -> 抓取器从各新闻源获取原始数据 -> 解析器提取结构化信息 -> 过滤器根据关键词匹配 -> 去重器对比历史记录 -> 将新内容存入数据库并触发推送器。整个架构是松耦合的,每个模块都可以独立替换或升级,比如今天用RSS源,明天可以增加API接口;今天推邮件,明天可以加个钉钉机器人。
2.2 关键技术栈选型与考量
技术选型上,我遵循“成熟、高效、易于维护”的原则,主要基于Python生态。
爬虫与解析:
requests+BeautifulSoup4:这是处理静态网页的黄金组合。requests负责网络请求,简单易用;BeautifulSoup4负责解析HTML,定位并提取所需数据。对于大多数新闻网站,这套组合拳足够用了。feedparser:对于提供了RSS/Atom订阅源的网站,这是首选。它比解析HTML更稳定、更规范,能直接获取结构化的标题、链接、摘要和发布时间,极大降低了开发复杂度。- 备用方案:
Selenium/Playwright:考虑到一些现代网站大量使用JavaScript渲染,纯requests无法获取内容。我预留了使用无头浏览器(Headless Browser)的接口。但这会显著增加资源消耗和运行时间,所以仅作为针对特定“顽固”目标的备选方案。
文本处理与过滤:
- 核心:正则表达式与字符串操作:初级的关键词过滤,直接用
in操作符或正则表达式匹配标题和正文即可。为了更灵活,我支持了多种匹配模式,如全词匹配、模糊匹配(包含子串)、以及多个关键词的“与/或”逻辑组合。 - 进阶:
jieba分词与TF-IDF(可选):如果需求升级,比如想追踪一个“概念”而非具体关键词(例如追踪“量子计算”相关的所有文章,而不仅限于出现这四个字的),可以引入中文分词库jieba和简单的TF-IDF算法来计算文章与主题的相关性。在newstracer的初期版本中,我将其作为可扩展的插件点,并未强制集成,以保持核心轻量。
- 核心:正则表达式与字符串操作:初级的关键词过滤,直接用
数据存储:
- SQLite:项目首选。它是一个单文件数据库,无需安装和配置独立的数据库服务,非常适合这种个人使用的、数据量不大的工具。我用它来存储已抓取文章的唯一标识(如URL的MD5值)、标题、链接、来源、抓取时间等,主要用于去重和历史查询。
- 可选:Redis:如果未来需要实现更快的去重判断(基于内存的集合操作)或者做简单的缓存,Redis是很好的选择。目前用SQLite的简单表查询已能满足需求。
任务调度:
schedule库:一个轻量级的、人性化的Python任务调度库。可以用类似schedule.every(1).hours.do(job)这样的语法来定义周期任务,非常直观。它运行在单进程中,适合本机或服务器上作为常驻脚本运行。- 系统级方案:Cron (Linux/macOS) 或 Task Scheduler (Windows):更传统和可靠的方式。将Python脚本写好,然后用系统的定时任务工具来调用。这种方式与程序本身解耦,更利于管理和监控。
newstracer的设计兼容这两种方式,脚本可以一次性运行,也可以被外部调度器周期性触发。
消息推送:
- 电子邮件 (
smtplib+email):最通用、最稳定的方式。几乎人人都有邮箱,手机也能实时提醒。我实现了通过SMTP服务器发送HTML格式的邮件,可以将多条新闻整理成一个美观的摘要。 - Telegram Bot:实时性更强,交互也更方便。Telegram提供了非常友好的Bot API,可以很方便地实现发送消息、甚至接收一些简单指令(如临时添加关键词)。这对于希望随时收到提醒的用户体验更好。
- 其他:架构上留出了推送接口,理论上可以接入钉钉、企业微信、Slack等任何支持Webhook的通讯工具。
- 电子邮件 (
注意:在编写爬虫时,务必遵守网站的
robots.txt协议,并设置合理的请求间隔(如每次请求后time.sleep(2-5)秒),避免对目标服务器造成压力,这也是基本的网络礼仪和规避反爬虫策略的有效手段。
3. 核心模块实现与实操详解
3.1 新闻源配置与抓取器实现
新闻追踪的源头是新闻源。我设计了一个灵活的配置系统,支持多种类型的源。
1. RSS/Atom 源配置:这是最推荐的方式。在配置文件(如sources.yaml或sources.json)里,可以这样配置:
rss_sources: - name: "Solidot" url: "https://www.solidot.org/index.rss" category: "科技" - name: "Reuters Technology" url: "http://feeds.reuters.com/reuters/technologyNews" category: "科技" language: "en"抓取器会遍历这个列表,使用feedparser解析每个URL。feedparser.parse(url)返回一个包含所有条目的对象,我们可以轻松地获取entry.title,entry.link,entry.published(发布时间,需解析)和entry.summary(摘要)。
2. 自定义爬虫(针对无RSS的网站):对于没有RSS的网站,需要编写特定的抓取和解析函数。例如,抓取某个新闻板块:
import requests from bs4 import BeautifulSoup def crawl_custom_news(): url = "https://example.com/news" headers = {'User-Agent': 'Mozilla/5.0...'} # 模拟浏览器头 try: resp = requests.get(url, headers=headers, timeout=10) resp.raise_for_status() # 检查请求是否成功 soup = BeautifulSoup(resp.content, 'html.parser') news_list = [] # 假设新闻在 <div class='news-item'> 里 for item in soup.select('div.news-item'): title_elem = item.select_one('h2 a') if not title_elem: continue title = title_elem.text.strip() link = title_elem['href'] # 补全可能为相对路径的链接 if link.startswith('/'): link = requests.compat.urljoin(url, link) # 尝试获取发布时间 time_elem = item.select_one('.time') pub_time = parse_time(time_elem.text) if time_elem else None news_list.append({'title': title, 'link': link, 'pub_time': pub_time}) return news_list except requests.RequestException as e: print(f"抓取 {url} 失败: {e}") return []这里的关键在于使用select或find方法,通过CSS选择器或标签属性精准定位到需要的元素。每个网站的页面结构都不同,所以每新增一个此类源,都需要单独分析其HTML结构并编写解析逻辑。建议将每个源的解析函数独立成模块,便于维护。
3.2 内容过滤与关键词匹配引擎
抓取到文章列表后,下一步是根据用户兴趣进行过滤。我实现了一个可配置的过滤引擎。
基础关键词匹配:用户在配置文件中定义自己关心的关键词列表,以及匹配规则。
keywords: - "人工智能" - "AI" - "大模型" - "LLM" match_mode: "or" # 匹配模式:'or' (任一关键词命中) 或 'and' (全部关键词命中) case_sensitive: false # 是否区分大小写过滤函数会遍历每篇文章的标题和摘要(或正文,如果已抓取),检查是否包含列表中的关键词。match_mode为or时,只要包含任意一个关键词即视为匹配;为and时,则需包含所有关键词(适用于非常精确的主题)。
进阶:正则表达式与排除词:为了更灵活,我支持了正则表达式模式,并增加了“排除词”功能。
keyword_patterns: - pattern: "苹果(公司|发布会)?" is_regex: true # 使用正则匹配,能匹配“苹果”、“苹果公司”、“苹果发布会” exclude_words: - "吃的苹果" # 排除与水果苹果相关的内容这样,当匹配到“苹果”时,如果文章内容中也出现了“吃的苹果”,这条新闻可能会被过滤掉(取决于排除逻辑的严格程度)。这能有效减少误报。
实操心得:关键词的设置是一门艺术。太宽泛(如“科技”)会抓到太多无关信息;太具体(如某个产品的完整型号)可能会遗漏相关报道。我的建议是:从核心术语开始,逐步根据推送结果进行调整。例如,追踪“电动汽车”时,可以同时加上“特斯拉”、“比亚迪”、“蔚来”等具体品牌,以及“电池”、“续航”、“自动驾驶”等相关技术词,形成一个小的关键词网络。
3.3 持久化存储与高效去重策略
为了避免重复推送同一条新闻,并保留历史记录,存储和去重是必不可少的。
1. 数据库设计:我使用SQLite,创建了一张简单的articles表。
CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash TEXT UNIQUE NOT NULL, -- 文章URL的哈希值,用于去重唯一标识 title TEXT NOT NULL, url TEXT NOT NULL, source TEXT NOT NULL, published_at TIMESTAMP, -- 新闻发布时间 fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, -- 抓取时间 summary TEXT -- 摘要或正文片段 );url_hash字段是去重的关键。通常对文章的URL进行MD5或SHA256哈希,生成一个固定长度的字符串作为唯一标识。相比直接存储URL,哈希值长度固定,索引效率更高。
2. 去重流程:每次抓取并过滤出一批新文章后,在存入数据库前,先计算每个文章的url_hash,然后查询数据库中是否已存在。
import hashlib import sqlite3 def get_url_hash(url): return hashlib.md5(url.encode('utf-8')).hexdigest() def is_duplicate(conn, url_hash): cursor = conn.cursor() cursor.execute("SELECT 1 FROM articles WHERE url_hash = ?", (url_hash,)) return cursor.fetchone() is not None只有is_duplicate返回False的文章,才会被插入数据库,并加入待推送列表。
3. 存储优化考虑:
- 索引:务必在
url_hash字段上创建索引,可以极大加速去重查询的速度。CREATE INDEX idx_url_hash ON articles (url_hash); - 数据清理:新闻的时效性很强,可以定期(比如每月)清理3个月或更久以前的数据,控制数据库大小。可以在调度任务中增加一个清理子任务。
3.4 多渠道通知推送实现
将过滤后的新文章推送给用户,是价值实现的最后一步。我实现了邮件和Telegram Bot两种主要方式。
1. 邮件推送:使用Python内置的smtplib和email库。需要先配置发件邮箱的SMTP服务器信息(如QQ邮箱、163邮箱或Gmail的SMTP)。
import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from datetime import datetime def send_email(subject, html_content, to_emails, smtp_config): msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = smtp_config['sender'] msg['To'] = ', '.join(to_emails) # 创建HTML版本 html_part = MIMEText(html_content, 'html', 'utf-8') msg.attach(html_part) try: with smtplib.SMTP_SSL(smtp_config['host'], smtp_config['port']) as server: server.login(smtp_config['user'], smtp_config['password']) server.send_message(msg) print(f"[{datetime.now()}] 邮件发送成功") except Exception as e: print(f"[{datetime.now()}] 邮件发送失败: {e}")可以设计一个漂亮的HTML模板,将多条新闻以列表形式清晰展示,包含标题(带链接)、来源、时间和简短摘要。
2. Telegram Bot推送:首先需要在Telegram上找@BotFather创建一个Bot,获取API Token。
import requests TELEGRAM_BOT_TOKEN = 'YOUR_BOT_TOKEN' TELEGRAM_CHAT_ID = 'YOUR_CHAT_ID' # 你的个人或群组Chat ID def send_telegram_message(text): url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" payload = { 'chat_id': TELEGRAM_CHAT_ID, 'text': text, 'parse_mode': 'HTML' # 支持简单HTML格式,如加粗、链接 } try: resp = requests.post(url, json=payload, timeout=10) resp.raise_for_status() return True except requests.RequestException as e: print(f"Telegram推送失败: {e}") return FalseTelegram消息更即时,适合推送突发新闻。可以将多条新闻合并成一条消息发送,避免刷屏。更高级的玩法,可以让Bot接收命令,动态修改追踪关键词。
提示:无论是邮箱密码还是Telegram Bot Token,都属于敏感信息。绝对不要将它们硬编码在脚本里或提交到公开的代码仓库。务必使用环境变量或单独的配置文件(如
config.ini),并将该配置文件加入.gitignore。
4. 系统集成、部署与自动化运行
4.1 配置管理与项目组织
一个清晰的项目结构能让维护变得轻松。我的newstracer项目目录大致如下:
newstracer/ ├── config/ │ ├── config.yaml # 主配置:数据库路径、推送方式、全局开关 │ └── sources.yaml # 新闻源定义 ├── core/ │ ├── __init__.py │ ├── fetcher.py # 抓取器抽象与RSS/自定义爬虫实现 │ ├── parser.py # 内容解析器 │ ├── filter.py # 关键词过滤引擎 │ ├── deduplicator.py # 去重逻辑(与数据库交互) │ ├── notifier.py # 邮件、Telegram等推送器 │ └── scheduler.py # 任务调度封装 ├── scripts/ │ └── run_tracer.py # 主运行脚本 ├── data/ │ └── news.db # SQLite数据库文件(.gitignore忽略) ├── logs/ # 日志目录(.gitignore忽略) ├── requirements.txt # Python依赖列表 └── README.md # 项目说明文档config.yaml示例:
database: path: "./data/news.db" notification: email: enabled: true smtp_host: "smtp.example.com" smtp_port: 465 sender: "your-email@example.com" receivers: - "receiver1@example.com" telegram: enabled: false # 按需开启 bot_token: "YOUR_TOKEN" chat_id: "YOUR_CHAT_ID" schedule: interval_minutes: 60 # 每60分钟运行一次(当使用schedule库时)通过配置文件,可以灵活地开关功能、调整参数,而无需修改代码。
4.2 主流程串联与错误处理
主脚本run_tracer.py的工作流非常直接:
# run_tracer.py import logging from core.fetcher import fetch_all_articles from core.filter import filter_articles from core.deduplicator import Deduplicator from core.notifier import notify_new_articles def main(): # 1. 初始化日志和去重器 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') deduper = Deduplicator('config/database_path') # 2. 抓取所有配置源的文章 all_raw_articles = fetch_all_articles() logging.info(f"共抓取到 {len(all_raw_articles)} 条原始文章。") # 3. 关键词过滤 filtered_articles = filter_articles(all_raw_articles) logging.info(f"关键词过滤后剩余 {len(filtered_articles)} 条。") # 4. 去重 new_articles = [] for article in filtered_articles: if not deduper.is_duplicate(article['url']): deduper.save_article(article) new_articles.append(article) logging.info(f"去重后,新增 {len(new_articles)} 条文章。") # 5. 推送新文章 if new_articles: notify_new_articles(new_articles) else: logging.info("没有新文章需要推送。") if __name__ == "__main__": main()错误处理是保证长期稳定运行的关键。在上述每个步骤(网络请求、解析、数据库操作、网络推送)都需要用try...except包裹,并记录详细的错误日志,而不是让整个程序崩溃。例如,某个新闻源临时不可用,应该记录错误并跳过,继续处理其他源。
4.3 部署方案:从本地到服务器
如何让这个脚本“永远”在后台运行?
本地开发/测试:直接运行
python run_tracer.py。可以使用schedule库让脚本循环运行,或者更简单地,用while True循环配合time.sleep(interval)。个人电脑后台运行(macOS/Linux):
- 使用
nohup命令:nohup python run_tracer.py > tracer.log 2>&1 &。这样即使关闭终端,脚本也会在后台运行,日志输出到tracer.log。 - 使用
tmux或screen会话:创建一个持久化的会话运行脚本,可以随时断开和重连查看。
- 使用
服务器部署(推荐):
- 使用系统Cron:这是最稳定、最标准的方式。将脚本修改为一次性任务(即运行一次就结束),然后通过Cron定时调用。
# 编辑当前用户的cron任务 crontab -e # 添加一行,例如每30分钟运行一次 */30 * * * * cd /path/to/newstracer && /usr/bin/python3 run_tracer.py >> /path/to/newstracer/cron.log 2>&1 - 使用Supervisor进程管理:如果脚本需要作为常驻进程(例如内部使用了
schedule库循环),可以用Supervisor来管理,它能保证进程意外退出后自动重启,并方便地管理日志。; /etc/supervisor/conf.d/newstracer.conf [program:newstracer] command=/usr/bin/python3 /path/to/newstracer/run_tracer.py directory=/path/to/newstracer autostart=true autorestart=true stderr_logfile=/var/log/newstracer.err.log stdout_logfile=/var/log/newstracer.out.log
- 使用系统Cron:这是最稳定、最标准的方式。将脚本修改为一次性任务(即运行一次就结束),然后通过Cron定时调用。
容器化部署(Docker):为了环境一致性,可以创建Docker镜像。编写
Dockerfile,安装依赖,设置启动命令为python run_tracer.py,然后使用Cron或Kubernetes的CronJob来调度容器运行。这对于在云服务器上管理多个此类小工具非常整洁。
5. 常见问题排查与优化经验
5.1 抓取失败与反爬虫应对
问题1:请求被拒绝,返回403或429状态码。
- 原因:网站检测到爬虫行为(如无User-Agent、请求过快)。
- 解决:
- 设置合理的请求头:始终在
requests.get()中模拟一个常见的浏览器User-Agent。 - 添加延迟:在连续请求之间加入随机延时,例如
time.sleep(random.uniform(2, 5))。 - 使用代理IP池(针对严格封锁):对于大规模抓取,可能需要轮换使用不同的IP地址。个人小规模使用通常不需要。
- 尊重robots.txt:使用
urllib.robotparser检查目标网站是否允许爬取特定路径。
- 设置合理的请求头:始终在
问题2:页面内容为空或结构不符,解析失败。
- 原因:网站改版,或内容由JavaScript动态加载。
- 解决:
- 定期检查与更新解析规则:自定义爬虫的解析逻辑(CSS选择器)需要随网站改版而更新。
- 切换到无头浏览器:对于JS渲染的页面,使用
Selenium或Playwright。但这会慢很多,仅作为最后手段。可以先检查网站是否有“移动版”或“简化版”页面,这些页面通常结构更简单。 - 寻找替代源:看看该网站是否提供了官方的RSS源或API,这是最稳定友好的方式。
5.2 误报与漏报的调优
问题:推送的内容不相关(误报),或者相关的内容没抓到(漏报)。
- 原因:关键词设置不合理,或者过滤只在标题/摘要进行。
- 优化:
- 精细化关键词:使用更长的短语、添加排除词、利用正则表达式精确匹配。
- 引入摘要或正文分析:如果RSS只提供标题,漏报率会很高。可以考虑抓取文章链接的正文前几段进行分析,但这会增加抓取负担和复杂度。一个折中方案是,对初步过滤后的文章,再发起一次轻量级请求获取摘要。
- 人工反馈循环:在推送的消息里,可以加入“这条不相关”的快速反馈按钮(Telegram Bot可以做到)。收集这些反馈,用来动态调整关键词权重或作为训练数据(如果未来想引入机器学习模型)。
5.3 性能与稳定性提升
问题:随着追踪源增多,单次运行时间变长,可能错过实时性。
- 优化:
- 异步抓取:使用
aiohttp和asyncio库进行异步HTTP请求,可以同时抓取多个新闻源,极大缩短总抓取时间。 - 增量抓取与缓存:对于RSS源,可以记录上次抓取的最新条目发布时间,下次只请求比这个时间更新的条目。对于自定义爬虫,可以分析网站的分页或时间归档规律,进行增量抓取。
- 数据库连接池与批量操作:如果推送量很大,频繁开关数据库连接会影响性能。可以使用连接池,并在去重和保存时采用批量操作(
executemany)。
- 异步抓取:使用
问题:脚本运行一段时间后内存占用高或意外退出。
- 解决:
- 资源释放:确保数据库连接、网络会话在使用后正确关闭。使用
with语句上下文管理器是很好的习惯。 - 异常捕获与日志:在最外层进行异常捕获,记录错误并优雅退出或重试,避免静默崩溃。
- 使用进程管理工具:如前所述,使用Supervisor或systemd来管理进程,可以自动重启失败的脚本。
- 资源释放:确保数据库连接、网络会话在使用后正确关闭。使用
5.4 扩展思路与高级玩法
当基础功能稳定后,可以考虑一些扩展方向,让工具更强大:
- 情感分析与趋势判断:对抓取到的新闻正文进行简单的情感分析(使用预训练模型或词典),判断舆论是正面还是负面。长期积累数据后,可以绘制某个关键词的“舆情热度趋势图”。
- 多语言支持:使用翻译API(如Google Translate API或开源库),将外文新闻的关键部分翻译后推送,实现跨语言信息追踪。
- 生成式摘要:对于长篇报道,可以调用大模型的API(如ChatGPT、文心一言等),生成一段简洁的要点摘要,附在推送信息里,节省阅读时间。
- 可视化仪表盘:将数据库中的历史数据用Flask或Streamlit做成一个简单的Web仪表盘,展示新闻数量随时间的变化、不同来源的占比等。
- 协同过滤与推荐:如果有多人使用,可以匿名收集大家的订阅关键词和点击行为,实现简单的“关注了X的人也关注了Y”的推荐功能。
这个项目的魅力在于,它从一个具体的个人需求出发,串联起了软件开发的多个核心环节:需求分析、架构设计、编码实现、部署运维。你可以根据自己的技术兴趣,在其中任何一个环节进行深化和拓展。它可能永远只是一个“小工具”,但正是这些解决实际问题的“小工具”,构成了我们数字生活的高效基石。
