Channel page (backend): channel tags/views + playlists + playlist videos

- shape_channel now carries tags (≤12) + channel view_count for the stats ribbon.
- Refactored entry-shaping into _shape_entries (shared by uploads + playlists).
- channel_playlists(id) → playlists (as 'seasons'); playlist_videos(id) → its
  videos. API: GET /youtube/playlists/<channel_id>, GET /youtube/playlist/<id>
  (with per-video wished hydration). 79 youtube+api tests green.
This commit is contained in:
BoulderBadgeDad 2026-06-17 08:51:53 -07:00
parent bbba2c8225
commit 87750e2464
4 changed files with 126 additions and 7 deletions

View file

@ -183,6 +183,32 @@ def register_routes(bp):
logger.exception("youtube video detail failed for %r", video_id)
return jsonify({"success": False, "error": "Could not load video"}), 500
@bp.route("/youtube/playlists/<channel_id>", methods=["GET"])
def video_youtube_playlists(channel_id):
"""The channel's playlists (rendered as 'seasons' on the channel page)."""
from core.video import youtube as yt
try:
return jsonify({"success": True, "playlists": yt.channel_playlists(channel_id)})
except Exception:
logger.exception("youtube playlists failed for %r", channel_id)
return jsonify({"success": False, "error": "Failed"}), 500
@bp.route("/youtube/playlist/<playlist_id>", methods=["GET"])
def video_youtube_playlist(playlist_id):
"""A playlist's videos (lazy-loaded when a playlist section expands), with
per-video wished flags so the toggles hydrate."""
from . import get_video_db
from core.video import youtube as yt
try:
vids = yt.playlist_videos(playlist_id)
wished = get_video_db().youtube_video_wish_state([v.get("youtube_id") for v in vids])
for v in vids:
v["wished"] = v.get("youtube_id") in wished
return jsonify({"success": True, "videos": vids})
except Exception:
logger.exception("youtube playlist failed for %r", playlist_id)
return jsonify({"success": False, "error": "Failed"}), 500
@bp.route("/youtube/wishlist/add", methods=["POST"])
def video_youtube_wishlist_add():
"""Wish specific videos (per-video add from the channel page). Body:

View file

@ -118,17 +118,16 @@ def _entry_date(e):
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]:
def _shape_entries(entries, limit):
"""Flat playlist/channel entries → our lightweight video shape (id-less and
null entries dropped)."""
out = []
for e in [x for x in (entries or []) if isinstance(x, dict)][:limit]:
vid = e.get("id")
if not vid:
continue
dur = e.get("duration")
videos.append({
out.append({
"youtube_id": vid,
"title": e.get("title") or "",
"published_at": _entry_date(e),
@ -137,7 +136,13 @@ def shape_channel(info, limit=30):
"view_count": e.get("view_count"),
"description": e.get("description"),
})
return out
def shape_channel(info, limit=30):
"""Map a yt-dlp channel info dict to our followable-channel shape."""
info = info or {}
videos = _shape_entries(info.get("entries"), limit)
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
@ -150,6 +155,8 @@ def shape_channel(info, limit=30):
"banner_url": _thumb_by_id(thumbs, "banner"),
"description": info.get("description"),
"subscriber_count": info.get("channel_follower_count"),
"view_count": info.get("view_count"),
"tags": (info.get("tags") or [])[:12],
"video_count": info.get("playlist_count") or len(videos),
"videos": videos,
}
@ -241,3 +248,33 @@ def resolve_channel(raw, limit=30, ydl_factory=None, db=None):
return None
shaped = shape_channel(info, limit)
return shaped if shaped.get("youtube_id") else None
def channel_playlists(channel_id, limit=50, ydl_factory=None):
"""The channel's playlists (flat) → [{playlist_id, title, video_count,
thumbnail_url}] rendered as "seasons" on the channel page. Lazy-loaded."""
if not channel_id:
return []
url = "https://www.youtube.com/channel/" + str(channel_id) + "/playlists"
info = _extract(url, _ydl_opts(limit), ydl_factory)
out = []
for e in [x for x in ((info or {}).get("entries") or []) if isinstance(x, dict)][:limit]:
pid = e.get("id")
if not pid:
continue
out.append({
"playlist_id": pid,
"title": e.get("title") or "",
"video_count": e.get("playlist_count") or e.get("video_count"),
"thumbnail_url": _best_thumb(e.get("thumbnails")) or e.get("thumbnail"),
})
return out
def playlist_videos(playlist_id, limit=50, ydl_factory=None):
"""A playlist's videos (flat), in the same shape as channel uploads."""
if not playlist_id:
return []
url = "https://www.youtube.com/playlist?list=" + str(playlist_id)
info = _extract(url, _ydl_opts(limit), ydl_factory)
return _shape_entries((info or {}).get("entries"), limit)

