YouTube playlists on the watchlist — backend (resolve, follow, detail)

A playlist is now a first-class watchlist type alongside channels:
- core: parse_playlist_id (URL/bare id; rejects mixes RD… + personal lists) +
  resolve_playlist → {playlist_id, title, channel_title, video_count, thumbnail,
  videos} in the CURATOR's order (partial set, not by year).
- db: add/remove/list/watch_state for kind='playlist' video_watchlist rows
  (mirrors channels; same surrogate scheme, separate kind so they don't collide).
- api: /youtube/resolve now detects playlist links; /youtube/playlist/follow +
  /unfollow; the existing /youtube/playlist/<id> is upgraded to also return the
  playlist meta + following (serves the channel-page expansion AND the new detail
  view from one route); /youtube/channels also returns followed playlists.

Validated live (Lex Fridman playlist resolves title+owner+videos). 190 tests green.
Frontend (search detect + result card + detail view + watchlist display) next.

api/video/youtube.py

@@ -50,16 +50,23 @@ def register_routes(bp):
         except (TypeError, ValueError):
             limit = _RESOLVE_LIMIT
         try:
+            # A playlist link → resolve as a playlist (curator-ordered, partial set).
+            if yt.parse_playlist_id(url):
+                pl = yt.resolve_playlist(url, limit=max(1, min(50, limit)))
+                if not pl:
+                    return jsonify({"success": False, "error": "Could not read that playlist"}), 404
+                following = bool(get_video_db().playlist_watch_state([pl["playlist_id"]]))
+                return jsonify({"success": True, "playlist": pl, "following": following})
             channel = yt.resolve_channel(url, limit=max(1, min(50, limit)))
             if not channel:
                 return jsonify({"success": False,
-                                "error": "Not a YouTube channel link (paste a channel URL like "
-                                         "youtube.com/@handle)"}), 404
+                                "error": "Not a YouTube channel or playlist link (paste a channel "
+                                         "URL like youtube.com/@handle, or a playlist link)"}), 404
             following = bool(get_video_db().channel_watch_state([channel["youtube_id"]]))
             return jsonify({"success": True, "channel": channel, "following": following})
         except Exception:
             logger.exception("youtube resolve failed for %r", url)
-            return jsonify({"success": False, "error": "Could not read that channel"}), 500
+            return jsonify({"success": False, "error": "Could not read that link"}), 500
 
     @bp.route("/youtube/follow", methods=["POST"])
     def video_youtube_follow():
@@ -109,6 +116,40 @@ def register_routes(bp):
             logger.exception("youtube unfollow failed")
             return jsonify({"success": False, "error": "Failed"}), 500
 
+    @bp.route("/youtube/playlist/follow", methods=["POST"])
+    def video_youtube_playlist_follow():
+        """Follow a YouTube playlist. Body: {playlist:{...}} (already resolved) or {url}."""
+        from . import get_video_db
+        from core.video import youtube as yt
+        body = request.get_json(silent=True) or {}
+        db = get_video_db()
+        try:
+            playlist = body.get("playlist")
+            if not playlist:
+                url = (body.get("url") or "").strip()
+                playlist = yt.resolve_playlist(url) if url else None
+            if not playlist or not playlist.get("playlist_id"):
+                return jsonify({"success": False, "error": "Could not resolve playlist"}), 404
+            ok = db.add_playlist_to_watchlist(playlist)
+            return jsonify({"success": ok, "following": ok,
+                            "playlist": {k: playlist.get(k) for k in ("playlist_id", "title", "thumbnail_url")}})
+        except Exception:
+            logger.exception("youtube playlist follow failed")
+            return jsonify({"success": False, "error": "Failed to follow playlist"}), 500
+
+    @bp.route("/youtube/playlist/unfollow", methods=["POST"])
+    def video_youtube_playlist_unfollow():
+        from . import get_video_db
+        pid = ((request.get_json(silent=True) or {}).get("playlist_id") or "").strip()
+        if not pid:
+            return jsonify({"success": False, "error": "playlist_id is required"}), 400
+        try:
+            get_video_db().remove_playlist_from_watchlist(pid)
+            return jsonify({"success": True, "following": False})
+        except Exception:
+            logger.exception("youtube playlist unfollow failed")
+            return jsonify({"success": False, "error": "Failed"}), 500
+
     @bp.route("/youtube/channels", methods=["GET"])
     def video_youtube_channels():
         """Followed channels (newest first) for the watchlist page. Also sweeps:
@@ -118,6 +159,7 @@ def register_routes(bp):
         try:
             db = get_video_db()
             channels = db.list_watchlist_channels()
+            playlists = db.list_watchlist_playlists()
             try:
                 from core.video.youtube_enrichment import get_youtube_date_enricher
                 enr = get_youtube_date_enricher()
@@ -126,7 +168,7 @@ def register_routes(bp):
                         enr.enqueue(c["youtube_id"], c.get("title"))
             except Exception:
                 pass
-            return jsonify({"success": True, "channels": channels,
+            return jsonify({"success": True, "channels": channels, "playlists": playlists,
                             "counts": db.youtube_wishlist_counts()})
         except Exception:
             logger.exception("youtube channels list failed")
@@ -313,16 +355,33 @@ def register_routes(bp):
 
     @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."""
