Python实现新闻资讯聚合系统：RSS抓取、智能去重与实时推送实战

ABC晚间新闻类资讯聚合系统开发实战RSS抓取、去重分类与实时推送如果你做过新闻资讯类 App、天气预警小程序或者企业内部情报监控系统大概率会被同一件事折磨过信源太多、格式太乱、重复内容太多、时效性又强。尤其是“晚间新闻”这类多主题摘要每一条都涉及不同领域天气、交通、文化、社会事件混在一起靠人去拆分整理累且容易漏。我见过很多团队的第一版资讯系统本质上就是“定时抓取一批 RSS把标题和正文塞进数据库前端按时间倒序展示”。结果上线后问题全出来了同一事件被五六家媒体重复发布展示列表全是重复标题抓下来的正文带着广告和推荐链接分类全是“未分类”凌晨突发新闻过了两小时才被系统抓下来。这篇文章不是讲怎么用现成的新闻 API 一键接入。我会以“ABC晚间新闻 20260809”这类多主题新闻摘要为背景完整拆解一个资讯聚合系统的核心模块RSS/API 多源抓取、内容清洗、相似去重、主题分类、定时任务与实时推送。整体代码使用 Python方案可以平移到 Java、Go 或 Node.js 项目。读完之后你能跑通一个最小可用的资讯聚合服务并且知道生产环境还有哪些坑等着你。1. 资讯聚合系统要解决的四个核心问题抛开具体业务所有资讯聚合系统做的事情都是同一个把多个来源的信息变成一份干净、有序、及时的清单。在这一步最值得先想清楚的不是用什么框架而是你要处理哪几类问题。第一是信源异构。有的源提供标准 RSS有的源只提供 JSON API有的源什么接口都没有只能靠 HTML 解析。你不能假设所有信源都长得一样。系统设计的第一原则就是把“抓取”和“解析”解耦每类信源只写一个适配器。第二是内容噪音。抓下来的正文里通常混着导航、广告、相关推荐、版权声明。如果不过滤就直接入库很快数据库里就会塞满垃圾信息搜索和推荐质量都会崩掉。清洗不是锦上添花而是必要步骤。第三是重复内容。同一事件被多个媒体转载、改写、洗稿标题相似、正文却不完全一致。只靠 URL 去重完全没用必须做基于文本相似度的去重。第四是时效性。新闻系统里数据晚到十分钟价值可能就完全不一样了。定时轮询的间隔、抓取超时时间、失败重试策略都要围绕时效性来设计。回到“ABC晚间新闻”这类摘要。它的一条新闻里可能同时包含天气灾害、航班动态、艺术品失窃等话题这正好对应了聚合系统的分类诉求你需要把不同主题的内容自动分到不同栏目而不是让用户自己在混杂列表里找。所以这篇文章的核心思路是先用最小系统跑通“抓取 - 清洗 - 去重 - 分类 - 推送”的全链路再针对生产环境做加固。2. RSS、API、网页解析三种数据源接入方式对比信息采集的第一步是拿数据。三种主流方式各有优劣实际项目往往是混合使用。2.1 RSS/Atom 订阅源RSS 是最传统也最稳定的方式。只要对方提供了标准 RSS 输出你只需要发一个 HTTP 请求解析 XML 就能拿到标题、链接、发布时间、正文摘要。RSS 的优点结构标准化字段清晰。服务器压力小很多源支持条件请求。实现成本最低。RSS 的缺点部分站点不提供或者不再更新 RSS。RSS 中的正文常常是摘要不是全文。字段缺失情况多比如作者、封面图经常没有。2.2 JSON API很多新闻平台和内容服务商提供官方 API返回结构化 JSON。相比 RSSAPI 的字段更丰富通常包含分类、标签、图片、作者。API 的优点数据结构完整不需要额外解析。权限控制方便可以按账号管理配额。更新频率和粒度可控。API 的缺点需要申请 key有配额限制。不同服务商的响应结构完全不同。收费或者免费额度限制。2.3 HTML 网页解析当一个信源既没有 RSS也没有 API只能从 HTML 页面里抽取正文。这种方式最灵活也最脆弱。HTML 解析的常见手段XPath 定位正文节点。基于正文文本密度算法自动抽取。用 Readability 类库提取正文。网页解析是最后手段。对方前端结构一改你的解析器就失效必须配合监控告警。2.4 三种方式的选择建议从材料看多数大众新闻源都保留了 RSS 输出。对于个人项目和中小型系统建议第一版全部走 RSS把 JSON API 留到需要更丰富数据时再接入。HTML 解析除非目标站点没有其他方式否则不值得优先投入。表格总结如下接入方式结构化程度维护成本稳定性适用场景RSS/Atom中低高绝大多数新闻源JSON API高中高官方服务、商业数据源HTML 解析低高低无接口的特定站点3. 环境准备与项目结构本文将使用 Python 实现一个最小可用的资讯聚合服务。版本和依赖以通用稳定版本为参考具体版本请以你的运行环境为准。建议环境Python 3.10 或更高版本。依赖库requests、feedparser、jieba、scikit-learn、APScheduler。数据库先用 SQLite 做演示生产环境建议换成 PostgreSQL。建议以虚拟环境运行python3 -m venv venv source venv/bin/activate pip install requests feedparser jieba scikit-learn apscheduler这里简单说明依赖的用途requests抓取 RSS、调用 API。feedparser解析 RSS/Atom XML。jieba中文分词用于关键词提取和分类特征构建。scikit-learn用 TF-IDF 向量计算文本相似度也可以做分类模型训练。APScheduler管理定时抓取任务。项目结构保持在单个 Python 包内即可news_aggregator/ ├── config.py # 配置文件信源列表、数据库路径、推送参数 ├── fetcher.py # 抓取模块定义数据源适配器 ├── cleaner.py # 内容清洗模块 ├── deduplicator.py # 重复判定模块 ├── classifier.py # 分类模块 ├── storage.py # 存储模块 ├── notifier.py # 推送模块 ├── scheduler.py # 定时任务入口 └── main.py # 一键运行脚本不要一开始就引入复杂的框架。先把链路跑通再决定要不要上 Celery、Airflow 或消息队列是更务实的做法。4. 核心模块设计与实现这一章是整个系统的骨架。每个模块我都会先说清楚职责边界再给核心代码。4.1 配置模块 config.py配置集中放在一个文件里后续所有模块从这个文件读取参数。这样信源变更、阈值调整不需要改业务代码。# 文件路径news_aggregator/config.py DATABASE_PATH news.db # 信源配置支持 rss / api / html 三种类型 SOURCES [ { name: 晚间新闻综合源, type: rss, url: https://example.com/feed.xml, category_hint: 综合, enabled: True, }, ] # 抓取间隔单位分钟 FETCH_INTERVAL_MINUTES 10 # 请求超时单位秒 FETCH_TIMEOUT 15 # 重复判定阈值范围 0-1越接近 1 表示越相似 SIMILARITY_THRESHOLD 0.85 # 分类关键词配置 CATEGORY_KEYWORDS { 天气: [暴雨, 台风, 野火, 高温, 寒潮, 预警], 交通: [航班, 机场, 延误, 返航, 铁路, 公路], 文化: [画作, 博物馆, 展览, 拍卖, 失窃], 科技: [芯片, AI, 模型, 开源, 算法], } # 推送配置 PUSH_WEBHOOK_URL PUSH_TOKEN 这段配置里category_hint是信源自带分类提示最终分类会结合关键词打分一起决定。4.2 抓取模块 fetcher.py抓取模块只做一件事把一个信源的原始内容变成统一的文章结构。每种信源类型写一个抓取函数对外暴露统一的fetch()接口。以 RSS 为例# 文件路径news_aggregator/fetcher.py import time import requests import feedparser from config import FETCH_TIMEOUT def parse_rss(source): resp requests.get(source[url], timeoutFETCH_TIMEOUT) resp.raise_for_status() feed feedparser.parse(resp.content) articles [] for entry in feed.entries: # 部分 RSS 源没有发布时间用当前时间兜底 published getattr(entry, published_parsed, None) if published: publish_time time.strftime(%Y-%m-%d %H:%M:%S, published) else: publish_time time.strftime(%Y-%m-%d %H:%M:%S) articles.append({ title: getattr(entry, title, ).strip(), url: getattr(entry, link, ).strip(), summary: getattr(entry, summary, ).strip(), content: getattr(entry, description, ).strip(), publish_time: publish_time, source: source[name], }) return articles这段代码的要点使用feedparser解析不需要自己写 XML 解析逻辑。每条新闻提取标题、链接、摘要、正文、发布时间、来源名称。published_parsed可能为空需要兜底处理。对外统一入口可以设计成def fetch_all_sources(sources): all_articles [] for source in sources: if not source.get(enabled, True): continue try: if source[type] rss: articles parse_rss(source) # 后续扩展 api 和 html 类型 else: articles [] all_articles.extend(articles) except Exception as e: # 单个信源失败不影响其他信源 print(f[抓取失败] {source[name]}: {e}) return all_articles这里的关键设计是单个信源抓取失败绝对不能中断整个任务。实际线上经常出现某个源超时或返回 500系统要有足够的韧性。4.3 内容清洗模块 cleaner.py清洗是资讯系统最容易低估的模块。RSS 描述字段里经常带 HTML 标签、跟踪链接、图片地址混排甚至是整段的页面框架。清洗步骤去掉 HTML 标签。去掉 JS/CSS 代码片段。去掉常见广告词所在的段落。合并空白字符。截断过长的正文。# 文件路径news_aggregator/cleaner.py import re BAD_PATTERNS [ rvar\s\w\s*, rfunction\s*\w*\s*\(, rhttp[s]?://[^\s], ] AD_KEYWORDS [广告, 推广, 点击购买, 扫码关注, 免责声明] def clean_html(raw_text): if not raw_text: return text raw_text # 去掉script和style标签块 text re.sub(r(script|style)[^]*.*?/\1, , text, flagsre.S | re.I) # 去掉普通 HTML 标签 text re.sub(r[^], , text) # 去掉 URL text re.sub(rhttp[s]?://\S, , text) # 去掉 JS 代码特征 for pattern in BAD_PATTERNS: text re.sub(pattern, , text) # 合并空白 text re.sub(r\s, , text) # 按广告关键词切分只保留第一段之前的内容 for kw in AD_KEYWORDS: idx text.find(kw) if idx ! -1: text text[:idx] return text.strip()对于一条晚间新闻摘要原始内容可能是p美国东西海岸遭遇恶劣天气多地发布预警。/pscripttrack(abc)/script清洗后就只剩下纯文本去掉了干扰信息。这里真正容易踩坑的一点是正则清洗会破坏正文中的中文标点或分段结构所以不要在清洗步骤中做太激进的“正文抽取”先保证干净再保证完整。4.4 存储模块 storage.py存储模块的职责是写入文章、判断 URL 是否已经存在、提供相似度去重所需的候选文章查询。# 文件路径news_aggregator/storage.py import sqlite3 from config import DATABASE_PATH def get_connection(): conn sqlite3.connect(DATABASE_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_connection() conn.execute( CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, url TEXT NOT NULL, summary TEXT, content TEXT, category TEXT, source TEXT, publish_time TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) conn.execute(CREATE INDEX IF NOT EXISTS idx_articles_url ON articles(url)) conn.execute(CREATE INDEX IF NOT EXISTS idx_articles_publish_time ON articles(publish_time)) conn.commit() conn.close() def article_url_exists(conn, url): row conn.execute(SELECT id FROM articles WHERE url ?, (url,)).fetchone() return row is not None def insert_article(conn, article): conn.execute( INSERT INTO articles (title, url, summary, content, category, source, publish_time) VALUES (?, ?, ?, ?, ?, ?, ?) , ( article[title], article[url], article[summary], article[content], article.get(category, ), article[source], article[publish_time], )) conn.commit() def get_recent_articles(conn, limit200): rows conn.execute( SELECT id, title, url, content, category, source, publish_time FROM articles ORDER BY publish_time DESC LIMIT ? , (limit,)).fetchall() return rows这里的去重策略分两层第一层是 URL 精确去重防止同一篇文章反复入库第二层是基于标题和正文的相似度去重防止不同 URL 下的转载文章重复。两层判断都要有缺一个都会出问题。4.5 重复判定模块 deduplicator.py重复判定是资讯系统里技术含量最高的模块之一。最常用的方法是用 TF-IDF 把文本转成向量再计算余弦相似度。相似度超过阈值就认为是重复文章。# 文件路径news_aggregator/deduplicator.py from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity from config import SIMILARITY_THRESHOLD def compute_similarity(text1, text2): if not text1 or not text2: return 0.0 vectorizer TfidfVectorizer() vectors vectorizer.fit_transform([text1, text2]) similarity cosine_similarity(vectors[0:1], vectors[1:2]) return float(similarity[0][0]) def is_duplicate(new_article, existing_articles): new_text f{new_article[title]} {new_article[summary]} {new_article[content]} for old in existing_articles: old_text f{old[title]} {old[summary]} {old[content]} similarity compute_similarity(new_text, old_text) if similarity SIMILARITY_THRESHOLD: return True, similarity return False, 0.0这个方案的问题在性能。如果候选文章有几千篇每次计算就非常慢。优化手段包括只查最近 24 小时内的文章作为候选。先比较标题的字符重叠度不满足就不做完整计算。使用 SimHash 或 MinHash 做粗筛。对中小型项目先用“标题重叠 TF-IDF 相似度”两层过滤就够了。如果你追求的只是把同一条新闻的转载过滤掉这套方案的效果已经很稳定。4.6 分类模块 classifier.py分类的核心是基于关键词打分。每一类配置一组关键词标题和正文中出现关键词就加分最终取最高分类别。# 文件路径news_aggregator/classifier.py import jieba from config import CATEGORY_KEYWORDS def classify_article(article): text f{article[title]} {article[summary]} {article[content]} tokens set(jieba.cut(text)) scores {} for category, keywords in CATEGORY_KEYWORDS.items(): score 0 for kw in keywords: if kw in text or kw in tokens: score kw.count( ) 1 scores[category] score # 分类时考虑来源自带提示 hint article.get(category_hint, ) if hint: scores[hint] scores.get(hint, 0) 2 if not scores or max(scores.values()) 0: return 综合 return max(scores, keyscores.get)这里的判断逻辑比较粗糙但胜在直观且容易调整。生产环境如果要提升精度可以考虑使用已经训练好的文本分类模型。引入实体识别识别地点、人物、机构。结合人工规则修正明显的错分。从材料看“失窃毕加索名画之谜终告破解”这类新闻很容易被错误分到“经济”或“社会”类。如果想让系统准确识别为“文化”类只需要在文化类的关键词列表里加入“名画”“毕加索”“失窃”等词。4.7 推送模块 notifier.py推送是很多资讯系统的“最后一公里”。常见的推送渠道有企业微信机器人、钉钉机器人、邮件、Webhook。以 Webhook 为例# 文件路径news_aggregator/notifier.py import requests from config import PUSH_WEBHOOK_URL, PUSH_TOKEN def format_message(article): message ( f标题{article[title]}\n f分类{article.get(category, 综合)}\n f来源{article[source]}\n f时间{article[publish_time]}\n f链接{article[url]} ) return message def push_article(article): if not PUSH_WEBHOOK_URL: return payload { token: PUSH_TOKEN, content: format_message(article), } try: resp requests.post(PUSH_WEBHOOK_URL, jsonpayload, timeout10) resp.raise_for_status() except Exception as e: print(f[推送失败] {article[title]}: {e})推送失败时要注意重试。简单场景下可以用“失败后写入本地失败队列下次任务再补推”的方式避免因为推送服务瞬时故障丢失重要信息。5. 完整示例从 RSS 抓取到分类入库下面把前面所有模块串成一个完整流程。以“ABC晚间新闻”类摘要信源为例跑通一次增量采集。# 文件路径news_aggregator/main.py from fetcher import fetch_all_sources from cleaner import clean_html from storage import init_db, get_connection, article_url_exists, insert_article, get_recent_articles from deduplicator import is_duplicate from classifier import classify_article from notifier import push_article from config import SOURCES def run_once(): init_db() conn get_connection() print(开始抓取信源...) articles fetch_all_sources(SOURCES) print(f抓取到 {len(articles)} 篇文章) recent_articles get_recent_articles(conn, limit200) new_count 0 duplicate_count 0 for article in articles: # 第一步URL 去重 if article_url_exists(conn, article[url]): duplicate_count 1 continue # 第二步内容清洗 article[content] clean_html(article[content]) article[summary] clean_html(article[summary]) # 第三步相似度去重 is_dup, score is_duplicate(article, recent_articles) if is_dup: duplicate_count 1 print(f相似重复: {article[title]}相似度 {score:.2f}) continue # 第四步分类 article[category] classify_article(article) # 第五步入库 insert_article(conn, article) # 第六步推送 push_article(article) new_count 1 print(f新增文章: [{article[category]}] {article[title]}) print(f本次运行完成新增 {new_count} 篇过滤重复 {duplicate_count} 篇) conn.close() if __name__ __main__: run_once()这是典型的管道式流水线。每个步骤只处理一个关注点数据从上游流到下游。注意一个细节recent_articles在循环外只加载一次。这样做是为了避免候选列表越长越慢。但如果一次抓取的文章数量很大还是建议分批次处理。执行方式python main.py预期输出开始抓取信源... 抓取到 12 篇文章 新增文章: [天气] 美国东西海岸遭遇恶劣天气 新增文章: [交通] 达美航班紧急返航 新增文章: [文化] 失窃毕加索名画之谜终告破解 新增文章: [综合] 其他新闻摘要 ... 本次运行完成新增 8 篇过滤重复 2 篇如果运行失败优先检查三处第一网络是否能访问目标 RSS 源第二SQLite 数据库文件是否可写第三依赖库是否安装完整。6. 定时调度与增量抓取新闻系统必须自动化。使用 APScheduler 可以很容易地把上面的run_once()变成定时任务。# 文件路径news_aggregator/scheduler.py from apscheduler.schedulers.blocking import BlockingScheduler from config import FETCH_INTERVAL_MINUTES from main import run_once def main(): scheduler BlockingScheduler() scheduler.add_job( run_once, triggerinterval, minutesFETCH_INTERVAL_MINUTES, idnews_fetch_job, max_instances1, coalesceTrue, ) print(f定时任务已启动每 {FETCH_INTERVAL_MINUTES} 分钟执行一次) scheduler.start() if __name__ __main__: main()max_instances1的意思是同一个任务不允许并发执行。coalesceTrue的意思是如果任务积压多次只执行最后一次。这两项配置对抓取类任务非常重要否则上一次还没跑完下一次又开始了数据库连接和信源压力都会出问题。更进一步生产环境建议把调度器独立成一个服务和 Web 服务分开部署。这样抓取任务卡住不会影响对外接口。7. 改造为 JSON API 或数据库增量同步如果你的数据不是来自 RSS而是来自第三方 API 的 JSON流程基本一致只是把parse_rss换成parse_api。以某新闻 API 的典型响应为例{ code: 0, data: { news: [ { title: 美国东西海岸遭遇恶劣天气与野火, url: https://example.com/news/12345, summary: 多地发布预警, publish_time: 2026-08-09 19:00:00, category: 天气 } ] } }对应的抓取函数def parse_api(source): resp requests.get(source[url], timeoutFETCH_TIMEOUT) resp.raise_for_status() data resp.json() articles [] for item in data.get(data, {}).get(news, []): articles.append({ title: item.get(title, ).strip(), url: item.get(url, ).strip(), summary: item.get(summary, ).strip(), content: item.get(content, item.get(summary, )).strip(), publish_time: item.get(publish_time, ), source: source[name], }) return articlesAPI 接入时最需要注意的是分页和增量。很多 API 支持按时间查询要维护一个last_sync_time每次只拉取这个时间之后的数据减少重复计算。8. 常见问题与排查方法资讯聚合系统在真实运行中会遇到很多意料之外的问题。下面整理几个高频问题问题现象可能原因排查方式解决方案某个信源一直抓取失败对方服务器超时或封禁 IP手工 curl 查看返回状态码增加重试机制降低抓取频率或更换代理出口入库文章大量重复只做了 URL 去重没有做相似度去重检查重复文章是否 URL 不同启用 TF-IDF 或 SimHash 相似度去重分类结果混乱关键词配置太少或词太泛抽样检查错误分类的文章持续补充领域关键词引入语料训练模型数据库越来越慢没有按发布时间建索引查询全表扫描查看慢查询日志和执行计划给 publish_time、url 加索引定期清理旧数据抓取任务重叠执行定时任务没有限流查看任务日志中的执行时间设置 max_instances1使用分布式锁凌晨突发新闻延迟严重轮询间隔太长统计信源更新时段高峰期缩短轮询间隔或接入 Webhook 回调正文清洗后为空原页面正文由 JS 动态渲染查看原始 HTML 是否包含正文改用无头浏览器渲染或只保留摘要这里重点提醒一下数据库索引一定要在项目初期就加上。等到数据量大了再补索引迁移成本会高很多而且容易漏掉正在执行的写入任务。9. 生产环境最佳实践9.1 抓取频率要克制抓取不是越快越好。对方服务器有压力你的 IP 也容易被封。通用建议是常规信源 10 到 15 分钟抓一次突发新闻场景下单独处理而不是让所有信源都变成 1 分钟一次。9.2 数据要留全量展示用增量清洗后的数据要保留原始抓取内容字段方便回溯问题。不要因为“数据库里不想存脏数据”就把原始内容直接丢掉。生产环境建议加一个raw_data字段或独立表。9.3 失败补偿机制抓取失败、推送失败、分类失败都要有补偿机制。最基础的做法是失败任务写入task_log表定时检查并重试。不要指望一次执行成功线上系统永远有不稳定因素。9.4 告警监控系统不能“悄悄死掉”。建议监控以下指标每个信源最近一次成功抓取时间。每小时新增文章数。去重率和分类置信度。推送接口的失败率。指标变化异常时通过企业微信或邮件通知维护人员。9.5 安全与权限接入外部 API 时API Key 不能写在代码仓库里。建议使用环境变量或密钥管理服务。数据库访问账号遵循最小权限原则只授权业务需要的表。生产环境的数据操作必须先备份、再执行尤其是清理重复数据和批量修改操作。10. 从最小系统到完整平台的演进方向跑通最小系统之后后续可以沿着四个方向演进。第一是增强去重算法。当前使用 TF-IDF 余弦相似度对于数据量中等的场景够用数据量大之后可以换成 SimHash先做哈希指纹粗筛再做精确相似度计算性能会好很多。第二是引入实体识别。新闻内容里最关键的信息是“谁、在哪、发生了什么事”。如果能把地名、人名、机构名提取出来分类、搜索、个性化推荐都会有质的提升。用现成的 HanLP 或 spaCy 就能做。第三是搜索结果优化。资讯系统最终要支持用户检索。建议给标题、正文建全文索引PostgreSQL 的 tsvector 或 Elasticsearch 都可以。不要把 SQLite 的 LIKE 查询当全文搜索用数据量大之后性能会很难看。第四是个性化推送。根据用户订阅的分类和关键词在推送层做过滤。比如关注天气的用户只推天气分类关注文化新闻的用户只推文化分类。这个改动只在推送模块加一层过滤逻辑不需要动采集链路。从“ABC晚间新闻”这种多主题摘要型信源来思考这类数据最适合验证的其实就是分类和个性化推送。一条摘要里同时包含天气、交通、文化信息系统如果能自动拆分到不同栏目就已经超过了很多手动编辑为主的资讯台。最后提醒一句任何资讯聚合系统都只能解决“信息收集和整理”的问题不能解决“信息准确性和时效性”的问题。生产环境接入信源前一定要确认对方的授权和版权要求面向用户展示时要保留原文链接和发布时间。技术做得再漂亮内容合规和来源可靠才是底线。