Panniantong
74c3df5c3d
BREAKING: Complete architectural rewrite.
Before: Copied x-reader's fetcher code into readers/ (1205 lines of borrowed code)
After: Pluggable channel system where each channel is a thin wrapper (~50 lines)
around the best external tool for that platform. Zero copied code.
Architecture:
- channels/base.py — Universal Channel interface (read, search, check)
- channels/web.py — Jina Reader API (swappable)
- channels/github.py — GitHub API (swappable)
- channels/twitter.py — birdx + Jina fallback (swappable)
- channels/youtube.py — yt-dlp (swappable)
- channels/reddit.py — Reddit JSON API + proxy (swappable)
- channels/rss.py — feedparser (swappable)
- channels/bilibili.py — Bilibili API (swappable)
- channels/exa_search.py — Exa semantic search (swappable)
Key design: every backend can be swapped by changing ONE file.
YouTube dies? Change youtube.py. Exa sucks? Swap exa_search.py for Tavily.
Nothing else changes.
Removed: reader.py, schema.py, readers/, search/, utils/ (all x-reader code)
Tests: 36/36 passing
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""RSS feeds — via feedparser (free, pip dependency).
|
|
|
|
Backend: feedparser (https://github.com/kurtmckee/feedparser)
|
|
Swap to: any RSS parser
|
|
"""
|
|
|
|
import feedparser
|
|
from urllib.parse import urlparse
|
|
from .base import Channel, ReadResult
|
|
|
|
|
|
class RSSChannel(Channel):
|
|
name = "rss"
|
|
description = "RSS and Atom feeds"
|
|
backends = ["feedparser"]
|
|
tier = 0
|
|
|
|
def can_handle(self, url: str) -> bool:
|
|
lower = url.lower()
|
|
domain = urlparse(url).netloc.lower()
|
|
return (lower.endswith(".xml") or "/rss" in lower or "/feed" in lower
|
|
or "/atom" in lower or "rss" in domain)
|
|
|
|
async def read(self, url: str, config=None) -> ReadResult:
|
|
feed = feedparser.parse(url)
|
|
|
|
if feed.bozo and not feed.entries:
|
|
raise ValueError(f"Failed to parse RSS feed: {url}")
|
|
|
|
if not feed.entries:
|
|
raise ValueError(f"No entries in RSS feed: {url}")
|
|
|
|
# Return latest entry
|
|
entry = feed.entries[0]
|
|
content = entry.get("summary", "") or entry.get("description", "")
|
|
|
|
# If multiple entries, summarize all
|
|
if len(feed.entries) > 1:
|
|
lines = [f"# {feed.feed.get('title', 'RSS Feed')}\n"]
|
|
for i, e in enumerate(feed.entries[:20], 1):
|
|
title = e.get("title", "Untitled")
|
|
link = e.get("link", "")
|
|
summary = e.get("summary", "")[:200]
|
|
lines.append(f"## {i}. {title}")
|
|
lines.append(f"🔗 {link}")
|
|
if summary:
|
|
lines.append(summary)
|
|
lines.append("")
|
|
content = "\n".join(lines)
|
|
|
|
return ReadResult(
|
|
title=feed.feed.get("title", entry.get("title", url)),
|
|
content=content,
|
|
url=url,
|
|
platform="rss",
|
|
)
|