+        """A playlist's videos + metadata + follow state. Serves BOTH the channel-page
+        playlist expansion (reads ``videos``) and the standalone playlist detail view
+        (reads ``playlist`` + ``following``). Curator-ordered — a partial set, NOT
+        grouped by year. Per-video wished + cached-date flags hydrate the toggles."""
         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])
+            limit = int(request.args.get("limit") or 200)
+        except (TypeError, ValueError):
+            limit = 200
+        try:
+            db = get_video_db()
+            pl = yt.resolve_playlist("https://www.youtube.com/playlist?list=" + playlist_id,
+                                     limit=max(1, min(500, limit)))
+            if not pl:
+                return jsonify({"success": False, "error": "Playlist not found"}), 404
+            vids = pl.get("videos") or []
+            ids = [v.get("youtube_id") for v in vids]
+            wished = db.youtube_video_wish_state(ids)
+            dates = db.get_video_dates(ids)
             for v in vids:
                 v["wished"] = v.get("youtube_id") in wished
-            return jsonify({"success": True, "videos": vids})
+                if not v.get("published_at") and dates.get(v.get("youtube_id")):
+                    v["published_at"] = dates[v["youtube_id"]]
+            return jsonify({"success": True, "videos": vids, "playlist": pl,
+                            "following": bool(db.playlist_watch_state([pl["playlist_id"]])),
+                            "kind": "playlist", "source": "youtube"})
         except Exception:
             logger.exception("youtube playlist failed for %r", playlist_id)
             return jsonify({"success": False, "error": "Failed"}), 500

core/video/youtube.py

@@ -706,3 +706,40 @@ def playlist_videos(playlist_id, limit=50, ydl_factory=None):
     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)
+
+
+_PLAYLIST_RE = re.compile(r"[?&]list=([A-Za-z0-9_-]+)")
+
+
+def parse_playlist_id(raw):
+    """A followable playlist id from a URL or bare id, or None. Rejects mixes/radios
+    (RD…) and the personal Watch-Later/Liked lists, which aren't real playlists."""
+    raw = (raw or "").strip()
+    m = _PLAYLIST_RE.search(raw)
+    pid = m.group(1) if m else (raw if re.match(r"^(PL|OL|FL|UU)[A-Za-z0-9_-]{8,}$", raw) else None)
+    if not pid or pid.startswith(("RD", "UL", "LL", "WL")):
+        return None
+    return pid
+
+
+def resolve_playlist(raw, limit=100, ydl_factory=None, db=None):
+    """Resolve a pasted playlist reference → ``{playlist_id, title, channel_title,
+    video_count, thumbnail_url, videos:[...]}``, or None if it isn't a real
+    followable playlist. Order is the curator's (NOT date) — playlists are
+    deliberately a partial, ordered set."""
+    pid = parse_playlist_id(raw)
+    if not pid:
+        return None
+    info = _extract("https://www.youtube.com/playlist?list=" + pid, _ydl_opts(limit, db), ydl_factory)
+    if not info:
+        return None
+    videos = _shape_entries(info.get("entries"), limit)
+    return {
+        "playlist_id": pid,
+        "title": info.get("title") or "Playlist",
+        "channel_title": info.get("channel") or info.get("uploader") or "",
+        "channel_id": info.get("channel_id") or info.get("uploader_id"),
+        "video_count": info.get("playlist_count") or len(videos),
+        "thumbnail_url": _best_thumb(info.get("thumbnails")) or (videos[0]["thumbnail_url"] if videos else None),
+        "videos": videos,
+    }

database/video_database.py

