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。相比 RSS，API 的字段更丰富，通常包含分类、标签、图片、作者。

API 的优点：

• 数据结构完整，不需要额外解析。
• 权限控制方便，可以按账号管理配额。
• 更新频率和粒度可控。

API 的缺点：

• 需要申请 key，有配额限制。
• 不同服务商的响应结构完全不同。
• 收费或者免费额度限制。

2.3 HTML 网页解析

当一个信源既没有 RSS，也没有 API，只能从 HTML 页面里抽取正文。这种方式最灵活，也最脆弱。

HTML 解析的常见手段：

• XPath 定位正文节点。
• 基于正文文本密度算法自动抽取。
• 用 Readability 类库提取正文。

网页解析是最后手段。对方前端结构一改，你的解析器就失效，必须配合监控告警。

2.4 三种方式的选择建议

从材料看，多数大众新闻源都保留了 RSS 输出。对于个人项目和中小型系统，建议第一版全部走 RSS，把 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"], timeout=FETCH_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 标签、跟踪链接、图片地址混排，甚至是整段的页面框架。

清洗步骤：

1. 去掉 HTML 标签。
2. 去掉 JS/CSS 代码片段。
3. 去掉常见广告词所在的段落。
4. 合并空白字符。
5. 截断过长的正文。

# 文件路径：news_aggregator/cleaner.py import re BAD_PATTERNS = [ r"var\s+\w+\s*=", r"function\s*\w*\s*\(", r"http[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, flags=re.S | re.I) # 去掉普通 HTML 标签 text = re.sub(r"<[^>]+>", "", text) # 去掉 URL text = re.sub(r"http[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>美国东西海岸遭遇恶劣天气，多地发布预警。</p><script>track('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, limit=200): 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, key=scores.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, json=payload, timeout=10) 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, limit=200) 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, trigger="interval", minutes=FETCH_INTERVAL_MINUTES, id="news_fetch_job", max_instances=1, coalesce=True, ) print(f"定时任务已启动，每 {FETCH_INTERVAL_MINUTES} 分钟执行一次") scheduler.start() if __name__ == "__main__": main()

max_instances=1的意思是同一个任务不允许并发执行。coalesce=True的意思是如果任务积压多次，只执行最后一次。这两项配置对抓取类任务非常重要，否则上一次还没跑完，下一次又开始了，数据库连接和信源压力都会出问题。

更进一步，生产环境建议把调度器独立成一个服务，和 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"], timeout=FETCH_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 articles

API 接入时最需要注意的是分页和增量。很多 API 支持按时间查询，要维护一个last_sync_time，每次只拉取这个时间之后的数据，减少重复计算。

8. 常见问题与排查方法

资讯聚合系统在真实运行中会遇到很多意料之外的问题。下面整理几个高频问题：

这里重点提醒一下：数据库索引一定要在项目初期就加上。等到数据量大了再补索引，迁移成本会高很多，而且容易漏掉正在执行的写入任务。

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晚间新闻”这种多主题摘要型信源来思考，这类数据最适合验证的其实就是分类和个性化推送。一条摘要里同时包含天气、交通、文化信息，系统如果能自动拆分到不同栏目，就已经超过了很多手动编辑为主的资讯台。

最后提醒一句：任何资讯聚合系统都只能解决“信息收集和整理”的问题，不能解决“信息准确性和时效性”的问题。生产环境接入信源前，一定要确认对方的授权和版权要求；面向用户展示时，要保留原文链接和发布时间。技术做得再漂亮，内容合规和来源可靠才是底线。