danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

YouTube next-level (backend): year=season nebula shape + channel detail API

Browse source 
- Resolver: capture banner_url + separate avatar (by thumbnail id), keeping
  per-video duration/views for the detail hero.
- query_youtube_wishlist reshaped to the EXACT TV-nebula shape: channel = show,
  upload YEAR = season (newest first), video = episode (newest=ep1 within a
  year), carrying source='youtube' + youtube_id + per-video source_id. Surrogate
  int returned as tmdb_id so the nebula's int keying just works.
- New API: GET /youtube/channel/<id> (full channel detail — meta + deeper
  uploads + following + per-video wished flags) and POST /youtube/wishlist/add
  (per-video wish). DB: youtube_video_wish_state, remove_one_video_from_wishlist.
- Tests updated for the nebula shape + banner + detail endpoints. 82 DB + 69
  api/youtube tests green.
This commit is contained in:
BoulderBadgeDad 2026-06-17 01:10:53 -07:00
parent f4f40c161b
commit 8941da8b60
6 changed files with 179 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
api/video/youtube.py
View file

@ -129,6 +129,51 @@ def register_routes(bp):
logger.exception("youtube wishlist list failed")
return jsonify({"success": False, "error": "Failed"}), 500
@bp.route("/youtube/channel/<channel_id>", methods=["GET"])
def video_youtube_channel_detail(channel_id):
"""Full channel detail for the in-app channel page: meta + a deeper page of
uploads, the follow state, and per-video wished flags. Resolves live."""
from . import get_video_db
from core.video import youtube as yt
try:
limit = int(request.args.get("limit") or 60)
except (TypeError, ValueError):
limit = 60
try:
db = get_video_db()
channel = yt.resolve_channel("https://www.youtube.com/channel/" + channel_id,
limit=max(1, min(90, limit)))
if not channel or not channel.get("youtube_id"):
return jsonify({"success": False, "error": "Channel not found"}), 404
cid = channel["youtube_id"]
following = bool(db.channel_watch_state([cid]))
wished = db.youtube_video_wish_state([v.get("youtube_id") for v in channel.get("videos") or []])
for v in channel.get("videos") or []:
v["wished"] = v.get("youtube_id") in wished
return jsonify({"success": True, "kind": "channel", "source": "youtube",
"channel": channel, "following": following})
except Exception:
logger.exception("youtube channel detail failed for %r", channel_id)
return jsonify({"success": False, "error": "Could not load channel"}), 500
@bp.route("/youtube/wishlist/add", methods=["POST"])
def video_youtube_wishlist_add():
"""Wish specific videos (per-video add from the channel page). Body:
{channel: {youtube_id, title, avatar_url?}, videos: [{youtube_id, title, …}]}."""
from . import get_video_db
body = request.get_json(silent=True) or {}
channel = body.get("channel") or {}
videos = body.get("videos") or []
if not channel.get("youtube_id") or not videos:
return jsonify({"success": False, "error": "channel and videos required"}), 400
try:
db = get_video_db()
n = db.add_videos_to_wishlist(channel, videos, server_source=_server())
return jsonify({"success": n > 0, "added": n, "counts": db.youtube_wishlist_counts()})
except Exception:
logger.exception("youtube wishlist add failed")
return jsonify({"success": False, "error": "Failed"}), 500
@bp.route("/youtube/wishlist/remove", methods=["POST"])
def video_youtube_wishlist_remove():
"""Remove wished videos. Body: {scope: 'channel'|'video', source_id}."""

12
core/video/youtube.py
View file

@ -96,6 +96,14 @@ def _best_thumb(thumbs):
return best
def _thumb_by_id(thumbs, keyword):
"""The highest-res thumbnail whose yt-dlp id mentions ``keyword`` (e.g.
'avatar' / 'banner'), or None — channels expose both in one thumbnails list."""
hits = [t for t in (thumbs or []) if isinstance(t, dict) and t.get("url")
and keyword in str(t.get("id", "")).lower()]
return _best_thumb(hits)
def _entry_date(e):
"""ISO date (YYYY-MM-DD) for a flat entry, or None — flat mode often omits it."""
ts = e.get("timestamp")
@ -133,11 +141,13 @@ def shape_channel(info, limit=30):
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
thumbs = info.get("thumbnails")
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")),
"avatar_url": _thumb_by_id(thumbs, "avatar") or _best_thumb(thumbs),
"banner_url": _thumb_by_id(thumbs, "banner"),
"description": info.get("description"),
"subscriber_count": info.get("channel_follower_count"),
"video_count": info.get("playlist_count") or len(videos),