View file

@ -684,3 +684,20 @@ def test_img_proxy_allows_youtube_cdn_only(tmp_path, monkeypatch):
# still SSRF-safe: arbitrary hosts rejected
assert client.get("/api/video/img?u=https://evil.example.com/x.jpg").status_code == 404
assert client.get("/api/video/img?u=http://yt3.googleusercontent.com/x").status_code == 404 # http
def test_youtube_playlists_and_playlist_videos(tmp_path, monkeypatch):
client, videoapi = _make_client(tmp_path)
import core.video.youtube as ytmod
monkeypatch.setattr(ytmod, "channel_playlists",
lambda cid: [{"playlist_id": "PL1", "title": "Trailers", "video_count": 3, "thumbnail_url": None}])
pls = client.get("/api/video/youtube/playlists/UCx").get_json()
assert pls["success"] is True and pls["playlists"][0]["playlist_id"] == "PL1"
monkeypatch.setattr(ytmod, "playlist_videos",
lambda pid: [{"youtube_id": "a", "title": "A"}, {"youtube_id": "b", "title": "B"}])
# 'a' is wished → should hydrate wished=True
videoapi._video_db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "X"}, [{"youtube_id": "a", "title": "A"}])
pv = client.get("/api/video/youtube/playlist/PL1").get_json()
wished = {v["youtube_id"]: v["wished"] for v in pv["videos"]}
assert wished == {"a": True, "b": False}

View file

@ -246,3 +246,42 @@ def test_video_detail_accepts_watch_url_and_handles_failure():
raise RuntimeError("unavailable")
assert yt.video_detail("x", ydl_factory=_Boom) is None
assert yt.video_detail("", ydl_factory=_FakeVideoYDL) is None
# ── channel tags + playlists ─────────────────────────────────────────────────
def test_shape_channel_includes_tags_and_views():
info = dict(_flat_info()); info["tags"] = ["gaming", "ps5"]; info["view_count"] = 9_000_000
out = yt.shape_channel(info)
assert out["tags"] == ["gaming", "ps5"] and out["view_count"] == 9_000_000
class _PlaylistsYDL:
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": "PL1", "title": "Trailers", "playlist_count": 12,
"thumbnails": [{"url": "http://t/pl1.jpg", "width": 320, "height": 180}]},
{"id": "PL2", "title": "Interviews", "video_count": 5},
None, {"title": "no id"},
]}
def test_channel_playlists_shapes_and_filters():
pls = yt.channel_playlists("UCx", ydl_factory=_PlaylistsYDL)
assert [p["playlist_id"] for p in pls] == ["PL1", "PL2"]
assert pls[0]["title"] == "Trailers" and pls[0]["video_count"] == 12
assert pls[0]["thumbnail_url"] == "http://t/pl1.jpg"
def test_playlist_videos_reuses_entry_shape():
vids = yt.playlist_videos("PL1", ydl_factory=_FakeYDL) # _FakeYDL returns _flat_info()
assert [v["youtube_id"] for v in vids] == ["vid1", "vid2", "vid3"]
assert vids[0]["published_at"] == "2023-11-14"
def test_channel_playlists_empty_on_no_id():
assert yt.channel_playlists("", ydl_factory=_PlaylistsYDL) == []
assert yt.playlist_videos("", ydl_factory=_FakeYDL) == []