@@ -1880,6 +1880,74 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    # A followed PLAYLIST mirrors a channel: a video_watchlist row (kind='playlist',
+    # source='youtube', source_id=PL id). Same surrogate scheme so dedup just works.
+    def add_playlist_to_watchlist(self, playlist: dict) -> bool:
+        """Follow a YouTube playlist. ``playlist`` = {playlist_id, title, thumbnail_url?}."""
+        pid = (playlist or {}).get("playlist_id")
+        title = (playlist or {}).get("title")
+        if not pid or not title:
+            return False
+        conn = self._get_connection()
+        try:
+            conn.execute(
+                """INSERT INTO video_watchlist (kind, tmdb_id, title, poster_url, source, source_id, state)
+                   VALUES ('playlist', ?, ?, ?, 'youtube', ?, 'follow')
+                   ON CONFLICT(kind, tmdb_id) DO UPDATE SET
+                       state='follow', title=excluded.title,
+                       poster_url=COALESCE(excluded.poster_url, video_watchlist.poster_url),
+                       source='youtube', source_id=excluded.source_id""",
+                (youtube_surrogate_id(pid), title, (playlist or {}).get("thumbnail_url"), pid))
+            conn.commit()
+            return True
+        except Exception:
+            logger.exception("add_playlist_to_watchlist failed (%s)", pid)
+            return False
+        finally:
+            conn.close()
+
+    def remove_playlist_from_watchlist(self, playlist_id: str) -> bool:
+        if not playlist_id:
+            return False
+        conn = self._get_connection()
+        try:
+            conn.execute("DELETE FROM video_watchlist WHERE kind='playlist' AND source_id=?", (playlist_id,))
+            conn.commit()
+            return True
+        finally:
+            conn.close()
+
+    def list_watchlist_playlists(self) -> list[dict]:
+        """Followed playlists (newest first)."""
+        conn = self._get_connection()
+        try:
+            rows = conn.execute(
+                "SELECT title, poster_url, source_id, date_added FROM video_watchlist "
+                "WHERE kind='playlist' AND state='follow' ORDER BY date_added DESC, id DESC").fetchall()
+            return [{"kind": "playlist", "playlist_id": r["source_id"], "title": r["title"],
+                     "poster_url": r["poster_url"], "date_added": r["date_added"]} for r in rows]
+        finally:
+            conn.close()
+
+    def playlist_watch_state(self, playlist_ids) -> dict:
+        """{playlist_id: True} for followed playlists — hydrates the Follow button."""
+        out: dict = {}
+        ids = [str(x) for x in (playlist_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_watchlist WHERE kind='playlist' "
+                        f"AND state='follow' AND source_id IN ({ph})", chunk):
+                    out[r["source_id"]] = True
+            return out
+        finally:
+            conn.close()
+
     def add_videos_to_wishlist(self, channel: dict, videos: list, *, server_source=None) -> int:
         """Wish for a channel's videos. ``channel`` = {youtube_id, title, avatar_url?};
         ``videos`` = [{youtube_id, title, published_at?, thumbnail_url?, description?}, …].

tests/test_video_api.py

@@ -689,6 +689,28 @@ def test_youtube_channel_videos_batch_streams_and_merges(tmp_path, monkeypatch):
     assert r2["continuation"] is None and calls[1] == "NEXT"
 
 
+def test_youtube_playlist_follow_detail_and_watchlist(tmp_path, monkeypatch):
+    client, videoapi = _make_client(tmp_path)
+    import core.video.youtube as ytmod
+    _PL = {"playlist_id": "PLx", "title": "Mix", "channel_title": "Lex", "video_count": 2,
+           "thumbnail_url": "t", "videos": [{"youtube_id": "a", "title": "A"}, {"youtube_id": "b", "title": "B"}]}
+    monkeypatch.setattr(ytmod, "parse_playlist_id", lambda u: "PLx" if "list=" in (u or "") else None)
+    monkeypatch.setattr(ytmod, "resolve_playlist", lambda *a, **k: dict(_PL))
+    # resolve detects a playlist link → returns a playlist (not a channel)
+    r = client.get("/api/video/youtube/resolve?url=https://youtube.com/playlist?list=PLx").get_json()
+    assert r["success"] and r["playlist"]["playlist_id"] == "PLx" and r["following"] is False
+    # follow → appears under the watchlist's playlists
+    assert client.post("/api/video/youtube/playlist/follow", json={"playlist": _PL}).get_json()["following"] is True
+    assert [p["playlist_id"] for p in client.get("/api/video/youtube/channels").get_json()["playlists"]] == ["PLx"]
+    # detail (kept in curator order; flagged following)
+    d = client.get("/api/video/youtube/playlist/PLx").get_json()
+    assert d["kind"] == "playlist" and d["following"] is True
+    assert [v["youtube_id"] for v in d["playlist"]["videos"]] == ["a", "b"]
+    # unfollow
+    assert client.post("/api/video/youtube/playlist/unfollow", json={"playlist_id": "PLx"}).get_json()["following"] is False
+    assert client.get("/api/video/youtube/channels").get_json()["playlists"] == []
+
+
 def test_youtube_wishlist_add_single_video(tmp_path):
     client, _ = _make_client(tmp_path)
     r = client.post("/api/video/youtube/wishlist/add", json={
@@ -750,12 +772,14 @@ def test_youtube_playlists_and_playlist_videos(tmp_path, monkeypatch):
     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"}])
+    monkeypatch.setattr(ytmod, "resolve_playlist",
+                        lambda *a, **k: {"playlist_id": "PL1", "title": "Trailers",
+                                         "videos": [{"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"]}
+    wished = {v["youtube_id"]: v["wished"] for v in pv["videos"]}    # still top-level `videos` for the expansion
     assert wished == {"a": True, "b": False}
 
 

tests/test_video_database.py

@@ -1194,6 +1194,21 @@ def test_legacy_enrichment_rows_upgrade(db):
     assert db.channel_dates_enriched_recently("UCleg") is True
 
 
+def test_playlist_watchlist_roundtrip(db):
+    assert db.add_playlist_to_watchlist({"playlist_id": "PLx", "title": "Mix", "thumbnail_url": "t"}) is True
+    assert db.playlist_watch_state(["PLx", "PLy"]) == {"PLx": True}
+    pls = db.list_watchlist_playlists()
+    assert len(pls) == 1 and pls[0]["playlist_id"] == "PLx" and pls[0]["title"] == "Mix"
+    # channels and playlists are separate kinds — they don't bleed into each other
+    db.add_channel_to_watchlist({"youtube_id": "UCx", "title": "Chan"})
+    assert {p["playlist_id"] for p in db.list_watchlist_playlists()} == {"PLx"}
+    assert db.channel_watch_state(["UCx"]) == {"UCx": True} and db.playlist_watch_state(["UCx"]) == {}
+    assert db.remove_playlist_from_watchlist("PLx") is True
+    assert db.list_watchlist_playlists() == [] and db.playlist_watch_state(["PLx"]) == {}
+    # name/title required
+    assert db.add_playlist_to_watchlist({"playlist_id": "PLn"}) is False
+
+
 def test_remembered_channel_videos_and_meta(db):
     # Cache a list (out of date order) + a date for one of them.
     db.cache_video_dates([{"youtube_id": "b", "published_at": "2020-05-01"}])

tests/test_video_youtube.py

@@ -156,6 +156,39 @@ def test_resolve_channel_rejects_non_channel_without_network():
     assert called == []
 
 
+def test_parse_playlist_id():
+    assert yt.parse_playlist_id("https://www.youtube.com/playlist?list=PLabc123def456") == "PLabc123def456"
+    assert yt.parse_playlist_id("https://www.youtube.com/watch?v=x&list=PLxyz789ghi012") == "PLxyz789ghi012"
+    assert yt.parse_playlist_id("PLabcdefghij123") == "PLabcdefghij123"
+    assert yt.parse_playlist_id("https://youtube.com/watch?v=x&list=RDmix123") is None    # mix → not followable
+    assert yt.parse_playlist_id("https://www.youtube.com/@PlayStation") is None           # channel → None
+    assert yt.parse_playlist_id("") is None
+
+
+class _FakePlaylistYDL:
+    last_url = None
+
+    def __init__(self, opts): self.opts = opts
+    def __enter__(self): return self
+    def __exit__(self, *a): return False
+
+    def extract_info(self, url, download=False):
+        _FakePlaylistYDL.last_url = url
+        return {"title": "Deep Learning", "channel": "Lex Fridman", "channel_id": "UClex", "playlist_count": 3,
+                "thumbnails": [{"url": "http://pl/cover.jpg", "width": 480, "height": 360}],
+                "entries": [{"id": "a", "title": "One"}, {"id": "b", "title": "Two"},
+                            None, {"title": "no id → skip"}]}
+
+
+def test_resolve_playlist_keeps_curator_order_and_meta():
+    pl = yt.resolve_playlist("https://www.youtube.com/playlist?list=PLdeep123", limit=5, ydl_factory=_FakePlaylistYDL)
+    assert pl["playlist_id"] == "PLdeep123" and pl["title"] == "Deep Learning"
+    assert pl["channel_title"] == "Lex Fridman" and pl["video_count"] == 3
+    assert [v["youtube_id"] for v in pl["videos"]] == ["a", "b"]    # order preserved, null/idless dropped
+    assert _FakePlaylistYDL.last_url == "https://www.youtube.com/playlist?list=PLdeep123"
+    assert yt.resolve_playlist("https://youtube.com/@chan", ydl_factory=_FakePlaylistYDL) is None   # not a playlist
+
+
 def test_resolve_channel_returns_none_on_extractor_error():
     class _Boom:
         def __init__(self, opts):