基于Python与边缘计算构建个性化新闻聚合系统
·
1. 项目概述：边缘新闻聚合的诞生
最近在折腾一个挺有意思的小项目，我把它叫做“News — At The Edge”。这个名字听起来可能有点抽象，但它的核心想法其实很直接：我们每天都被海量的新闻信息淹没，但真正有价值、与我们切身相关的，可能只是其中很小一部分。传统的新闻App要么推送过于庞杂，要么算法推荐形成“信息茧房”。这个项目的初衷，就是想做一个完全运行在“边缘”的、高度定制化的新闻筛选与聚合器。
这里的“边缘”，并不是一个地理概念，而是技术架构上的“边缘计算”思想。我不希望依赖一个庞大的中心服务器去抓取、分析和推送新闻，那样延迟高、隐私存疑，而且模式僵化。我希望这个工具能更“轻”、更“近”、更“智能”地运行在我自己的设备或家庭网络环境中，只为我一个人服务。它就像一个驻扎在我网络边缘的私人新闻管家，按照我设定的规则，从指定的、可信的新闻源（主要是各大媒体的RSS）抓取内容，然后根据我关心的关键词、主题进行实时过滤、去重和优先级排序，最后生成一份简洁的、属于我个人的“边缘新闻简报”。
这个项目特别适合那些对信息质量有要求，又希望保持技术掌控感的开发者、科技爱好者或深度阅读者。它不追求大而全，而是追求精准和效率。在5月12日这个版本里，我基本跑通了核心流程，实现了一个可用的最小可行产品。接下来，我就把这套从设计思路到踩坑实录的完整过程分享出来。
2. 核心架构与设计思路拆解
2.1 为什么选择“边缘”架构？
在项目启动前，我首先评估了三种常见的方案：一是使用现成的新闻API服务，二是自建中心服务器进行爬取和分发，三是采用边缘计算模式。
现成的API（如News API、GNews等）虽然方便，但存在调用次数限制、内容可能被二次处理、以及无法定制特定小众信源的问题。自建中心服务器则意味着需要维护一台24小时运行的VPS，处理爬虫管理、反爬对抗、数据存储和多个终端的数据同步，运维成本和复杂度陡然上升。
而边缘架构的优势在于：
- 隐私与数据自主 ：所有抓取、过滤、分析的动作都发生在你控制的设备上（比如家里的NAS、一台始终开机的旧电脑，甚至是一个树莓派），原始新闻数据不过任何第三方的手，阅读历史、偏好设置完全本地化。
- 成本与灵活性 ：省去了中心服务器的费用。规则可以极端个性化，你可以轻松添加任何一个提供RSS的小众博客，而无需担心服务器爬虫策略的通用性问题。
- 实时性与低延迟 ：边缘设备直接向新闻源发起请求，减少了中间转发环节。结合定时任务，可以实现近实时的新闻获取。
- 学习与可控性 ：整个数据流水线透明可见，从网络请求到文本处理，每个环节都可以根据你的需求调整和优化，是一个绝佳的学习和实践项目。
因此，我决定采用边缘架构。核心模型是： “一个调度中心 + 多个轻量级工作节点” 。调度中心（Cron Job或系统定时任务）负责按计划触发；工作节点（Python脚本）则执行具体的抓取、解析、过滤和生成任务。
2.2 技术栈选型与考量
技术选型上，我遵循“轻量、高效、易维护”的原则。
- 核心语言：Python 3.8+ 。选择Python是因为其在数据处理、网络爬虫和快速原型开发上无与伦比的生态优势。
requests、feedparser、BeautifulSoup4这些库能极大简化开发。 - 依赖管理：Pipenv 。它比单纯的
pip能更好地管理项目依赖和虚拟环境，特别是当项目需要部署到不同边缘设备时，能确保环境一致性。 - 数据解析：Feedparser + BeautifulSoup4 。绝大多数新闻网站和博客都提供RSS或Atom订阅源，
feedparser是解析它们的利器，稳定且兼容性好。对于少数需要从HTML中提取额外信息的源，用BeautifulSoup4补足。 - 文本处理与过滤：Jieba (中文) / NLTK (英文) + 正则表达式 。关键词过滤是核心。对于中文新闻，我选用
jieba进行分词，结合自定义词典加入专业词汇，以提高关键词匹配的准确性。初步过滤用正则表达式处理标题和摘要，更复杂的语义分析（在这个MVP版本中暂未深入）可以留待后续。 - 数据存储：SQLite 。边缘场景下，SQLite是天然的选择。它无需单独的服务进程，一个文件就是一个数据库，非常适合存储订阅源列表、抓取记录、新闻条目和用户过滤规则。轻量且足够可靠。
- 任务调度：Systemd Timer (Linux) 或 Crontab 。为了保持简单，我没有引入Celery这类重型任务队列。在Linux边缘设备上，
systemd timer是比crontab更现代和强大的选择，它能更好地管理日志、处理服务依赖和自动重启。对于macOS或Windows，也有相应的定时任务方案。 - 输出与通知：Markdown文件 + 邮件 / Webhook 。将最终生成的新闻简报保存为Markdown文件，便于阅读和归档。同时，可以通过
smtplib发送邮件到Kindle或邮箱，或者调用requests发送Webhook到钉钉、Slack等协作工具，实现推送。
这个技术栈组合，在保证功能完整的前提下，最大限度地降低了资源消耗和架构复杂度，非常适合在资源有限的边缘设备上长期稳定运行。
3. 核心模块实现细节解析
3.1 新闻源管理与RSS抓取器
新闻源的配置是整个系统的输入源头，我设计了一个 JSON 格式的配置文件 sources.json ，因为它比 YAML 更易于Python原生解析，也比 INI 格式能表达更复杂的结构。
[
{
"name": "Solidot",
"url": "https://www.solidot.org/index.rss",
"language": "zh",
"category": "tech",
"enabled": true,
"priority": 1
},
{
"name": "Reuters Technology",
"url": "http://feeds.reuters.com/reuters/technologyNews",
"language": "en",
"category": "tech",
"enabled": true,
"priority": 2
},
{
"name": "某个人博客",
"url": "https://example.com/feed",
"language": "zh",
"category": "blog",
"enabled": true,
"priority": 3,
"requires_js": false,
"custom_parser": null
}
]
每个源除了基本的名称和URL，还定义了语言、类别、启用状态和优先级。优先级用于在最终简报中排序。 requires_js 和 custom_parser 是为未来扩展预留的，用于处理那些动态加载或结构特殊的网站。
抓取器的核心函数如下：
import feedparser
import requests
from datetime import datetime, timedelta
import sqlite3
import logging
logger = logging.getLogger(__name__)
def fetch_feed(source_config):
"""抓取单个RSS源"""
name = source_config['name']
url = source_config['url']
logger.info(f"开始抓取源: {name}")
try:
# 设置请求头，模拟浏览器，避免被简单的反爬拦截
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
# 对于简单情况，feedparser可以直接处理URL
feed = feedparser.parse(url)
if feed.bozo: # bozo标志位表示解析可能出错
logger.warning(f"源 {name} RSS解析异常，尝试使用requests获取内容后解析。异常信息: {feed.bozo_exception}")
# 降级方案：使用requests获取原始内容
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
feed = feedparser.parse(resp.content)
entries = []
for entry in feed.entries:
# 标准化条目信息，处理可能缺失的字段
pub_time = entry.get('published_parsed') or entry.get('updated_parsed')
if pub_time:
pub_time = datetime(*pub_time[:6]) # 转换为datetime对象
else:
pub_time = datetime.utcnow() # 如果没有时间，使用当前时间
# 只抓取最近24小时的内容，避免历史数据堆积
if datetime.utcnow() - pub_time > timedelta(hours=24):
continue
standardized_entry = {
'title': entry.title,
'link': entry.link,
'summary': entry.get('summary', ''),
'published': pub_time,
'source': name,
'category': source_config.get('category', 'general')
}
entries.append(standardized_entry)
logger.info(f"源 {name} 抓取到 {len(entries)} 条新条目")
return entries
except requests.exceptions.RequestException as e:
logger.error(f"抓取源 {name} 时网络错误: {e}")
return []
except Exception as e:
logger.error(f"抓取源 {name} 时发生未知错误: {e}")
return []
注意 ：
feedparser的bozo标志位非常有用，它能捕获到XML格式不规范等解析错误。当bozo为True时，通过bozo_exception可以查看具体错误。采用requests获取原始内容再解析是一个有效的降级策略。另外，务必为网络请求设置超时（如30秒），避免因某个源响应慢而阻塞整个抓取流程。
3.2 基于关键词与规则的内容过滤引擎
抓取到的新闻条目是原始的，我们需要一个过滤引擎来筛选出真正感兴趣的内容。我设计了一个两层过滤机制： 规则过滤 和 关键词评分 。
首先，在 rules.json 中定义规则：
{
"must_have_keywords": ["人工智能", "机器学习", "LLM", "开源"],
"block_keywords": ["八卦", "娱乐", "彩票"],
"category_weights": {
"tech": 2.0,
"science": 1.5,
"finance": 1.2,
"general": 1.0,
"blog": 0.8
},
"source_priorities": {
"Solidot": 3,
"Reuters Technology": 2
}
}
过滤引擎的工作流程如下：
- 硬性规则过滤（Must/Block） ：如果一条新闻的标题或摘要中包含
block_keywords中的任何一个词，则直接丢弃。这是一个“一票否决”机制，用于过滤完全不感兴趣的内容。 - 关键词匹配与评分 ：对通过硬性过滤的新闻，计算其与
must_have_keywords的匹配度。这里不是简单的布尔判断，而是引入评分制。- 基础分：新闻条目所在类别的权重（
category_weights）。 - 加分项：标题中每匹配一个关键词，加1分；摘要中每匹配一个关键词，加0.5分。同时，匹配关键词的权重可以不同，例如“人工智能”比“开源”权重更高，这可以通过在关键词列表中用元组
(“人工智能”, 1.5)来表示。 - 源优先级加成：乘以新闻来源的优先级系数（
source_priorities）。
- 基础分：新闻条目所在类别的权重（
- 时间衰减因子 ：新闻的价值随时间推移而降低。可以为发布时间引入一个衰减因子，例如，发布后每小时得分衰减1%，鼓励系统优先呈现更新鲜的内容。
import jieba
from datetime import datetime
class ContentFilter:
def __init__(self, rules_path):
self.load_rules(rules_path)
# 初始化jieba，加载用户词典（如果有）
jieba.initialize()
# 可以添加用户词典 jieba.load_userdict('my_dict.txt')
def calculate_score(self, entry):
"""计算新闻条目的得分"""
score = self.category_weights.get(entry['category'], 1.0)
# 关键词匹配加分
title_words = set(jieba.lcut_for_search(entry['title']))
summary_words = set(jieba.lcut_for_search(entry['summary']))
for keyword, weight in self.keywords_with_weight:
if keyword in title_words:
score += weight * 1.0 # 标题匹配权重高
elif keyword in summary_words:
score += weight * 0.5 # 摘要匹配权重低
# 源优先级加成
score *= self.source_priorities.get(entry['source'], 1.0)
# 时间衰减（示例：每过1小时，得分乘以0.99）
hours_passed = (datetime.utcnow() - entry['published']).total_seconds() / 3600
time_decay = 0.99 ** hours_passed
score *= time_decay
return score
def filter_and_rank(self, entries, top_n=20):
"""过滤并排名新闻条目"""
filtered = []
for entry in entries:
# 1. 硬性屏蔽词检查
if self.has_blocked_keywords(entry):
continue
# 2. 计算得分
entry['score'] = self.calculate_score(entry)
filtered.append(entry)
# 按得分降序排序，返回前top_n条
filtered.sort(key=lambda x: x['score'], reverse=True)
return filtered[:top_n]
实操心得 ：关键词匹配的准确性严重依赖分词质量。对于中文，
jieba.lcut_for_search比lcut更适合搜索场景，它能将长词拆开。务必根据你的领域构建一个自定义词典，例如加入“Stable Diffusion”、“Transformer”等专有名词，避免被错误切分。评分规则是系统的“大脑”，需要反复调整权重参数来逼近你个人的阅读偏好，这是一个持续迭代的过程。
3.3 数据持久化与去重机制
为了避免每次抓取都推送相同的新闻，一个健壮的去重机制是必须的。我选择使用SQLite数据库，并设计了两张核心表。
-- sources 表：记录订阅源信息
CREATE TABLE IF NOT EXISTS sources (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
url TEXT NOT NULL,
last_fetched TIMESTAMP DEFAULT NULL
);
-- articles 表：记录抓取到的文章
CREATE TABLE IF NOT EXISTS articles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
link TEXT UNIQUE NOT NULL, -- 链接作为唯一标识，用于去重
summary TEXT,
published TIMESTAMP NOT NULL,
source_id INTEGER NOT NULL,
fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_read BOOLEAN DEFAULT 0,
score REAL DEFAULT 0.0,
FOREIGN KEY (source_id) REFERENCES sources (id)
);
-- 为link和published字段创建索引以提高查询速度
CREATE INDEX IF NOT EXISTS idx_articles_link ON articles (link);
CREATE INDEX IF NOT EXISTS idx_articles_published ON articles (published);
去重逻辑在存储抓取结果时实现：
def save_articles_if_new(articles, db_path='news_edge.db'):
"""保存文章列表，自动去重"""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
new_count = 0
for article in articles:
# 以link作为唯一键进行去重
cursor.execute("SELECT id FROM articles WHERE link = ?", (article['link'],))
if cursor.fetchone() is None:
# 获取source_id，如果不存在则插入新源
cursor.execute("INSERT OR IGNORE INTO sources (name, url) VALUES (?, ?)",
(article['source'], article.get('source_url', '')))
cursor.execute("SELECT id FROM sources WHERE name = ?", (article['source'],))
source_id = cursor.fetchone()[0]
# 插入新文章
cursor.execute('''
INSERT INTO articles (title, link, summary, published, source_id, score)
VALUES (?, ?, ?, ?, ?, ?)
''', (article['title'], article['link'], article['summary'],
article['published'].isoformat(), source_id, article.get('score', 0)))
new_count += 1
conn.commit()
# 更新源的最近抓取时间
for source_name in set([a['source'] for a in articles]):
cursor.execute("UPDATE sources SET last_fetched = CURRENT_TIMESTAMP WHERE name = ?", (source_name,))
conn.commit()
conn.close()
logger.info(f"保存了 {new_count} 篇新文章，去重了 {len(articles) - new_count} 篇重复文章。")
这个机制确保了即使抓取脚本被多次执行，同一篇新闻也只会被记录和推送一次。 is_read 字段可以用于未来实现“已读”状态同步。
4. 系统集成与自动化部署
4.1 使用Systemd Timer实现可靠调度
在Linux边缘设备上， crontab 虽然简单，但在服务管理、日志集成和故障重启方面不如 systemd 。我选择使用 systemd timer 来调度我的新闻抓取脚本。
首先，创建一个系统服务单元文件 /etc/systemd/system/news-edge-fetcher.service ：
[Unit]
Description=News At The Edge Fetcher Service
After=network-online.target
Wants=network-online.target
[Service]
Type=oneshot
User=pi # 替换为你的用户名
Group=pi # 替换为你的用户组
WorkingDirectory=/home/pi/news-at-the-edge # 替换为你的项目路径
Environment="PATH=/home/pi/.local/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
ExecStart=/usr/bin/bash /home/pi/news-at-the-edge/run_fetch.sh
StandardOutput=journal
StandardError=journal
# 如果脚本失败，在10秒后重试一次
Restart=on-failure
RestartSec=10
这个服务单元定义了 做什么 。它指定了执行的身份、工作目录、环境变量和具体的启动命令。 Type=oneshot 表示这是一个执行完就退出的任务。 Restart=on-failure 确保了脚本因临时网络问题失败时会自动重试。
然后，创建对应的定时器单元文件 /etc/systemd/system/news-edge-fetcher.timer ：
[Unit]
Description=Run News Fetcher every 2 hours
Requires=news-edge-fetcher.service
[Timer]
OnCalendar=*-*-* 0/2:00:00 # 每两小时执行一次
Persistent=true # 如果错过执行时间（如设备关机），下次启动后立即补执行
RandomizedDelaySec=300 # 随机延迟0-300秒，避免所有边缘设备同时请求，对新闻源友好
[Install]
WantedBy=timers.target
这个定时器单元定义了 何时做 。 OnCalendar 使用了 systemd 的时间表达式，非常灵活。 Persistent=true 是一个关键设置，它保证了即使设备在计划执行时间处于关机状态，开机后也会尽快运行一次任务，不会因为关机而完全错过一次抓取。 RandomizedDelaySec 则是一个礼貌性的设置，将请求时间分散开。
启用并启动定时器：
sudo systemctl daemon-reload
sudo systemctl enable news-edge-fetcher.timer
sudo systemctl start news-edge-fetcher.timer
# 查看定时器状态
sudo systemctl list-timers --all
4.2 输出格式化与推送渠道
经过过滤和排序的新闻，需要以一种友好的格式呈现。我选择生成Markdown文件，因为它格式简单、通用，便于后续处理（如转换为PDF、HTML或发送到支持Markdown的阅读器）。
def generate_markdown_report(filtered_articles, output_path='news_report.md'):
"""生成Markdown格式的新闻简报"""
with open(output_path, 'w', encoding='utf-8') as f:
f.write(f"# 边缘新闻简报 {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n")
f.write(f"> 生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | 共 {len(filtered_articles)} 条精选新闻\n\n")
current_category = None
for article in filtered_articles:
if article['category'] != current_category:
current_category = article['category']
f.write(f"\n## {current_category.upper()}\n\n")
# 使用星级表示得分高低（简单可视化）
score_star = '★' * min(5, int(article.get('score', 0) / 2))
f.write(f"### {article['title']} {score_star}\n")
f.write(f"**来源**：{article['source']} | **发布时间**：{article['published'].strftime('%m-%d %H:%M')}\n\n")
f.write(f"{article['summary'][:200]}...\n\n")
f.write(f"[阅读原文]({article['link']})\n\n")
f.write("---\n\n")
logger.info(f"Markdown简报已生成：{output_path}")
生成文件后，可以通过多种方式推送：
- 邮件推送 ：使用
smtplib将Markdown内容作为邮件正文发送到指定邮箱。可以设置邮件客户端规则，自动将此类邮件转发到Kindle。 - Webhook通知 ：将简报的核心内容（如前5条标题和链接）通过
requests库发送到钉钉、飞书或Slack的Webhook，实现即时通知。 - 同步到云存储 ：使用
rclone或云服务商SDK，将生成的news_report.md自动同步到Dropbox、iCloud或坚果云，方便在手机、平板等多设备查看。
我在 run_fetch.sh 脚本中集成了这些步骤：
#!/bin/bash
cd /home/pi/news-at-the-edge
source $(pipenv --venv)/bin/activate # 激活虚拟环境
# 1. 运行抓取和过滤脚本
python main.py fetch
python main.py filter
# 2. 生成简报
python main.py generate
# 3. 可选：发送邮件
# python main.py send_mail
# 4. 可选：发送Webhook通知
# python main.py send_webhook
logger "News Edge Fetcher completed at $(date)"
5. 常见问题、优化与排查实录
在实际部署和运行过程中，我遇到了不少典型问题，这里记录下排查思路和解决方案。
5.1 抓取失败与网络问题处理
问题现象 ：日志中频繁出现网络超时或连接拒绝错误，导致部分新闻源抓取失败。
排查与解决 ：
- 增加重试机制与超时设置 ：在
requests请求中，必须设置timeout参数（如timeout=(10, 30)，表示连接超时10秒，读取超时30秒）。对于重要源，可以封装一个带重试的请求函数。from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def requests_retry_session(retries=3, backoff_factor=0.5, status_forcelist=(500, 502, 504)): session = requests.Session() retry = Retry( total=retries, read=retries, connect=retries, backoff_factor=backoff_factor, status_forcelist=status_forcelist, ) adapter = HTTPAdapter(max_retries=retry) session.mount('http://', adapter) session.mount('https://', adapter) return session session = requests_retry_session() response = session.get(url, timeout=30) - 使用备用源或镜像 ：对于一些知名的新闻源，可以配置多个URL（如官方源和社区维护的镜像）。在抓取时，按顺序尝试，直到有一个成功。
- 错峰抓取与随机延迟 ：在定时任务中，不要在整点准时运行。使用
systemd timer的RandomizedDelaySec或在脚本开始执行时time.sleep(random.randint(0, 300))，将请求分散开，避免对目标服务器造成压力，也降低被屏蔽的风险。 - 记录详细日志 ：每次抓取都应记录成功/失败状态、响应时间、数据条数。长期观察日志可以帮助你发现某些源是否变得不稳定，以便及时调整或寻找替代源。
5.2 内容解析异常与格式兼容
问题现象 ： feedparser 解析某些RSS源时抛出异常，或者解析出的标题、链接为空。
排查与解决 ：
- 检查
bozo_exception：如前所述，这是第一道防线。根据异常信息判断是网络问题、编码问题还是XML格式问题。 - 处理编码问题 ：有些源的HTTP头声明编码为
UTF-8，但实际内容可能是GB2312。可以在使用feedparser.parse(response.content)前，先尝试用chardet库检测编码，然后手动解码。import chardet detected = chardet.detect(response.content) content = response.content.decode(detected['encoding'] or 'utf-8', errors='ignore') feed = feedparser.parse(content) - 字段缺失的容错处理 ：在提取
entry.title、entry.link等字段时，务必使用entry.get('title', '')的方式，并提供默认值。对于时间字段，如果published_parsed不存在，可以尝试updated_parsed，或者直接使用当前时间，并在日志中标记警告。 - HTML标签清理 ：有些RSS的
summary字段包含大量HTML标签。可以使用BeautifulSoup快速清理，只获取文本。from bs4 import BeautifulSoup summary_text = BeautifulSoup(entry.get('summary', ''), 'html.parser').get_text(strip=True)
5.3 过滤效果不理想与规则调优
问题现象 ：推送的新闻要么太多无关内容，要么漏掉了重要新闻。
排查与解决 ：
- 建立反馈循环 ：在生成的Markdown简报中，为每条新闻添加一个简单的反馈机制。例如，在每条新闻末尾加上
[有用]、[无关]的伪链接（实际可以是本地脚本的调用）。运行一个辅助脚本，定期收集这些反馈日志，并统计哪些关键词或源经常被标记为“无关”，哪些被标记为“有用”。 - 引入否定关键词 ：除了
block_keywords（完全屏蔽），可以增加negative_keywords列表。包含这些词的新闻不会直接被丢弃，但会在评分中被大幅扣减分数。这提供了更细粒度的控制。 - 动态调整权重 ：可以根据历史阅读行为（如果实现了“已读”标记）动态调整规则。例如，某个类别下的新闻如果连续多次被标记为未读或快速跳过，可以自动调低其
category_weights。 - 人工干预入口 ：保留一个“白名单”或“紧急推送”功能。可以创建一个特殊的配置文件
urgent_rules.json，其中定义的规则（如包含某个特定关键词组合）会让新闻直接获得最高分，确保重要信息不被遗漏。
5.4 系统资源与长期运行维护
问题现象 ：运行一段时间后，数据库文件变大，内存使用增加，或日志文件占满磁盘。
排查与解决 ：
- 数据库清理 ：定期清理旧数据。可以在抓取脚本中增加一个维护任务，例如每周删除
published时间超过30天的文章。# 定期清理旧数据 cutoff_date = (datetime.utcnow() - timedelta(days=30)).isoformat() cursor.execute("DELETE FROM articles WHERE published < ?", (cutoff_date,)) logger.info(f"清理了 {cursor.rowcount} 条30天前的旧文章。") - 日志轮转 ：使用
logrotate工具管理应用日志。创建一个配置文件/etc/logrotate.d/news-edge：
这会将日志按天切割，保留最近7天，并压缩旧日志。/home/pi/news-at-the-edge/logs/*.log { daily missingok rotate 7 compress delaycompress notifempty create 644 pi pi } - 监控与告警 ：为关键指标添加简单监控。例如，检查每次抓取的新文章数量是否在正常范围内（如非零）。如果连续多次抓取到0篇文章，可能意味着抓取逻辑失效或网络异常，可以通过发送邮件或Webhook通知自己。
这个“News — At The Edge”项目从构思到落地，让我深刻体会到边缘计算的魅力所在——将控制权和隐私权牢牢握在自己手中，通过可编程的方式定制专属的信息流。它现在每天安静地运行在我的树莓派上，早晨准时将一份精心筛选的简报推送到我的邮箱，成为了我高效获取行业信息不可或缺的工具。整个系统模块清晰，扩展性强，你可以很容易地为其添加新的数据源（比如监控特定GitHub仓库的Release、Hacker News的特定主题）、更复杂的过滤逻辑（如简单的情绪分析），或者更丰富的输出格式。希望这份详细的构建实录，能帮助你打造属于自己的那个“边缘信息管家”。
更多推荐
7
0- 0
已为社区贡献24条内容
所有评论(0)