AI 驱动的 RSS 智能摘要阅读器：每天自动消化 50+ 篇文章

pip install feedparser requests

ollama pull qwen2.5:7b

#!/usr/bin/env python3
"""AI RSS 智能摘要阅读器"""

import json
import os
import time
import hashlib
from datetime import datetime
from pathlib import Path

import feedparser
import requests

# ============ 配置区 ============

# RSS 源列表（替换为你关注的源）
RSS_FEEDS = [
    {"name": "Hacker News", "url": "https://hnrss.org/frontpage"},
    {"name": "InfoQ 中文站", "url": "https://www.infoq.cn/feed"},
    {"name": "阮一峰周刊", "url": "https://www.ruanyifeng.com/blog/atom.xml"},
    {"name": "V2EX", "url": "https://www.v2ex.com/index.xml"},
]

# 摘要数量（每个源取最新 N 条）
ARTICLES_PER_FEED = 3

# 摘要提示词模板
SUMMARY_PROMPT = """你是一个技术资讯编辑。请对以下文章进行摘要。

要求：
1. 用一句话概括核心内容
2. 列出 2-3 个关键要点
3. 标注难度等级（入门/进阶/高阶）

文章标题：{title}
作者：{author}
发布时间：{pub_date}
正文摘要：{content}

只输出 JSON 格式，不要其他文字：
{{"summary": "...", "key_points": ["...", "..."], "difficulty": "..."}}
"""

# 钉钉 Webhook（留空则只打印到终端）
DINGTALK_WEBHOOK = os.getenv("DINGTALK_WEBHOOK", "")

# 缓存文件路径
CACHE_FILE = Path.home() / ".ai_rss_cache.json"


def load_cache() -> dict:
    """加载已处理的文章 ID 缓存"""
    if CACHE_FILE.exists():
        return json.loads(CACHE_FILE.read_text())
    return {}


def save_cache(cache: dict):
    """保存缓存"""
    # 只保留最近 1000 条记录，防止无限增长
    items = list(cache.items())[-1000:]
    CACHE_FILE.write_text(json.dumps(dict(items)))


def fetch_articles(feed: dict) -> list:
    """从 RSS 源获取文章列表"""
    entries = []
    try:
        d = feedparser.parse(feed["url"])
        for entry in d.entries[:ARTICLES_PER_FEED]:
            entries.append({
                "feed_name": feed["name"],
                "title": entry.get("title", "无标题"),
                "link": entry.get("link", ""),
                "author": entry.get("author", "未知"),
                "pub_date": entry.get("published", "未知"),
                "content": (entry.get("summary", "") or "")[:500],
                "id": hashlib.md5(entry.get("link", "").encode()).hexdigest(),
            })
    except Exception as e:
        print(f"[错误] 获取 {feed['name']} 失败: {e}")
    return entries


def summarize_with_ollama(article: dict) -> dict:
    """调用 Ollama 生成摘要"""
    prompt = SUMMARY_PROMPT.format(
        title=article["title"],
        author=article["author"],
        pub_date=article["pub_date"],
        content=article["content"],
    )
    
    try:
        resp = requests.post(
            "http://localhost:11434/api/generate",
            json={
                "model": "qwen2.5:7b",
                "prompt": prompt,
                "stream": False,
                "options": {
                    "temperature": 0.3,
                    "num_predict": 300,
                },
            },
            timeout=30,
        )
        result = resp.json()
        raw_output = result.get("response", "")
        
        # 尝试从返回中提取 JSON
        start = raw_output.find("{")
        end = raw_output.rfind("}") + 1
        if start >= 0 and end > start:
            return json.loads(raw_output[start:end])
        
        # 降级：直接返回文本摘要
        return {
            "summary": raw_output[:200],
            "key_points": [],
            "difficulty": "未识别",
        }
    except Exception as e:
        print(f"[警告] Ollama 请求失败: {e}")
        return {
            "summary": article["title"],
            "key_points": [article["content"][:100]],
            "difficulty": "未知",
        }


def send_dingtalk(notifications: list):
    """发送钉钉通知"""
    if not DINGTALK_WEBHOOK:
        return
    
    text = "📰 AI RSS 每日精选\n\n"
    for i, item in enumerate(notifications, 1):
        text += f"{i}. **{item['feed_name']}**: {item['title']}\n"
        text += f"   {item['summary']}\n\n"
    
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": "AI RSS 每日精选",
            "text": text,
        },
    }
    
    requests.post(DINGTALK_WEBHOOK, json=payload, timeout=10)


def main():
    cache = load_cache()
    notifications = []
    new_count = 0
    
    for feed in RSS_FEEDS:
        articles = fetch_articles(feed)
        for article in articles:
            # 跳过已处理的文章
            if article["id"] in cache:
                continue
            
            cache[article["id"]] = datetime.now().isoformat()
            
            # 生成摘要
            summary = summarize_with_ollama(article)
            
            notification = {
                "feed_name": article["feed_name"],
                "title": article["title"],
                "link": article["link"],
                "summary": summary["summary"],
                "key_points": summary.get("key_points", []),
                "difficulty": summary.get("difficulty", "未知"),
            }
            notifications.append(notification)
            new_count += 1
    
    save_cache(cache)
    
    if notifications:
        print(f"\n✅ 共发现 {new_count} 条新内容\n")
        for item in notifications:
            print(f"[{item['feed_name']}] {item['title']}")
            print(f"  摘要: {item['summary']}")
            print(f"  难度: {item['difficulty']}")
            print()
        
        send_dingtalk(notifications)
        print("📤 钉钉通知已发送")
    else:
        print("📭 没有新内容")


if __name__ == "__main__":
    main()

crontab -e

python3 ai_rss_summarizer.py