RSS 聚合器

按计划监控 RSS/Atom 订阅源，支持按关键词或发布日期过滤，并将摘要路由至您偏好的通知渠道。

安装配置

需安装 Python 包 feedparser：

pip install feedparser

核心脚本

保存为 scripts/fetch_feeds.py：

#!/usr/bin/env python3
"""RSS/Atom feed fetcher with filtering and summarization."""
import feedparser
import sys
import json
from datetime import datetime, timedelta
from pathlib import Path

def parse_date(entry):
    """Extract publication date from entry."""
    for field in ('published_parsed', 'updated_parsed', 'created_parsed'):
        if hasattr(entry, field) and entry.get(field):
            return datetime(*entry[field][:6])
    return None

def fetch_feed(url, max_age_days=None, keyword_filter=None):
    """Fetch and filter feed entries."""
    feed = feedparser.parse(url)
    entries = feed.entries

    # Filter by age
    if max_age_days:
        cutoff = datetime.now() - timedelta(days=max_age_days)
        entries = [e for e in entries if parse_date(e) and parse_date(e) >= cutoff]

    # Filter by keyword
    if keyword_filter:
        kw_lower = keyword_filter.lower()
        entries = [e for e in entries if kw_lower in (e.get('title', '') + e.get('summary', '')).lower()]

    return {
        'title': feed.feed.get('title', url),
        'url': url,
        'entries': [
            {
                'title': e.get('title', 'No title'),
                'link': e.get('link', ''),
                'published': parse_date(e).isoformat() if parse_date(e) else None,
                'summary': e.get('summary', e.get('description', ''))[:500]
            }
            for e in entries
        ]
    }

if __name__ == '__main__':
    url = sys.argv[1] if len(sys.argv) > 1 else ''
    max_age = int(sys.argv[2]) if len(sys.argv) > 2 else None
    keyword = sys.argv[3] if len(sys.argv) > 3 else None

    if not url:
        print(json.dumps({'error': 'URL required'}))
        sys.exit(1)

    result = fetch_feed(url, max_age, keyword)
    print(json.dumps(result, indent=2))

使用示例（Recipes）

示例 1：每日新闻简报

cron_add(
  name="Tech news digest",
  schedule={"kind": "cron", "expr": "0 8 * * 1-5", "tz": "Africa/Johannesburg"},
  payload={
    "kind": "agentTurn",
    "message": "Run: python scripts/fetch_feeds.py https://news.ycombinator.com/rss 7. Then summarize the top 5 stories as a clean bullet list with titles and links."
  },
  delivery={"mode": "announce"},
  sessionTarget="isolated"
)

示例 2：多订阅源监控

// First, create scripts/multi_fetch.py:
"""
import feedparser, json, sys
from scripts.fetch_feeds import fetch_feed

feeds = [
    "https://techcrunch.com/feed/",
    "https://www.theverge.com/rss/index.xml",
    "https://feeds.feedburner.com/TechCrunch/"
]

results = [fetch_feed(url, max_age_days=1) for url in feeds]
print(json.dumps(results, indent=2))
"""

然后添加定时任务：

cron_add(
  name="Industry pulse",
  schedule={"kind": "cron", "expr": "0 */6 * * *", "tz": "UTC"},
  payload={
    "kind": "agentTurn",
    "message": "Run: python scripts/multi_fetch.py. Filter entries from last 6 hours. Post new articles to #news channel on Discord with title + link."
  },
  delivery={"mode": "announce"},
  sessionTarget="isolated"
)

示例 3：关键词告警

cron_add(
  name="AI keyword alert",
  schedule={"kind": "cron", "expr": "0 */4 * * *", "tz": "UTC"},
  payload={
    "kind": "agentTurn",
    "message": "Run: python scripts/fetch_feeds.py https://feeds.feedburner.com/venturebeat/Settings 1 "AI OR machine learning OR LLM". If results have entries, format as: **Alert** [Article Title](URL). Send to Discord #alerts channel."
  },
  delivery={"mode": "webhook", "to": "https://discord.com/api/webhooks/..."},
  sessionTarget="isolated"
)

示例 4：订阅源健康检查

cron_add(
  name="Feed health check",
  schedule={"kind": "cron", "expr": "0 9 * * *", "tz": "UTC"},
  payload={
    "kind": "agentTurn",
    "message": "Check if these feeds are still live: Hacker News (https://news.ycombinator.com/rss), TechCrunch (https://techcrunch.com/feed/). Run fetch without filters. If any feed returns 0 entries or error, alert via webhook."
  },
  delivery={"mode": "announce"},
  sessionTarget="isolated",
  failureAlert={"after": 3, "mode": "announce", "cooldownMs": 86400000}
)

示例 5：订阅源转“稍后阅读”（Notion）

cron_add(
  name="RSS to Notion",
  schedule={"kind": "cron", "expr": "0 7 * * *", "tz": "Africa/Johannesburg"},
  payload={
    "kind": "agentTurn",
    "message": "Run: python scripts/fetch_feeds.py https://example.com/rss 1. Create Notion page for each entry in your Reading List database with title, link, and summary as page content."
  },
  delivery={"mode": "none"},
  sessionTarget="isolated"
)

订阅源管理

# Test a feed directly
python scripts/fetch_feeds.py  [max-age-days] [keyword-filter]

# Example
python scripts/fetch_feeds.py https://news.ycombinator.com/rss 7
python scripts/fetch_feeds.py https://techcrunch.com/feed/ 1 "AI"

订阅源发现

可通过以下方式在任意网站查找 RSS 订阅源：

• 在网址末尾添加 /feed 或 /rss
• 检查网页源码中是否存在
• 在 Google 中使用 site:rss 进行搜索

常见订阅源地址格式：

• YouTube：https://www.youtube.com/feeds/videos.xml?channel_id=CHANNEL_ID
• Twitter/X：无原生 RSS 支持 — 可借助第三方服务（如 Nitter）获取 Twitter 列表的 RSS
• Reddit：https://www.reddit.com/r/SUBREDDIT.rss（获取完整内容需身份认证）

故障排查

参见

• fuzzy-cron-scheduler skill — 实现周期性订阅源检查的调度能力
• notion-integration skill — 将文章存入 Notion 数据库
• discord skill — 将文章推送至 Discord 频道
• webhook-automation skill — 通过 HTTP webhook 向任意终端发送数据