youtube-automate/youtube_automate/discovery.py

"""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