基于Python FastAPI与NLP技术的智能RSS新闻聚合系统设计与实现

一、项目背景与目标

在当今信息爆炸的时代，用户往往需要从数十个甚至上百个RSS订阅源中获取关键信息。然而，多源订阅带来的信息冗余、重复内容泛滥等问题严重影响了信息获取效率。本项目旨在构建一个智能RSS热点聚合与订阅系统，通过自然语言处理（NLP）技术实现新闻的自动去重、聚类分析和加权排序，帮助用户从海量信息中快速筛选出高价值的热点内容。

系统采用Python FastAPI作为后端框架，结合Miniflux进行底层RSS抓取，通过SimHash算法实现新闻去重，并基于关键词权重和信源权重构建综合评分排序算法。

二、系统架构设计

2.1 整体架构

系统采用分层架构设计，主要包括以下组件：

┌─────────────────────────────────────────────────────────────────┐ │ 用户端 │ │ (Web浏览器 / 移动端H5页面) │ └──────────────────────────┬──────────────────────────────────────┘ │ HTTP/HTTPS ┌──────────────────────────▼──────────────────────────────────────┐ │ Nginx 负载均衡 │ └──────────────────────────┬──────────────────────────────────────┘ │ ┌──────────────────────────▼──────────────────────────────────────┐ │ FastAPI 应用服务器 │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │ │ │ 聚合算法服务 │ │ 排序算法服务 │ │ RSS同步服务 │ │ │ └─────────────┘ └─────────────┘ └─────────────────────────┘ │ └──────────────────────────┬──────────────────────────────────────┘ │ ┌──────────────────┼──────────────────┐ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ MySQL 8.0 │ │ Redis 6.0 │ │ Miniflux │ │ (业务数据) │ │ (缓存层) │ │ (RSS引擎) │ └─────────────┘ └─────────────┘ └─────────────┘ │ ▼ ┌─────────────┐ │ RSS订阅源 │ └─────────────┘

2.2 技术栈

三、核心算法实现

3.1 SimHash相似度去重算法

为解决多源RSS订阅中同一新闻重复出现的问题，系统采用SimHash算法计算新闻内容的特征指纹，通过相似度阈值判断是否属于同一聚类。

# app/services/aggregator.pyimportjiebafromcollectionsimportCounterclassSimHashCalculator:"""SimHash算法实现类"""def__init__(self,hash_bits:int=64):self.hash_bits=hash_bitsdef_tokenize(self,text:str)->list[str]:"""中文分词处理"""# 去除停用词和特殊字符words=jieba.cut(text)# 过滤停用词（实际实现中需要加载停用词表）return[wforwinwordsiflen(w)>1]def_hash_word(self,word:str)->int:"""将词语转换为哈希值"""# 使用Python内置hash函数（实际生产中建议使用MurmurHash）returnint.from_bytes(hash(word).to_bytes(8,byteorder='big'),byteorder='big')defcalculate_fingerprint(self,text:str)->int:""" 计算文本的SimHash指纹 算法步骤： 1. 分词并计算每个词的哈希值 2. 对哈希值按位累加 3. 根据位运算结果生成指纹 """words=self._tokenize(text)# 初始化特征向量（64位）hash_vectors=[0]*self.hash_bitsforwordinwords:word_hash=self._hash(word)# 按位累加foriinrange(self.hash_bits):ifword_hash>>i&1:hash_vectors[i]+=1else:hash_vectors[i]-=1# 生成指纹fingerprint=0foriinrange(self.hash_bits):ifhash_vectors[i]>0:fingerprint|=(1<<i)returnfingerprintdef_hash(self,word:str)->int:"""优化后的哈希方法"""returnint.from_bytes(hash(word).to_bytes(8,byteorder='little'),byteorder='big')defsimilarity(self,fp1:int,fp2:int)->float:""" 计算两个指纹的相似度（海明距离） 公式：similarity = (hash_bits - hamming_distance) / hash_bits """xor=fp1^fp2# 计算二进制中1的个数（海明距离）distance=bin(xor).count('1')return(self.hash_bits-distance)/self.hash_bits

聚类判断逻辑：

classNewsCluster:"""新闻聚类管理器"""def__init__(self,similarity_threshold:float=0.85,time_window_hours:int=24):self.similarity_threshold=similarity_threshold self.time_window=timedelta(hours=time_window_hours)self.simhash=SimHashCalculator()defshould_cluster(self,new_fingerprint:int,existing_clusters:list)->bool:""" 判断新新闻是否应归入现有聚类 参数： new_fingerprint: 新新闻的SimHash指纹 existing_clusters: 现有聚类列表 返回： True表示应归入某个现有聚类，False表示创建新聚类 """forclusterinexisting_clusters:similarity=self.simhash.similarity(new_fingerprint,cluster.signature)ifsimilarity>self.similarity_threshold:cluster.add_news(new_fingerprint)returnTruereturnFalse

3.2 加权排序算法

排序算法综合考虑关键词命中权重和信源权重，实现"关键词优先，信源热度叠加"的原则。

# app/services/ranker.pyfromtypingimportOptionalclassNewsRanker:"""新闻加权排序服务"""def__init__(self):# 关键词权重映射self.keyword_weights:dict[str,float]={}# 信源权重映射self.source_weights:dict[str,float]={}defset_keyword_weights(self,weights:dict[str,float])->None:"""设置关键词权重"""self.keyword_weights=weightsdefset_source_weights(self,weights:dict[str,float])->None:"""设置信源权重"""self.source_weights=weightsdefcalculate_keyword_score(self,title:str,summary:str)->float:""" 计算关键词匹配得分 说明： - 命中多个关键词时，取最高权重 - 未命中关键词时得分为0 """text=f"{title}{summary}"max_weight=0.0forkeyword,weightinself.keyword_weights.items():ifkeywordintext:max_weight=max(max_weight,weight)returnmax_weightdefcalculate_source_score(self,source_weights:list[float])->float:""" 计算信源热度得分 说明： - 累加所有来源的权重 - 被越多高权重媒体报道，得分越高 """returnsum(source_weights)defcalculate_total_score(self,title:str,summary:str,source_weights:list[float],debug:bool=False)->dict:""" 计算综合得分 评分公式：Score = (Wk × 100) + ΣWsi 其中： - Wk: 命中关键词的最高权重 (0-1) - ΣWsi: 所有源权重总和 返回包含详细得分信息的字典 """# 计算关键词得分keyword_score=self.calculate_keyword_score(title,summary)keyword_contribution=keyword_score*100# 计算信源热度得分source_score=self.calculate_source_score(source_weights)# 综合得分total_score=keyword_contribution+source_score result={"total_score":round(total_score,2),"keyword_score":keyword_score,"keyword_contribution":round(keyword_contribution,2),"source_score":round(source_score,2),"source_weights":source_weights,"is_high_priority":keyword_score>=0.8}ifdebug:result["ranking_formula"]={"expression":"(Wk × 100) + ΣWsi","Wk":keyword_score,"ΣWsi":source_score,"explanation":"关键词得分确保高优先级内容排在前面，信源热度叠加确保被权威媒体广泛报道的内容更突出"}returnresultdefrank_news(self,news_list:list[dict])->list[dict]:""" 对新闻列表进行排序 按综合得分降序排列 """scored_news=[]fornewsinnews_list:score_result=self.calculate_total_score(news["title"],news["summary"],news.get("source_weights",[]))news["score"]