62
database/video_database.py
View file

@ -1934,6 +1934,31 @@ class VideoDatabase:
finally:
conn.close()
def youtube_video_wish_state(self, video_ids) -> set:
"""Which of ``video_ids`` (youtube ids) are already wished — hydrates the
per-video buttons on the channel detail page."""
out: set = set()
ids = [str(x) for x in (video_ids or []) if x]
if not ids:
return out
conn = self._get_connection()
try:
for i in range(0, len(ids), 400):
chunk = ids[i:i + 400]
ph = ",".join("?" * len(chunk))
for r in conn.execute(
f"SELECT source_id FROM video_wishlist WHERE kind='video' "
f"AND source_id IN ({ph})", chunk):
out.add(r["source_id"])
return out
finally:
conn.close()
def remove_one_video_from_wishlist(self, video_id) -> int:
"""Remove a single wished video by its youtube id. (Thin alias kept explicit
for the detail page's per-video toggle.)"""
return self.remove_youtube_from_wishlist("video", video_id)
def youtube_wishlist_counts(self) -> dict:
"""{'channel': n distinct channels, 'video': n videos} in the wishlist."""
conn = self._get_connection()
@ -1946,9 +1971,10 @@ class VideoDatabase:
conn.close()
def query_youtube_wishlist(self, *, search=None, sort="added", page=1, limit=60) -> dict:
"""Wished YouTube videos grouped by channel (channel = orb, videos = flat
newest-first feed). ``sort`` ∈ added | oldest | title | wanted. Mirrors the
{items, pagination} shape of query_wishlist."""
"""Wished YouTube videos shaped exactly like the TV nebula: channel = show,
YEAR = season, video = episode. Each channel returns ``seasons`` grouped by
upload year (newest first), videos as episodes (newest first within a year).
``sort`` ∈ added | oldest | title | wanted. {items, pagination} like query_wishlist."""
try:
page = max(1, int(page or 1))
limit = max(1, min(200, int(limit or 60)))
@ -1967,8 +1993,8 @@ class VideoDatabase:
order = {"title": "title COLLATE NOCASE", "wanted": "video_count DESC, last_added DESC",
"oldest": "last_added ASC", "added": "last_added DESC"}.get(sort, "last_added DESC")
chan_rows = conn.execute(
"SELECT parent_source_id, MAX(title) AS title, MAX(poster_url) AS poster_url, "
"COUNT(*) AS video_count, "
"SELECT parent_source_id, MAX(tmdb_id) AS surrogate, MAX(title) AS title, "
"MAX(poster_url) AS poster_url, COUNT(*) AS video_count, "
"SUM(CASE WHEN status='downloaded' THEN 1 ELSE 0 END) AS done, "
"MAX(date_added) AS last_added "
"FROM video_wishlist" + wsql +
@ -1980,13 +2006,27 @@ class VideoDatabase:
"SELECT source_id, episode_title, still_url, episode_overview, air_date, status "
"FROM video_wishlist WHERE kind='video' AND parent_source_id=? "
"ORDER BY (air_date IS NULL), air_date DESC, id DESC", (cr["parent_source_id"],)).fetchall()
# group by upload year → "seasons"; newest video in a year = episode 1
by_year: dict = {}
for v in vids:
ad = v["air_date"]
yr = int(ad[:4]) if ad and len(ad) >= 4 and ad[:4].isdigit() else 0
by_year.setdefault(yr, []).append(v)
seasons = []
for yr in sorted(by_year, reverse=True): # newest year first
eps = []
for i, v in enumerate(by_year[yr]):
eps.append({"episode_number": i + 1, "title": v["episode_title"],
"still_url": v["still_url"], "overview": v["episode_overview"],
"air_date": v["air_date"], "status": v["status"],
"source_id": v["source_id"]})
poster = next((e["still_url"] for e in eps if e["still_url"]), cr["poster_url"])
seasons.append({"season_number": yr, "year": yr, "poster_url": poster, "episodes": eps})
items.append({
"kind": "channel", "youtube_id": cr["parent_source_id"], "title": cr["title"],
"poster_url": cr["poster_url"], "video_count": cr["video_count"],
"done": cr["done"] or 0,
"videos": [{"youtube_id": v["source_id"], "title": v["episode_title"],
"still_url": v["still_url"], "overview": v["episode_overview"],
"published_at": v["air_date"], "status": v["status"]} for v in vids]})
"kind": "channel", "source": "youtube",
"tmdb_id": cr["surrogate"], "youtube_id": cr["parent_source_id"],
"title": cr["title"], "poster_url": cr["poster_url"],
"wanted": cr["video_count"], "done": cr["done"] or 0, "seasons": seasons})
finally:
conn.close()
total_pages = max(1, (total + limit - 1) // limit)

40
tests/test_video_api.py
View file

@ -569,11 +569,12 @@ def test_youtube_follow_then_channels_and_wishlist(tmp_path):
assert chans["channels"][0]["youtube_id"] == "UCPlay"
assert chans["channels"][0]["video_count"] == 2
# appears in the youtube wishlist grouped by channel, newest-first
wl = client.get("/api/video/youtube/wishlist?kind=channel").get_json()
# appears in the youtube wishlist as a nebula channel (year=season, video=episode)
wl = client.get("/api/video/youtube/wishlist").get_json()
grp = wl["items"][0]
assert grp["youtube_id"] == "UCPlay" and grp["video_count"] == 2
assert [v["youtube_id"] for v in grp["videos"]] == ["v1", "v2"]
assert grp["youtube_id"] == "UCPlay" and grp["wanted"] == 2 and grp["source"] == "youtube"
vids = [e["source_id"] for se in grp["seasons"] for e in se["episodes"]]
assert set(vids) == {"v1", "v2"}
# resolve now reports following=True (hydration) — stub resolve to avoid network
import core.video.youtube as ytmod
@ -610,3 +611,34 @@ def test_youtube_follow_requires_url_or_channel(tmp_path):
client, _ = _make_client(tmp_path)
r = client.post("/api/video/youtube/follow", json={})
assert r.status_code == 400
def test_youtube_channel_detail_hydrates_follow_and_wished(tmp_path, monkeypatch):
client, videoapi = _make_client(tmp_path)
import core.video.youtube as ytmod
monkeypatch.setattr(ytmod, "resolve_channel", lambda url, limit=60: dict(_CHANNEL))
# follow first so detail reports following + one wished video
client.post("/api/video/youtube/follow", json={"channel": _CHANNEL})
videoapi._video_db.remove_one_video_from_wishlist("v2") # leave only v1 wished
d = client.get("/api/video/youtube/channel/UCPlay").get_json()
assert d["success"] is True and d["kind"] == "channel" and d["following"] is True
wished = {v["youtube_id"]: v["wished"] for v in d["channel"]["videos"]}
assert wished == {"v1": True, "v2": False}
def test_youtube_channel_detail_404_on_unresolvable(tmp_path, monkeypatch):
client, _ = _make_client(tmp_path)
import core.video.youtube as ytmod
monkeypatch.setattr(ytmod, "resolve_channel", lambda url, limit=60: None)
assert client.get("/api/video/youtube/channel/UCnope").status_code == 404
def test_youtube_wishlist_add_single_video(tmp_path):
client, _ = _make_client(tmp_path)
r = client.post("/api/video/youtube/wishlist/add", json={
"channel": {"youtube_id": "UCPlay", "title": "PlayStation"},
"videos": [{"youtube_id": "solo1", "title": "One Video", "published_at": "2024-03-03"}]})
assert r.get_json()["added"] == 1
assert r.get_json()["counts"] == {"channel": 1, "video": 1}
r = client.post("/api/video/youtube/wishlist/remove", json={"scope": "video", "source_id": "solo1"})
assert r.get_json()["counts"] == {"channel": 0, "video": 0}

46
tests/test_video_database.py
View file

@ -1025,40 +1025,58 @@ def test_unfollow_channel_removes_row(db):
assert db.channel_watch_state(["UCPlay"]) == {}
def test_add_videos_groups_under_channel_newest_first(db):
def test_youtube_wishlist_nebula_shape_year_as_season(db):
"""Channel = show, YEAR = season, video = episode — exact TV-nebula shape."""
ch = {"youtube_id": "UCPlay", "title": "PlayStation", "avatar_url": "http://a/p.jpg"}
vids = [
{"youtube_id": "v1", "title": "Old Trailer", "published_at": "2023-01-01",
"thumbnail_url": "http://t/1.jpg", "description": "older"},
{"youtube_id": "v2", "title": "New State of Play", "published_at": "2024-06-01",
"thumbnail_url": "http://t/2.jpg", "description": "newer"},
{"youtube_id": "v2b", "title": "Mid 2024", "published_at": "2024-02-01", "thumbnail_url": "http://t/2b.jpg"},
{"youtube_id": "v3", "title": "Undated", "thumbnail_url": "http://t/3.jpg"},
]
assert db.add_videos_to_wishlist(ch, vids) == 3
assert db.add_videos_to_wishlist(ch, vids) == 4
res = db.query_youtube_wishlist()
assert res["pagination"]["total_count"] == 1
grp = res["items"][0]
assert grp["kind"] == "channel" and grp["source"] == "youtube"
assert grp["youtube_id"] == "UCPlay" and grp["title"] == "PlayStation"
assert grp["poster_url"] == "http://a/p.jpg" and grp["video_count"] == 3
# newest-first; dated before undated
assert [v["youtube_id"] for v in grp["videos"]] == ["v2", "v1", "v3"]
v2 = grp["videos"][0]
assert v2["title"] == "New State of Play" and v2["still_url"] == "http://t/2.jpg"
assert v2["overview"] == "newer" and v2["published_at"] == "2024-06-01"
assert v2["status"] == "wanted"
assert grp["poster_url"] == "http://a/p.jpg" and grp["wanted"] == 4
assert isinstance(grp["tmdb_id"], int) # surrogate the nebula keys on
# seasons = years, newest first: 2024, 2023, then 0 (undated)
assert [se["season_number"] for se in grp["seasons"]] == [2024, 2023, 0]
y2024 = grp["seasons"][0]
# newest video in the year is episode 1; its still seeds the season poster
assert [e["title"] for e in y2024["episodes"]] == ["New State of Play", "Mid 2024"]
assert [e["episode_number"] for e in y2024["episodes"]] == [1, 2]
assert y2024["poster_url"] == "http://t/2.jpg"
e1 = y2024["episodes"][0]
assert e1["source_id"] == "v2" and e1["overview"] == "newer" and e1["air_date"] == "2024-06-01"
def test_add_videos_is_idempotent_per_video(db):
ch = {"youtube_id": "UCPlay", "title": "PlayStation"}
db.add_videos_to_wishlist(ch, [{"youtube_id": "v1", "title": "A"}])
db.add_videos_to_wishlist(ch, [{"youtube_id": "v1", "title": "A (updated)"},
{"youtube_id": "v2", "title": "B"}])
db.add_videos_to_wishlist(ch, [{"youtube_id": "v1", "title": "A", "published_at": "2024-01-01"}])
db.add_videos_to_wishlist(ch, [{"youtube_id": "v1", "title": "A (updated)", "published_at": "2024-01-01"},
{"youtube_id": "v2", "title": "B", "published_at": "2024-02-01"}])
grp = db.query_youtube_wishlist()["items"][0]
assert grp["video_count"] == 2
titles = {v["youtube_id"]: v["title"] for v in grp["videos"]}
assert grp["wanted"] == 2
titles = {}
for se in grp["seasons"]:
for e in se["episodes"]:
titles[e["source_id"]] = e["title"]
assert titles == {"v1": "A (updated)", "v2": "B"}
def test_youtube_video_wish_state_hydrates(db):
db.add_videos_to_wishlist({"youtube_id": "UCx", "title": "X"},
[{"youtube_id": "a", "title": "A"}, {"youtube_id": "b", "title": "B"}])
assert db.youtube_video_wish_state(["a", "b", "c"]) == {"a", "b"}
assert db.remove_one_video_from_wishlist("a") == 1
assert db.youtube_video_wish_state(["a", "b"]) == {"b"}
def test_youtube_counts_and_removal_scopes(db):
a = {"youtube_id": "UCa", "title": "Chan A"}
b = {"youtube_id": "UCb", "title": "Chan B"}

6
tests/test_video_youtube.py
View file

@ -56,7 +56,8 @@ def _flat_info():
"playlist_count": 1200,
"thumbnails": [
{"url": "http://img/small.jpg", "width": 88, "height": 88},
{"url": "http://img/big.jpg", "width": 800, "height": 800},
{"url": "http://img/avatar.jpg", "id": "avatar_uncropped", "width": 800, "height": 800},
{"url": "http://img/banner.jpg", "id": "banner_uncropped", "width": 2048, "height": 1152},
],
"entries": [
{"id": "vid1", "title": "State of Play", "timestamp": 1_700_000_000,
@ -76,7 +77,8 @@ def test_shape_channel_maps_channel_fields():
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["avatar_url"] == "http://img/avatar.jpg" # picked by id, not just size
assert out["banner_url"] == "http://img/banner.jpg" # banner separated from avatar
assert out["subscriber_count"] == 14_000_000
assert out["video_count"] == 1200 # playlist_count, not len(videos)