YouTube channels (1/4): channel resolver in core/video/youtube.py
First slice of 'follow a YouTube channel as a show'. Pure, testable resolver: - parse_channel_url() normalizes any channel reference (@handle, /channel/UC.., /c/.., /user/.., bare handle, with/without scheme or tab suffix) to a canonical /videos uploads URL, and rejects non-channels (watch/playlist/ shorts/home/non-youtube). - shape_channel() maps yt-dlp's flat-extract dict to our shape (channel meta + recent uploads), picking best-res thumbs, parsing sparse publish dates (timestamp or upload_date → ISO, None when flat mode omits them), filtering null/id-less entries. - resolve_channel() ties them together via an injectable YoutubeDL factory. yt-dlp only (already a pinned dep), extract_flat for cheap listing. 28 seam tests, no network. Schema bridge + API + UI are the next slices.
This commit is contained in:
BoulderBadgeDad
parent
b7cafb8b72
commit
68383a16b3
2 changed files with 380 additions and 0 deletions
192
core/video/youtube.py
Normal file
192
core/video/youtube.py
Normal file
|
|
@ -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/<base>/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/<id>, 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
|
||||
188
tests/test_video_youtube.py
Normal file
188
tests/test_video_youtube.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue