AI大模型趣味實戰 第7集:多端適配 個人新聞頭條 基于大模型和RSS聚合打造個人新聞電臺(Flask WEB版) 1
摘要
在信息爆炸的時代,如何高效獲取和篩選感興趣的新聞內容成為一個現實問題。本文將帶領讀者通過Python和Flask框架,結合大模型的強大能力,構建一個個性化的新聞聚合平臺,不僅能夠自動收集整理各類RSS源的新聞,還能以語音播報的形式提供"新聞電臺"功能。我們將重點探討如何利用AI大模型優化新聞內容提取、自動生成標簽分類,以及如何通過語音合成技術實現新聞播報功能,打造一個真正實用的個人新聞助手。
項目代碼倉庫:https://github.com/wyg5208/rss_news_flask
系統運行截圖如下:
核心概念和知識點
1. RSS技術與信息聚合
RSS(Really Simple Syndication)是一種用于發布頻繁更新的網站內容的XML格式,它允許用戶訂閱網站的更新內容。我們的項目利用RSS技術,從多個新聞源自動獲取最新內容,無需手動訪問各個網站。
主要涉及知識點:
- RSS格式解析與內容提取
- Web爬蟲技術與內容清洗
- 增量式數據更新策略
2. Web應用開發與交互設計
采用Flask框架構建Web應用,實現用戶友好的界面和交互體驗。
主要涉及知識點:
- Flask應用結構設計
- 前后端交互與API設計
- 用戶認證與會話管理
- 響應式界面設計
3. 大模型應用
項目中大模型的應用主要體現在兩個方面:
- 新聞內容優化:使用大模型智能提取文章核心內容,去除廣告等干擾元素
- 自動標簽生成:分析文章內容,自動提取關鍵詞作為標簽
主要涉及知識點:
- 大模型API調用方法
- Prompt設計與優化
- 文本分析與關鍵信息提取
4. 語音合成技術
將文本轉換為語音,實現新聞播報功能。
主要涉及知識點:
- 文本到語音(TTS)技術
- 音頻文件處理與管理
- 瀏覽器語音API集成
5. 系統設計與優化
包括數據庫設計、任務調度系統、資源管理等方面。
主要涉及知識點:
- SQLite數據庫設計與優化
- 多線程任務處理
- 定時任務調度系統
- 日志系統設計與管理
實戰案例
接下來,我們將通過詳細的代碼示例和實現步驟,展示如何從零開始構建這個新聞聚合平臺。
1. 項目初始化與環境配置
首先,我們需要創建項目目錄結構并安裝必要的依賴包。
# 項目依賴
# requirements.txt
alembic==1.15.1
aniso8601==10.0.0
anyio==4.9.0
attrs==25.3.0
beautifulsoup4==4.13.3
blinker==1.9.0
bs4==0.0.2
certifi==2025.1.31
cffi==1.17.1
charset-normalizer==3.4.1
click==8.1.8
colorama==0.4.6
comtypes==1.4.10
feedparser==6.0.10
Flask==2.3.3
Flask-Login==0.6.3
Flask-Migrate==4.0.5
Flask-RESTful==0.3.10
Flask-SQLAlchemy==3.1.1
Flask-WTF==1.2.1
greenlet==3.1.1
gunicorn==21.2.0
h11==0.14.0
httpcore==1.0.7
httpx==0.25.2
idna==3.10
itsdangerous==2.2.0
Jinja2==3.1.2
lxml==4.9.3
Mako==1.3.9
MarkupSafe==3.0.2
ollama==0.1.5
outcome==1.3.0.post0
packaging==24.2
pycparser==2.22
pypiwin32==223
PySocks==1.7.1
python-dotenv==1.0.1
pyttsx3==2.98
pytz==2025.1
pywin32==310
requests==2.31.0
schedule==1.2.1
selenium==4.15.2
sgmllib3k==1.0.0
six==1.17.0
sniffio==1.3.1
sortedcontainers==2.4.0
soupsieve==2.6
SQLAlchemy==2.0.39
trio==0.29.0
trio-websocket==0.12.2
typing_extensions==4.12.2
urllib3==2.3.0
webdriver-manager==4.0.1
Werkzeug==2.3.7
wsproto==1.2.0
WTForms==3.2.1
使用如下命令安裝依賴:
pip install -r requirements.txt
2. Flask應用框架搭建
創建基礎的Flask應用結構:
# app.py (基礎結構)
from flask import Flask, render_template, request, jsonify, redirect, url_for, flash
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
import osapp = Flask(__name__)
app.config['SECRET_KEY'] = os.urandom(24)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///rss_news.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = Falsedb = SQLAlchemy(app)# 配置登錄管理器
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'# 數據庫模型定義
class User(UserMixin, db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(100), unique=True)password = db.Column(db.String(100))def __repr__(self):return f'<User {self.username}>'# 路由定義
@app.route('/')
def index():return render_template('index.html')# 應用入口
if __name__ == '__main__':with app.app_context():db.create_all()app.run(debug=True)
3. 數據庫模型設計
我們需要設計完整的數據庫模型來存儲新聞和相關信息:
# 數據庫模型定義
class News(db.Model):id = db.Column(db.Integer, primary_key=True)title = db.Column(db.String(500))link = db.Column(db.String(500), unique=True, index=True)description = db.Column(db.Text)content = db.Column(db.Text)source = db.Column(db.String(100))pub_date = db.Column(db.DateTime)add_date = db.Column(db.DateTime, default=datetime.datetime.now)class Tag(db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(100))news_id = db.Column(db.Integer, db.ForeignKey('news.id'))news = db.relationship('News', backref=db.backref('tags', lazy=True))class TagLibrary(db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(100), unique=True)category = db.Column(db.String(50))frequency = db.Column(db.Integer, default=0)class ScheduledTask(db.Model):id = db.Column(db.Integer, primary_key=True)task_id = db.Column(db.String(100), unique=True)task_type = db.Column(db.String(20)) # fetch or broadcastschedule_type = db.Column(db.String(20)) # daily, weekly, monthlyvalue = db.Column(db.Integer) # day number for weekly/monthlytime_value = db.Column(db.String(10)) # HH:MMextra_params = db.Column(db.String(100)) # JSON string for additional params
4. RSS內容抓取與處理
RSS內容抓取是整個系統的核心功能之一:
def fetch_rss_task(use_selenium=False, use_llm=True):logger.info("開始執行RSS抓取任務...")# 創建一個應用上下文對象app_ctx = app.app_context()# 推送上下文app_ctx.push()try:# 獲取RSS源with open('rss_list.txt', 'r', encoding='utf-8') as f:rss_urls = [line.strip() for line in f.readlines() if line.strip()]logger.info(f"讀取到{len(rss_urls)}個RSS源")if not rss_urls:logger.warning("RSS列表為空,沒有要抓取的源")returntotal_fetched = 0newly_added = 0# 初始化WebDriver(如果需要)driver = Noneif use_selenium:try:driver = WebDriverManager.get_instance().get_driver()except Exception as e:logger.error(f"初始化WebDriver時出錯: {e}")use_selenium = Falsetry:for url in rss_urls:try:logger.info(f"開始處理RSS源: {url}")# 解析RSS Feedfeed = feedparser.parse(url)if not feed.entries:logger.warning(f"{url} 沒有條目")continuesource = feed.feed.title if hasattr(feed.feed, 'title') else urlfor entry in feed.entries:title = entry.title if hasattr(entry, 'title') else "無標題"link = entry.link if hasattr(entry, 'link') else ""description = entry.description if hasattr(entry, 'description') else ""# 清理描述中的HTML標簽clean_description = ""if description:soup = BeautifulSoup(description, 'html.parser')clean_description = soup.get_text(separator=' ', strip=True)if not link:logger.warning("跳過無鏈接的條目")continue# 檢查鏈接是否已存在existing_news = News.query.filter_by(link=link).first()if existing_news:logger.warning(f"跳過已存在的新聞: {title}")total_fetched += 1continue# 獲取正文內容content = ""if use_selenium:try:content = extract_content_with_selenium(link, driver)except Exception as e:logger.error(f"使用Selenium提取內容時出錯: {e}")content = extract_content(link)else:content = extract_content(link)# 使用大模型優化內容(如果啟用)if use_llm and content:try:content = optimize_content_with_llm(content)except Exception as e:logger.error(f"使用大模型優化內容時出錯: {e}")# 創建新聞條目news = News(title=title,link=link,description=clean_description,content=content,source=source,pub_date=pub_date)db.session.add(news)db.session.commit()# 生成并保存標簽if content:generate_tags_for_news(news)total_fetched += 1newly_added += 1logger.info(f"成功添加新聞: {title}")except Exception as e:logger.error(f"處理RSS源 {url} 時出錯: {e}")continuefinally:# 確保資源被釋放if use_selenium:logger.info("抓取任務完成,資源將在應用上下文關閉時釋放")finally:# 彈出上下文 - 確保在所有情況下都釋放上下文app_ctx.pop()
5. 大模型內容優化
使用大模型進行內容優化,提取核心新聞內容:
def optimize_content_with_llm(content):"""使用大模型優化內容"""try:prompt = f"""
你是一個智能的內容提取助手。請從以下HTML內容中提取出真正的新聞文章內容,
移除所有廣告、導航、頁腳、側邊欄等無關內容。
保留原始的段落結構,返回整潔的HTML格式。
只返回正文內容,不要添加任何解釋。內容:
{content[:10000]} # 限制輸入長度
"""# 調用Ollama APIresponse = ollama.chat(model='glm4', messages=[{'role': 'user','content': prompt}])extracted_content = response['message']['content']# 確保返回的是HTML格式if not extracted_content.strip().startswith('<'):extracted_content = f"<p>{extracted_content}</p>"return extracted_contentexcept Exception as e:logger.error(f"使用大模型優化內容時出錯: {e}")return content # 出錯時返回原始內容
6. 自動標簽生成
使用大模型自動為新聞生成標簽:
def generate_tags_for_news(news):"""為新聞生成標簽"""try:# 使用大模型生成標簽prompt = f"""
分析以下新聞文章,提取5個關鍵詞作為標簽。
標簽應該是單個詞或短語,不超過10個字符,用逗號分隔。
只返回標簽列表,不要添加任何解釋。標題: {news.title}
描述: {news.description or ""}
內容: {news.content[:5000] if news.content else ""}
"""try:# 調用Ollama APIresponse = ollama.chat(model='glm4', messages=[{'role': 'user','content': prompt}])tags_text = response['message']['content']# 解析返回的標簽tags = [tag.strip() for tag in re.split(r'[,,、]', tags_text) if tag.strip()]# 過濾長度超過10個字符的標簽tags = [tag for tag in tags if len(tag) <= 10]# 最多保留5個標簽tags = tags[:5]except Exception as e:logger.error(f"使用大模型生成標簽時出錯: {e}")# 如果大模型失敗,嘗試使用簡單的關鍵詞提取words = re.findall(r'\b\w{3,15}\b', news.title + " " + (news.description or ""))word_count = {}for word in words:if word.lower() not in ['the', 'and', 'for', 'with', 'that', 'this']:word_count[word] = word_count.get(word, 0) + 1tags = [word for word, count in sorted(word_count.items(), key=lambda x: x[1], reverse=True) if len(word) <= 10][:5]# 保存標簽for tag_name in tags:# 檢查標簽庫是否有該標簽tag_in_library = TagLibrary.query.filter_by(name=tag_name).first()if not tag_in_library:# 創建新標簽庫條目tag_in_library = TagLibrary(name=tag_name, frequency=1)db.session.add(tag_in_library)else:# 更新使用頻率tag_in_library.frequency += 1# 創建新標簽關聯tag = Tag(name=tag_name, news_id=news.id)db.session.add(tag)db.session.commit()except Exception as e:logger.error(f"生成標簽時出錯: {e}")db.session.rollback()
7. 語音合成與新聞播報
實現新聞語音播報功能:
@app.route('/api/text_to_speech', methods=['POST'])
@login_required
def text_to_speech():"""將文本轉換為語音文件并返回URL"""try:# 獲取請求數據data = request.get_json()if not data or 'text' not in data:return jsonify({'status': 'error', 'message': '缺少文本參數'})text = data['text']if not text or len(text) == 0:return jsonify({'status': 'error', 'message': '文本內容為空'})# 限制文本長度,避免處理過長的文本if len(text) > 10000:text = text[:10000] + "..."# 確保存儲目錄存在audio_dir = os.path.join(app.static_folder, 'audio')if not os.path.exists(audio_dir):os.makedirs(audio_dir)# 生成唯一文件名filename = f"tts_{uuid.uuid4().hex}.mp3"filepath = os.path.join(audio_dir, filename)# 啟動后臺線程生成語音文件tts_thread = threading.Thread(target=generate_tts_file,args=(text, filepath))tts_thread.start()# 等待生成完成(最多30秒)tts_thread.join(timeout=30)# 檢查文件是否生成成功if os.path.exists(filepath) and os.path.getsize(filepath) > 0:# 返回文件URLaudio_url = url_for('static', filename=f'audio/{filename}')return jsonify({'status': 'success','audio_url': audio_url})else:return jsonify({'status': 'error','message': '語音生成失敗或超時'})except Exception as e:logger.error(f"文本轉語音出錯: {e}")return jsonify({'status': 'error','message': str(e)})def generate_tts_file(text, output_file):"""生成語音文件的后臺任務"""try:# 初始化語音引擎engine = pyttsx3.init()# 設置語音屬性engine.setProperty('rate', 160) # 語速engine.setProperty('volume', 1.0) # 音量# 選擇中文語音(如果可用)voices = engine.getProperty('voices')for voice in voices:if 'chinese' in voice.id.lower() or 'zh' in voice.id.lower():engine.setProperty('voice', voice.id)break# 保存為音頻文件engine.save_to_file(text, output_file)engine.runAndWait()logger.info(f"語音文件已生成: {output_file}")except Exception as e:logger.error(f"生成語音文件出錯: {e}")
8. 定時任務調度系統
設計定時任務系統,自動執行新聞抓取和播報:
def init_scheduler():"""初始化調度器任務"""with app.app_context():# 清空現有任務schedule.clear()# 加載數據庫中的任務tasks = ScheduledTask.query.all()for task in tasks:if task.task_type == 'fetch':add_fetch_task(task.task_id, task.schedule_type, task.value, task.time_value)elif task.task_type == 'broadcast':extra_params = json.loads(task.extra_params) if task.extra_params else {}count = extra_params.get('count', 5)add_broadcast_task(task.task_id, task.schedule_type, task.value, task.time_value, count)# 添加定期清理音頻文件的任務schedule.every(1).hours.do(cleanup_audio_files).tag('cleanup_audio')logger.info("已添加音頻文件清理任務,每小時執行一次")# 添加定期清理過期日志文件的任務schedule.every(12).hours.do(cleanup_log_files).tag('cleanup_logs')logger.info("已添加日志文件清理任務,每12小時執行一次")def add_broadcast_task(task_id, schedule_type, value, time_value, count=5):"""添加新聞播報任務到調度器"""def task_func():logger.info(f"執行新聞播報任務: {task_id}")# 創建應用上下文ctx = app.app_context()ctx.push()try:# 獲取最新的新聞news_list = db.session.query(News).order_by(News.add_date.desc()).limit(count).all()if news_list:# 初始化語音引擎try:engine = pyttsx3.init()# 設置語音參數engine.setProperty('rate', 150)engine.setProperty('volume', 0.9)# 播報開始提示engine.say("開始播報最新新聞")engine.runAndWait()# 逐條播報新聞for i, news in enumerate(news_list):# 播報標題engine.say(f"第{i+1}條新聞:{news.title}")engine.runAndWait()# 播報簡短描述if news.description and len(news.description) > 0:short_desc = news.description[:200] + "..." if len(news.description) > 200 else news.descriptionengine.say(short_desc)engine.runAndWait()# 短暫停頓,區分不同新聞time.sleep(1)# 播報結束提示engine.say("新聞播報結束")engine.runAndWait()except Exception as e:logger.error(f"語音引擎初始化或播報過程出錯: {e}")finally:# 釋放上下文ctx.pop()# 根據不同的調度類型添加任務if schedule_type == 'daily':schedule.every().day.at(time_value).do(task_func).tag(task_id)elif schedule_type == 'weekly':days = {1: schedule.every().monday,2: schedule.every().tuesday,3: schedule.every().wednesday,4: schedule.every().thursday,5: schedule.every().friday,6: schedule.every().saturday,7: schedule.every().sunday}days[value].at(time_value).do(task_func).tag(task_id)elif schedule_type == 'monthly':# 設置每月指定日期執行job = schedule.every().day.at(time_value).do(task_func).tag(task_id)# 自定義月度任務的執行條件def monthly_condition():return datetime.datetime.now().day == valuejob.do_run = lambda: monthly_condition() and task_func()
9. 日志系統設計
為應用添加完善的日志系統,便于監控和調試:
# 日志系統配置
LOG_DIR = 'logs'
if not os.path.exists(LOG_DIR):os.makedirs(LOG_DIR)# 內存緩沖區,用于在UI中顯示最新日志
log_buffer = deque(maxlen=1000)# 創建自定義的日志記錄器
class MemoryHandler(logging.Handler):"""將日志記錄到內存緩沖區,用于Web界面顯示"""def emit(self, record):log_entry = self.format(record)log_buffer.append({'time': datetime.datetime.fromtimestamp(record.created).strftime('%Y-%m-%d %H:%M:%S'),'level': record.levelname,'message': record.getMessage(),'formatted': log_entry})# 配置日志記錄器
logger = logging.getLogger('rss_app')
logger.setLevel(logging.INFO)# 控制臺處理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(console_format)
logger.addHandler(console_handler)# 內存處理器,用于UI顯示
memory_handler = MemoryHandler()
memory_handler.setLevel(logging.INFO)
memory_handler.setFormatter(console_format)
logger.addHandler(memory_handler)# 小時文件處理器,每小時自動創建一個新文件
hourly_handler = TimedRotatingFileHandler(filename=os.path.join(LOG_DIR, 'rss_app.log'),when='H',interval=1,backupCount=72, # 保留3天的日志encoding='utf-8'
)
# 設置日志文件后綴格式為 年-月-日_小時
hourly_handler.suffix = "%Y-%m-%d_%H"
hourly_handler.setLevel(logging.INFO)
hourly_handler.setFormatter(console_format)
logger.addHandler(hourly_handler)@app.route('/system_logs')
@login_required
def system_logs():"""顯示系統日志頁面"""logger.info('訪問系統日志頁面')# 獲取日志文件列表log_files = []try:# 獲取所有日志文件并按修改時間排序log_pattern = os.path.join(LOG_DIR, 'rss_app.log*')all_log_files = glob.glob(log_pattern)all_log_files.sort(key=os.path.getmtime, reverse=True)for file_path in all_log_files:file_name = os.path.basename(file_path)# 獲取文件大小和修改時間file_stats = os.stat(file_path)file_size = file_stats.st_size / 1024 # KBfile_time = datetime.datetime.fromtimestamp(file_stats.st_mtime).strftime('%Y-%m-%d %H:%M:%S')# 添加文件信息if file_name == 'rss_app.log':display_name = f"當前日志 ({file_size:.1f} KB) - {file_time}"log_files.append({'name': display_name,'path': file_path})else:# 格式化時間戳timestamp = file_name.replace('rss_app.log.', '')try:# 嘗試解析時間戳parsed_time = datetime.datetime.strptime(timestamp, '%Y-%m-%d_%H')display_name = f"{parsed_time.strftime('%Y-%m-%d %H:00')} ({file_size:.1f} KB)"except:display_name = f"{file_name} ({file_size:.1f} KB) - {file_time}"log_files.append({'name': display_name,'path': file_path})except Exception as e:logger.error(f"獲取日志文件列表出錯: {str(e)}")flash(f"獲取日志文件列表出錯: {str(e)}", 'danger')# 統計信息stats = {'total': len(log_buffer),'error': sum(1 for log in log_buffer if log['level'] == 'ERROR'),'warning': sum(1 for log in log_buffer if log['level'] == 'WARNING'),'files': len(log_files)}return render_template('system_logs.html', log_files=log_files, stats=stats)