peter
b1832afd18
Mechanical, whole-tree `ruff format` pass — no behaviour change. Recorded in .git-blame-ignore-revs (next commit) so git blame skips it.
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
"""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
|