1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
| import feedparser
import requests
from datetime import datetime
from typing import List, Dict
import sqlite3
class RSSAggregator:
def __init__(self, db_path='news.db'):
self.db_path = db_path
self.init_database()
def init_database(self):
"""初始化数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS articles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT,
link TEXT UNIQUE,
description TEXT,
published DATETIME,
source TEXT,
category TEXT,
read BOOLEAN DEFAULT 0
)
''')
conn.commit()
conn.close()
def fetch_feed(self, feed_url: str, source_name: str):
"""获取并解析 RSS/Atom feed"""
try:
# 使用 feedparser 自动检测格式
feed = feedparser.parse(feed_url)
articles = []
for entry in feed.entries:
article = {
'title': entry.get('title', ''),
'link': entry.get('link', ''),
'description': entry.get('description') or entry.get('summary', ''),
'published': self._parse_date(entry.get('published', '')),
'source': source_name,
'category': self._extract_category(entry)
}
articles.append(article)
return articles
except Exception as e:
print(f"Error fetching {feed_url}: {e}")
return []
def _parse_date(self, date_str):
"""解析日期"""
if not date_str:
return datetime.now()
try:
# feedparser 已经解析了日期
from email.utils import parsedate_to_datetime
return parsedate_to_datetime(date_str)
except:
return datetime.now()
def _extract_category(self, entry):
"""提取分类"""
# 从 tags 中提取
if hasattr(entry, 'tags'):
return ', '.join([tag.term for tag in entry.tags[:3]])
# 从 category 中提取
if hasattr(entry, 'category'):
return entry.category
return 'General'
def save_articles(self, articles: List[Dict]):
"""保存文章到数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
for article in articles:
try:
cursor.execute('''
INSERT OR IGNORE INTO articles
(title, link, description, published, source, category)
VALUES (?, ?, ?, ?, ?, ?)
''', (
article['title'],
article['link'],
article['description'],
article['published'],
article['source'],
article['category']
))
except Exception as e:
print(f"Error saving article: {e}")
conn.commit()
conn.close()
def aggregate_multiple_feeds(self, feeds: Dict[str, str]):
"""
聚合多个 feeds
feeds: {'Source Name': 'feed_url', ...}
"""
all_articles = []
for source_name, feed_url in feeds.items():
print(f"Fetching {source_name}...")
articles = self.fetch_feed(feed_url, source_name)
all_articles.extend(articles)
self.save_articles(all_articles)
print(f"Total articles fetched: {len(all_articles)}")
return all_articles
def get_latest_articles(self, limit=20, category=None):
"""获取最新文章"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
if category:
cursor.execute('''
SELECT * FROM articles
WHERE category LIKE ?
ORDER BY published DESC
LIMIT ?
''', (f'%{category}%', limit))
else:
cursor.execute('''
SELECT * FROM articles
ORDER BY published DESC
LIMIT ?
''', (limit,))
articles = cursor.fetchall()
conn.close()
return articles
# 使用示例
aggregator = RSSAggregator()
# 定义 feeds
feeds = {
'Hacker News': 'https://news.ycombinator.com/rss',
'TechCrunch': 'https://techcrunch.com/feed/',
'Ars Technica': 'https://feeds.arstechnica.com/arstechnica/index',
'The Verge': 'https://www.theverge.com/rss/index.xml',
'Reddit Python': 'https://www.reddit.com/r/python/.rss'
}
# 聚合所有 feeds
aggregator.aggregate_multiple_feeds(feeds)
# 获取最新 20 篇文章
latest = aggregator.get_latest_articles(limit=20)
for article in latest:
print(f"[{article[5]}] {article[1]}") # [source] title
print(f" {article[2]}") # link
print()
|
支付宝
微信