"""Discovery: RSS polling and the subscribe-time backfill. Primary path is the undocumented UULF uploads playlist feed, which excludes Shorts and livestreams at the cheapest possible point (verified — see specs.md §4). The channel_id feed is the fallback, and rows discovered that way carry `discovery_source='uc_feed'` so the download step knows to apply the duration and live-status match filter. """ from __future__ import annotations import logging import sqlite3 import urllib.error import urllib.request import xml.etree.ElementTree as ET from datetime import date, timedelta from . import channels, config, util, videos, ytdlp from .settings import Settings log = logging.getLogger(__name__) NS = { "atom": "http://www.w3.org/2005/Atom", "yt": "http://www.youtube.com/xml/schemas/2015", "media": "http://search.yahoo.com/mrss/", } FEED_BASE = "https://www.youtube.com/feeds/videos.xml" class FeedUnavailable(Exception): """The feed could not be fetched at all (network/5xx). Not the same as 404.""" def uulf_feed_url(channel_id: str) -> str: return f"{FEED_BASE}?playlist_id={channels.uulf_playlist_id(channel_id)}" def uc_feed_url(channel_id: str) -> str: return f"{FEED_BASE}?channel_id={channel_id}" def fetch_feed(url: str, timeout: float = 30.0) -> bytes | None: """Return the feed body, or None if YouTube says it doesn't exist. A 404 on UULF/UUSH/UULV means "no such playlist", i.e. the channel has none of that kind of video — it is not an error. """ request = urllib.request.Request(url, headers={"User-Agent": config.USER_AGENT}) try: with urllib.request.urlopen(request, timeout=timeout) as response: return response.read() except urllib.error.HTTPError as exc: if exc.code == 404: return None raise FeedUnavailable(f"HTTP {exc.code}") from exc except OSError as exc: raise FeedUnavailable(str(exc)) from exc def parse_entries(payload: bytes) -> list[dict]: """Parse an Atom feed into video dicts. The feed carries no duration.""" try: root = ET.fromstring(payload) except ET.ParseError as exc: raise FeedUnavailable(f"unparseable feed: {exc}") from exc entries = [] for entry in root.findall("atom:entry", NS): video_id = entry.findtext("yt:videoId", "", NS) if not video_id: continue published = entry.findtext("atom:published", "", NS) try: published_date = date.fromisoformat(published[:10]) except ValueError: continue description = entry.findtext("media:group/media:description", "", NS) entries.append( { "video_id": video_id, "title": (entry.findtext("atom:title", "", NS) or "").strip(), "published": published_date, "description": description or "", } ) return entries def effective_retention_days(settings: Settings, channel: sqlite3.Row) -> int: override = channel["retention_days"] if "retention_days" in channel.keys() else None if override: return int(override) return settings.get_int("retention_days") # -------------------------------------------------------------------------- def _record( conn: sqlite3.Connection, channel: sqlite3.Row, entry: dict, source: str, cutoff: date, ) -> str: """Insert or repair one discovered video. Returns what happened.""" existing = videos.get(conn, entry["video_id"]) if existing is not None: # The only repair we ever perform: a video the fallback path rejected as # too short, later confirmed long-form by the authoritative UULF feed. # Deleted rows are tombstones and are never touched here. if ( source == videos.SOURCE_UULF and existing["state"] == videos.SKIPPED_SHORT and existing["discovery_source"] == videos.SOURCE_UC ): with conn: conn.execute( "UPDATE video SET state = ?, discovery_source = ?, " "last_error = NULL WHERE video_id = ?", (videos.PENDING, videos.SOURCE_UULF, entry["video_id"]), ) log.info( "re-queued %s: UULF confirms it is long-form", entry["video_id"] ) return "repaired" return "known" state = videos.PENDING if entry["published"] >= cutoff else videos.SKIPPED_OLD videos.insert( conn, channel_pk=channel["id"], video_id=entry["video_id"], title=entry["title"], upload_date=entry["published"].isoformat(), state=state, discovery_source=source, ) return "queued" if state == videos.PENDING else "old" def poll_channel(conn: sqlite3.Connection, settings: Settings, channel: sqlite3.Row) -> dict: """Poll one channel. Never raises for feed problems — records them instead.""" stats = {"queued": 0, "old": 0, "known": 0, "repaired": 0, "source": None} source = videos.SOURCE_UULF try: payload = fetch_feed(uulf_feed_url(channel["channel_id"])) entries = parse_entries(payload) if payload else [] if not entries: # UULF 404'd or came back empty — fall back to the channel feed. log.warning( "UULF feed empty for %s, falling back to channel_id feed", channel["title"], ) source = videos.SOURCE_UC payload = fetch_feed(uc_feed_url(channel["channel_id"])) entries = parse_entries(payload) if payload else [] except FeedUnavailable as exc: _record_poll_failure(conn, channel, str(exc)) stats["error"] = str(exc) return stats cutoff = util.today() - timedelta(days=effective_retention_days(settings, channel)) for entry in entries: stats[_record(conn, channel, entry, source, cutoff)] += 1 stats["source"] = source _record_poll_success(conn, channel) return stats def _record_poll_success(conn: sqlite3.Connection, channel: sqlite3.Row) -> None: with conn: conn.execute( "UPDATE channel SET last_polled_at = ?, last_poll_ok = 1, " "consecutive_poll_failures = 0 WHERE id = ?", (util.utcnow_iso(), channel["id"]), ) def _record_poll_failure(conn: sqlite3.Connection, channel: sqlite3.Row, error: str) -> None: log.error("poll failed for %s: %s", channel["title"], error) with conn: conn.execute( "UPDATE channel SET last_polled_at = ?, last_poll_ok = 0, " "consecutive_poll_failures = consecutive_poll_failures + 1 WHERE id = ?", (util.utcnow_iso(), channel["id"]), ) # -------------------------------------------------------------------------- # backfill def _flat_playlist(settings: Settings, channel_id: str, limit: int) -> list[dict]: """Reverse-chronological upload list. Entries carry no upload date.""" args = [ "--flat-playlist", "--playlist-end", str(limit), "-J", "--no-warnings", "--ignore-config", *ytdlp.extractor_args(settings.get_str("pot_provider_url")), f"https://www.youtube.com/playlist?list={channels.uulf_playlist_id(channel_id)}", ] try: data = ytdlp.run_json(args, timeout=300) except ytdlp.YtdlpError as exc: log.warning("flat playlist failed for %s: %s", channel_id, exc) return [] return [entry for entry in (data.get("entries") or []) if entry.get("id")] def _upload_date(settings: Settings, video_id: str) -> date | None: """One extraction to learn a single video's upload date.""" args = [ "--skip-download", "--no-warnings", "--ignore-config", "--no-playlist", "--print", "%(upload_date)s", *ytdlp.extractor_args(settings.get_str("pot_provider_url")), f"https://www.youtube.com/watch?v={video_id}", ] result = ytdlp.run(args, timeout=180) text = (result.stdout or "").strip().splitlines() if result.returncode != 0 or not text: return None try: from . import naming return naming.parse_upload_date(text[-1]) except ValueError: return None def backfill_channel( conn: sqlite3.Connection, settings: Settings, channel: sqlite3.Row ) -> dict: """Queue the last `backfill_days` for a freshly subscribed channel. The RSS feed is the primary source because it is the only one that carries upload dates — `--flat-playlist` reports `timestamp: None` for every entry. RSS returns ~15 items, which covers the default 7-day window for any channel uploading less than twice a day. Only when the feed's oldest entry is still inside the window do we extend via the playlist, resolving those extra dates one video at a time. """ days = settings.get_int("backfill_days") cutoff = util.today() - timedelta(days=days) stats = {"queued": 0, "old": 0, "known": 0, "repaired": 0, "extended": 0} try: payload = fetch_feed(uulf_feed_url(channel["channel_id"])) entries = parse_entries(payload) if payload else [] source = videos.SOURCE_UULF if not entries: source = videos.SOURCE_UC payload = fetch_feed(uc_feed_url(channel["channel_id"])) entries = parse_entries(payload) if payload else [] except FeedUnavailable as exc: _record_poll_failure(conn, channel, str(exc)) stats["error"] = str(exc) return stats seen = set() for entry in entries: seen.add(entry["video_id"]) stats[_record(conn, channel, entry, source, cutoff)] += 1 oldest = min((entry["published"] for entry in entries), default=None) if oldest is not None and oldest >= cutoff: # The feed did not reach past the window, so there may be more. log.info( "%s: RSS reaches only to %s, extending backfill via playlist", channel["title"], oldest.isoformat(), ) for item in _flat_playlist(settings, channel["channel_id"], 50): video_id = item["id"] if video_id in seen or videos.exists(conn, video_id): continue upload_date = _upload_date(settings, video_id) if upload_date is None: continue if upload_date < cutoff: break # playlist is reverse-chronological; everything after is older videos.insert( conn, channel_pk=channel["id"], video_id=video_id, title=(item.get("title") or "").strip(), upload_date=upload_date.isoformat(), duration=item.get("duration"), state=videos.PENDING, discovery_source=videos.SOURCE_BACKFILL, ) stats["extended"] += 1 with conn: conn.execute("UPDATE channel SET backfilled = 1 WHERE id = ?", (channel["id"],)) _record_poll_success(conn, channel) return stats def rescan_channel( conn: sqlite3.Connection, settings: Settings, channel: sqlite3.Row ) -> int: """Re-queue `skipped_old` rows that the current retention window now covers. `skipped_old` is judged against whatever window was in force at discovery time, and it is otherwise terminal. Without this, raising a channel's retention_days would appear to do nothing for an infrequent uploader — every one of their videos is already marked old. This is deliberately an explicit action rather than something poll does silently, because doing it on every poll would make `backfill_days` meaningless: it would immediately re-queue everything the initial backfill had deliberately left behind. Tombstones (`deleted`) are never touched. """ cutoff = util.today() - timedelta(days=effective_retention_days(settings, channel)) with conn: cursor = conn.execute( "UPDATE video SET state = ?, last_error = NULL " "WHERE channel_pk = ? AND state = ? AND upload_date >= ?", (videos.PENDING, channel["id"], videos.SKIPPED_OLD, cutoff.isoformat()), ) if cursor.rowcount: log.info( "%s: re-queued %d video(s) now inside the %s window", channel["title"], cursor.rowcount, cutoff.isoformat(), ) return cursor.rowcount def poll_all(conn: sqlite3.Connection, settings: Settings, channel_pk: int | None = None) -> dict: """Backfill anything new, then poll everything. Returns aggregate counts.""" if channel_pk is not None: rows = [row for row in [channels.get(conn, channel_pk)] if row is not None] else: rows = channels.all_channels(conn) totals = {"queued": 0, "old": 0, "known": 0, "repaired": 0, "extended": 0, "failed": 0} for channel in rows: if not channel["backfilled"]: stats = backfill_channel(conn, settings, channel) else: stats = poll_channel(conn, settings, channel) if "error" in stats: totals["failed"] += 1 for key in ("queued", "old", "known", "repaired", "extended"): totals[key] += stats.get(key, 0) log.info("%s: %s", channel["title"], stats) return totals