diff --git a/core/video/youtube.py b/core/video/youtube.py new file mode 100644 index 00000000..5673e8ac --- /dev/null +++ b/core/video/youtube.py @@ -0,0 +1,192 @@ +"""Resolve a YouTube *channel* reference into a followable channel + its recent +uploads — the data source behind "follow a YouTube channel" on the video side. + +yt-dlp only (no API key). The upload list is fetched with ``extract_flat`` so it +stays cheap even for channels with thousands of videos; the trade-off is that +per-video publish dates are sparse in flat mode, so they get backfilled later +(same pattern as the wishlist art backfill). Real *downloads* will additionally +need Deno + ffmpeg; flat *listing* does not. + +The URL parsing and the yt-dlp-dict → our-shape mapping are pure functions so +they're unit-testable without touching the network; the one network call goes +through ``_extract`` which accepts an injectable factory for tests. +""" + +from __future__ import annotations + +import logging +import re +from datetime import datetime, timezone +from urllib.parse import urlparse + +try: + import yt_dlp +except Exception: # pragma: no cover - yt-dlp is a hard dep, but stay import-safe + yt_dlp = None + +logger = logging.getLogger(__name__) + +# Path prefixes that identify a channel (everything after is the channel key). +# Tabs like /videos, /streams, /shorts, /featured, /playlists are stripped. +_CHANNEL_TABS = {"videos", "streams", "shorts", "featured", "playlists", "community", "about"} +_HANDLE_RE = re.compile(r"^@?[A-Za-z0-9._-]{2,}$") +_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36") + + +def parse_channel_url(raw): + """Normalize a pasted channel reference to a canonical uploads URL, or return + None if it isn't a channel (a /watch video, a playlist, the YT home, etc.). + + Accepts: ``https://www.youtube.com/@PlayStation`` (+ /videos etc.), + ``/channel/UC...``, ``/c/Name``, ``/user/Name``, a bare ``@handle``, or a + bare ``handle``. Returns ``https://www.youtube.com//videos``. + """ + if not raw or not str(raw).strip(): + return None + raw = str(raw).strip() + + # Bare handle (no scheme, no slash) → treat as @handle. + if "/" not in raw and "." not in raw and " " not in raw: + if _HANDLE_RE.match(raw): + handle = raw if raw.startswith("@") else "@" + raw + return "https://www.youtube.com/" + handle + "/videos" + return None + + # Give bare "youtube.com/..." a scheme so urlparse populates netloc. + parse_target = raw if "://" in raw else "https://" + raw + try: + u = urlparse(parse_target) + except Exception: + return None + host = (u.netloc or "").lower() + if host.startswith("www."): + host = host[4:] + if host not in ("youtube.com", "m.youtube.com", "music.youtube.com"): + return None + + segs = [s for s in (u.path or "").split("/") if s] + if not segs: + return None + + first = segs[0] + if first.startswith("@"): + base = first # /@handle + elif first in ("channel", "c", "user") and len(segs) >= 2: + base = first + "/" + segs[1] # /channel/UC.., /c/Name, /user/Name + else: + return None # /watch, /playlist, /shorts/, home… + + return "https://www.youtube.com/" + base + "/videos" + + +def _best_thumb(thumbs): + """Pick the highest-resolution thumbnail URL from a yt-dlp thumbnails list.""" + if not thumbs: + return None + best, best_area = None, -1 + for t in thumbs: + if not isinstance(t, dict) or not t.get("url"): + continue + area = (t.get("width") or 0) * (t.get("height") or 0) + # preference value as a tie-breaker when dimensions are absent + score = area if area else (t.get("preference") or 0) + if score >= best_area: + best, best_area = t.get("url"), score + return best + + +def _entry_date(e): + """ISO date (YYYY-MM-DD) for a flat entry, or None — flat mode often omits it.""" + ts = e.get("timestamp") + if isinstance(ts, (int, float)): + try: + return datetime.fromtimestamp(ts, tz=timezone.utc).date().isoformat() + except Exception: + pass + ud = e.get("upload_date") # 'YYYYMMDD' + if isinstance(ud, str) and len(ud) == 8 and ud.isdigit(): + return f"{ud[0:4]}-{ud[4:6]}-{ud[6:8]}" + return None + + +def shape_channel(info, limit=30): + """Map a yt-dlp channel info dict to our followable-channel shape.""" + info = info or {} + entries = [e for e in (info.get("entries") or []) if isinstance(e, dict)] + videos = [] + for e in entries[:limit]: + vid = e.get("id") + if not vid: + continue + dur = e.get("duration") + videos.append({ + "youtube_id": vid, + "title": e.get("title") or "", + "published_at": _entry_date(e), + "duration_seconds": int(dur) if isinstance(dur, (int, float)) else None, + "thumbnail_url": _best_thumb(e.get("thumbnails")) or e.get("thumbnail"), + "view_count": e.get("view_count"), + "description": e.get("description"), + }) + + handle = info.get("uploader_id") or info.get("channel_id_handle") + if handle and not str(handle).startswith("@"): + handle = None # uploader_id is sometimes the UC id, not a handle + return { + "youtube_id": info.get("channel_id") or info.get("id"), + "title": info.get("channel") or info.get("uploader") or info.get("title") or "", + "handle": handle, + "avatar_url": _best_thumb(info.get("thumbnails")), + "description": info.get("description"), + "subscriber_count": info.get("channel_follower_count"), + "video_count": info.get("playlist_count") or len(videos), + "videos": videos, + } + + +def _ydl_opts(limit, db=None): + opts = { + "quiet": True, + "no_warnings": True, + "extract_flat": True, # fast: enumerate uploads without per-video format probing + "skip_download": True, + "playlistend": int(limit), + "user_agent": _UA, + } + # Reuse the music client's cookie convention so age/region-gated channels work. + try: + from config.settings import config_manager + cb = config_manager.get("youtube.cookies_browser", "") + if cb: + opts["cookiesfrombrowser"] = (cb,) + except Exception: + pass + return opts + + +def _extract(url, opts, ydl_factory=None): + """Run yt-dlp's extract_info; isolated for test injection. Returns a dict or None.""" + factory = ydl_factory or (yt_dlp.YoutubeDL if yt_dlp else None) + if factory is None: + logger.warning("yt-dlp unavailable; cannot resolve YouTube channel") + return None + try: + with factory(opts) as ydl: + return ydl.extract_info(url, download=False) + except Exception as e: + logger.info("YouTube channel resolve failed for %s: %s", url, e) + return None + + +def resolve_channel(raw, limit=30, ydl_factory=None, db=None): + """Resolve a pasted channel reference to ``{youtube_id, title, handle, + avatar_url, videos:[...], ...}``, or None if it isn't a resolvable channel.""" + url = parse_channel_url(raw) + if not url: + return None + info = _extract(url, _ydl_opts(limit, db), ydl_factory) + if not info: + return None + shaped = shape_channel(info, limit) + return shaped if shaped.get("youtube_id") else None diff --git a/tests/test_video_youtube.py b/tests/test_video_youtube.py new file mode 100644 index 00000000..ff8d16d5 --- /dev/null +++ b/tests/test_video_youtube.py @@ -0,0 +1,188 @@ +"""Seam tests for core/video/youtube.py — the YouTube channel resolver. + +Pure URL parsing + yt-dlp-dict → our-shape mapping are tested directly; the one +network call is exercised through an injected fake YoutubeDL, so nothing here +touches the network. +""" + +import pytest + +from core.video import youtube as yt + + +# ── parse_channel_url ──────────────────────────────────────────────────────── + +@pytest.mark.parametrize("raw,expected", [ + ("https://www.youtube.com/@PlayStation", "https://www.youtube.com/@PlayStation/videos"), + ("https://www.youtube.com/@PlayStation/videos", "https://www.youtube.com/@PlayStation/videos"), + ("https://www.youtube.com/@PlayStation/streams", "https://www.youtube.com/@PlayStation/videos"), + ("http://youtube.com/@GoodMythicalMorning", "https://www.youtube.com/@GoodMythicalMorning/videos"), + ("youtube.com/@PlayStation", "https://www.youtube.com/@PlayStation/videos"), + ("m.youtube.com/@PlayStation", "https://www.youtube.com/@PlayStation/videos"), + ("https://www.youtube.com/channel/UCabc123", "https://www.youtube.com/channel/UCabc123/videos"), + ("https://www.youtube.com/c/LinusTechTips", "https://www.youtube.com/c/LinusTechTips/videos"), + ("https://www.youtube.com/user/PewDiePie", "https://www.youtube.com/user/PewDiePie/videos"), + ("@PlayStation", "https://www.youtube.com/@PlayStation/videos"), + ("PlayStation", "https://www.youtube.com/@PlayStation/videos"), # bare → @handle +]) +def test_parse_channel_url_accepts_channel_forms(raw, expected): + assert yt.parse_channel_url(raw) == expected + + +@pytest.mark.parametrize("raw", [ + "", + " ", + None, + "https://www.youtube.com/watch?v=dQw4w9WgXcQ", # a video, not a channel + "https://www.youtube.com/playlist?list=PL123", # a playlist + "https://www.youtube.com/", # home + "https://youtu.be/dQw4w9WgXcQ", # short video link + "https://vimeo.com/@someone", # not youtube + "https://www.youtube.com/shorts/abc123", # a short, not a channel +]) +def test_parse_channel_url_rejects_non_channels(raw): + assert yt.parse_channel_url(raw) is None + + +# ── shape_channel ──────────────────────────────────────────────────────────── + +def _flat_info(): + return { + "channel_id": "UCPlayStation", + "channel": "PlayStation", + "uploader": "PlayStation", + "uploader_id": "@PlayStation", + "channel_follower_count": 14_000_000, + "playlist_count": 1200, + "thumbnails": [ + {"url": "http://img/small.jpg", "width": 88, "height": 88}, + {"url": "http://img/big.jpg", "width": 800, "height": 800}, + ], + "entries": [ + {"id": "vid1", "title": "State of Play", "timestamp": 1_700_000_000, + "duration": 3600, "view_count": 50000, + "thumbnails": [{"url": "http://t/1.jpg", "width": 320, "height": 180}]}, + {"id": "vid2", "title": "Trailer", "upload_date": "20240115", "duration": 120, + "thumbnail": "http://t/2.jpg"}, + {"id": "vid3", "title": "No date video"}, # sparse flat entry + None, # yt-dlp can yield Nones + {"title": "missing id — skip"}, # no id → skipped + ], + } + + +def test_shape_channel_maps_channel_fields(): + out = yt.shape_channel(_flat_info()) + assert out["youtube_id"] == "UCPlayStation" + assert out["title"] == "PlayStation" + assert out["handle"] == "@PlayStation" + assert out["avatar_url"] == "http://img/big.jpg" # highest-res thumb wins + assert out["subscriber_count"] == 14_000_000 + assert out["video_count"] == 1200 # playlist_count, not len(videos) + + +def test_shape_channel_maps_and_filters_videos(): + out = yt.shape_channel(_flat_info()) + vids = out["videos"] + # the None and the id-less entry are dropped + assert [v["youtube_id"] for v in vids] == ["vid1", "vid2", "vid3"] + # timestamp → ISO date + assert vids[0]["published_at"] == "2023-11-14" + assert vids[0]["duration_seconds"] == 3600 + assert vids[0]["thumbnail_url"] == "http://t/1.jpg" + # upload_date 'YYYYMMDD' → ISO date; plain 'thumbnail' string honored + assert vids[1]["published_at"] == "2024-01-15" + assert vids[1]["thumbnail_url"] == "http://t/2.jpg" + # sparse entry: no date, no crash + assert vids[2]["published_at"] is None + assert vids[2]["duration_seconds"] is None + + +def test_shape_channel_respects_limit(): + out = yt.shape_channel(_flat_info(), limit=1) + assert len(out["videos"]) == 1 + assert out["videos"][0]["youtube_id"] == "vid1" + + +def test_shape_channel_uploader_id_not_a_handle_is_dropped(): + info = {"channel_id": "UCx", "channel": "X", "uploader_id": "UCx", "entries": []} + assert yt.shape_channel(info)["handle"] is None + + +# ── resolve_channel (network call injected) ────────────────────────────────── + +class _FakeYDL: + """Stand-in for yt_dlp.YoutubeDL: records opts, returns a canned info dict.""" + last_opts = None + last_url = None + + def __init__(self, opts): + _FakeYDL.last_opts = opts + self.opts = opts + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def extract_info(self, url, download=False): + _FakeYDL.last_url = url + assert download is False + return _flat_info() + + +def test_resolve_channel_happy_path_uses_canonical_url_and_limit(): + out = yt.resolve_channel("https://www.youtube.com/@PlayStation", limit=2, + ydl_factory=_FakeYDL) + assert out["youtube_id"] == "UCPlayStation" + assert len(out["videos"]) == 2 + # resolver normalizes to the /videos uploads URL and passes the limit through + assert _FakeYDL.last_url == "https://www.youtube.com/@PlayStation/videos" + assert _FakeYDL.last_opts["playlistend"] == 2 + assert _FakeYDL.last_opts["extract_flat"] is True + + +def test_resolve_channel_rejects_non_channel_without_network(): + called = [] + + def factory(opts): + called.append(opts) + raise AssertionError("should not be called for a non-channel URL") + + assert yt.resolve_channel("https://www.youtube.com/watch?v=abc", ydl_factory=factory) is None + assert called == [] + + +def test_resolve_channel_returns_none_on_extractor_error(): + class _Boom: + def __init__(self, opts): + pass + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def extract_info(self, url, download=False): + raise RuntimeError("DownloadError: channel not found") + + assert yt.resolve_channel("@nope", ydl_factory=_Boom) is None + + +def test_resolve_channel_none_when_info_has_no_channel_id(): + class _NoId: + def __init__(self, opts): + pass + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def extract_info(self, url, download=False): + return {"entries": [{"id": "v", "title": "t"}]} # no channel_id/id + + assert yt.resolve_channel("@x", ydl_factory=_NoId) is None