"""Free per-channel RSS feed reader (no API quota). Returns the channel's most recent ~15 uploads as lightweight stubs for fast fresh-video detection.""" from datetime import datetime, timezone import feedparser import httpx RSS_URL = "https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}" class RssError(Exception): """The feed could not be read (transport failure or a non-200 answer). Deliberately NOT the same as an empty feed: a failed poll used to return `[]`, which the caller stamped as a successful `last_rss_at` — so a channel whose feed 404s permanently (deleted, renamed id) looked like it was being polled fine forever, and nothing surfaced it.""" def fetch_channel_feed(channel_id: str) -> list[dict]: """The channel's recent uploads as stubs. `[]` means the feed really is empty; a failure raises RssError so the caller can leave the channel un-stamped and log it.""" url = RSS_URL.format(channel_id=channel_id) try: resp = httpx.get(url, timeout=20.0, headers={"User-Agent": "Siftlode/1.0"}) except httpx.HTTPError as exc: raise RssError(f"RSS fetch failed for {channel_id}: {exc}") from exc if resp.status_code != 200: raise RssError(f"RSS fetch for {channel_id} returned HTTP {resp.status_code}") parsed = feedparser.parse(resp.content) out: list[dict] = [] for entry in parsed.entries: video_id = entry.get("yt_videoid") if not video_id: continue published = None if entry.get("published_parsed"): published = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc) thumbnail = None media = entry.get("media_thumbnail") or [] if media: thumbnail = media[0].get("url") out.append( { "id": video_id, "channel_id": channel_id, "title": entry.get("title"), "published_at": published, "thumbnail_url": thumbnail, } ) return out