From a661d079eeefb1251bbb00cb13d303a12565e36a Mon Sep 17 00:00:00 2001 From: npeter83 Date: Mon, 15 Jun 2026 19:37:03 +0200 Subject: [PATCH] feat(playlists): YouTube playlist read-sync (backend) Mirror each user's own YouTube playlists into local source='youtube' playlists, one-way (YT -> local). New client methods iter_my_playlists / iter_my_playlist_video_ids (OAuth, so private playlists work); sync/playlists.py reconciles the mirror (matching YT order) and ingests any playlist videos not in the shared catalog yet (with stub channels). POST /api/playlists/sync-youtube for manual sync (read-scope gated, per-user quota) plus a scheduler job (playlist_sync_minutes, default 6h) that syncs all read-scope users. YouTube's Watch Later / History are not API-accessible and are never synced. --- backend/app/config.py | 1 + backend/app/routes/playlists.py | 20 ++++- backend/app/scheduler.py | 13 +++ backend/app/sync/playlists.py | 152 ++++++++++++++++++++++++++++++++ backend/app/youtube/client.py | 43 +++++++++ 5 files changed, 228 insertions(+), 1 deletion(-) create mode 100644 backend/app/sync/playlists.py diff --git a/backend/app/config.py b/backend/app/config.py index 3e7db21..52bbd8a 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -89,6 +89,7 @@ class Settings(BaseSettings): autotag_title_sample: int = 40 autotag_interval_minutes: int = 30 subscriptions_resync_minutes: int = 360 + playlist_sync_minutes: int = 360 # live_status values hidden from the feed by default. Completed-stream VODs # ("was_live") are real watchable content and stay visible. feed_default_hidden_live: str = "live,upcoming" diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index 885cd4f..7cc3bfb 100644 --- a/backend/app/routes/playlists.py +++ b/backend/app/routes/playlists.py @@ -5,10 +5,12 @@ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import func, select, and_ from sqlalchemy.orm import Session, aliased -from app.auth import current_user +from app import quota +from app.auth import current_user, has_read_scope from app.db import get_db from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState from app.routes.feed import _saved_expr, _serialize +from app.sync.playlists import sync_user_playlists router = APIRouter(prefix="/api/playlists", tags=["playlists"]) @@ -187,6 +189,22 @@ def remove_watch_later( return {"saved": False} +@router.post("/sync-youtube") +def sync_youtube( + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + """Pull the user's YouTube playlists into local mirrors now (read-only direction). + Also runs automatically on the scheduler; this is the manual 'sync now'.""" + if not has_read_scope(user): + raise HTTPException( + status_code=403, + detail="Connect YouTube (read access) to sync your playlists.", + ) + with quota.attribute(user.id, "playlist_sync"): + return sync_user_playlists(db, user) + + @router.get("/{playlist_id}") def get_playlist( playlist_id: int, diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index bd93d07..8d6d0c3 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -9,6 +9,7 @@ from app.config import settings from app.db import SessionLocal from app.state import is_sync_paused from app.sync.autotag import run_autotag_all +from app.sync.playlists import sync_all_playlists from app.sync.runner import ( run_deep_backfill, run_enrich, @@ -79,6 +80,12 @@ def _subscriptions_job() -> None: _job("subscriptions", work) +def _playlist_sync_job() -> None: + # Mirror each read-scope user's YouTube playlists; per-user quota attribution is + # handled inside sync_all_playlists. + _job("playlist_sync", sync_all_playlists) + + def start_scheduler() -> None: global _scheduler if not settings.scheduler_enabled or _scheduler is not None: @@ -114,6 +121,12 @@ def start_scheduler() -> None: minutes=settings.subscriptions_resync_minutes, id="subscriptions", ) + scheduler.add_job( + _playlist_sync_job, + "interval", + minutes=settings.playlist_sync_minutes, + id="playlist_sync", + ) scheduler.start() _scheduler = scheduler logger.info("scheduler started") diff --git a/backend/app/sync/playlists.py b/backend/app/sync/playlists.py new file mode 100644 index 0000000..1b185e7 --- /dev/null +++ b/backend/app/sync/playlists.py @@ -0,0 +1,152 @@ +"""Read-direction YouTube playlist sync: mirror each user's own YouTube playlists into +local `source='youtube'` playlists (kept fresh by the scheduler). One-way (YT -> local); +the write-back direction (local edits -> YouTube) is a later phase. + +YouTube's special Watch Later / History playlists are not exposed by the Data API and are +never synced. Videos in a playlist that aren't in our shared catalog yet are fetched and +ingested (with a stub channel) so the mirror is faithful.""" +import logging +from datetime import datetime, timezone + +from sqlalchemy import delete, select +from sqlalchemy.orm import Session + +from app import quota +from app.auth import has_read_scope +from app.models import Channel, OAuthToken, Playlist, PlaylistItem, User, Video +from app.sync.videos import apply_video_details, parse_dt +from app.youtube.client import YouTubeClient, YouTubeError + +log = logging.getLogger("subfeed.sync") + + +def _ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None: + """Make sure every given video id exists in the catalog, fetching + ingesting the + missing ones (and a stub channel for any unknown channel). Deleted/private videos that + the API won't return are simply left out.""" + if not video_ids: + return + have = set( + db.execute(select(Video.id).where(Video.id.in_(video_ids))).scalars().all() + ) + missing = [v for v in dict.fromkeys(video_ids) if v not in have] + if not missing: + return + items = yt.get_videos(missing) + chan_ids = { + it.get("snippet", {}).get("channelId") + for it in items + if it.get("snippet", {}).get("channelId") + } + existing_ch = set( + db.execute(select(Channel.id).where(Channel.id.in_(chan_ids))).scalars().all() + ) + now = datetime.now(timezone.utc) + for it in items: + sn = it.get("snippet", {}) + cid = sn.get("channelId") + if cid and cid not in existing_ch: + db.add(Channel(id=cid, title=sn.get("channelTitle"))) + existing_ch.add(cid) + db.flush() + seen: set[str] = set() + for it in items: + vid = it.get("id") + sn = it.get("snippet", {}) + cid = sn.get("channelId") + if not vid or not cid or vid in seen: + continue + seen.add(vid) + v = Video(id=vid, channel_id=cid, published_at=parse_dt(sn.get("publishedAt"))) + apply_video_details(v, it) + v.enriched_at = now + db.add(v) + db.commit() + + +def sync_user_playlists(db: Session, user: User) -> dict: + """Mirror the user's YouTube playlists into local source='youtube' playlists. Leaves + the user's local playlists and the built-in Watch later untouched.""" + if not has_read_scope(user): + return {"synced": 0, "reason": "no read scope"} + synced = 0 + with YouTubeClient(db, user) as yt: + yt_playlists = list(yt.iter_my_playlists()) + yt_ids = {p["id"] for p in yt_playlists if p.get("id")} + existing = { + pl.yt_playlist_id: pl + for pl in db.execute( + select(Playlist).where( + Playlist.user_id == user.id, Playlist.source == "youtube" + ) + ).scalars() + } + # Drop mirrors whose YouTube playlist no longer exists. + for ytid, pl in existing.items(): + if ytid not in yt_ids: + db.delete(pl) + db.flush() + + for idx, p in enumerate(yt_playlists): + ytid = p.get("id") + if not ytid: + continue + pl = existing.get(ytid) + if pl is None: + pl = Playlist( + user_id=user.id, + name=p.get("title") or "Untitled", + kind="user", + source="youtube", + yt_playlist_id=ytid, + position=1000 + idx, # sort mirrored lists after local ones + ) + db.add(pl) + db.flush() + else: + pl.name = p.get("title") or pl.name + ids = list(yt.iter_my_playlist_video_ids(ytid)) + _ensure_videos(db, yt, ids) + present = ( + set(db.execute(select(Video.id).where(Video.id.in_(ids))).scalars().all()) + if ids + else set() + ) + ordered: list[str] = [] + seen: set[str] = set() + for v in ids: + if v in present and v not in seen: + seen.add(v) + ordered.append(v) + # Mirror is authoritative: replace the items to match YouTube's order. + db.execute(delete(PlaylistItem).where(PlaylistItem.playlist_id == pl.id)) + for pos, vid in enumerate(ordered): + db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=pos)) + db.commit() + synced += 1 + return {"synced": synced} + + +def sync_all_playlists(db: Session) -> dict: + """Scheduler entry point: mirror playlists for every user with a read scope + token.""" + users = ( + db.execute( + select(User).join(OAuthToken).where(OAuthToken.refresh_token_enc.is_not(None)) + ) + .scalars() + .all() + ) + total = 0 + for user in users: + if not has_read_scope(user): + continue + try: + with quota.attribute(user.id, "playlist_sync"): + total += sync_user_playlists(db, user).get("synced", 0) + except YouTubeError: + db.rollback() + log.warning("Playlist sync failed for user %s", user.id) + except Exception: + db.rollback() + log.exception("Playlist sync crashed for user %s", user.id) + return {"users": len(users), "playlists_synced": total} diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py index 62f2125..9d22961 100644 --- a/backend/app/youtube/client.py +++ b/backend/app/youtube/client.py @@ -125,6 +125,49 @@ class YouTubeClient: if not page_token: break + def iter_my_playlists(self) -> Iterator[dict]: + """Yield the authenticated user's own playlists (OAuth, so private ones are + included). Note: YouTube's special Watch Later / History playlists are NOT + returned by the Data API and cannot be synced.""" + page_token = None + while True: + params = {"part": "snippet,contentDetails", "mine": "true", "maxResults": 50} + if page_token: + params["pageToken"] = page_token + data = self._get("playlists", params, cost=1, allow_key=False) + for item in data.get("items", []): + sn = item.get("snippet", {}) + yield { + "id": item.get("id"), + "title": sn.get("title"), + "item_count": item.get("contentDetails", {}).get("itemCount"), + "thumbnail_url": best_thumbnail(sn.get("thumbnails")), + } + page_token = data.get("nextPageToken") + if not page_token: + break + + def iter_my_playlist_video_ids(self, playlist_id: str) -> Iterator[str]: + """Yield the video ids of one of the user's own playlists, in playlist order + (OAuth, so private/unlisted playlists work).""" + page_token = None + while True: + params = { + "part": "contentDetails", + "playlistId": playlist_id, + "maxResults": 50, + } + if page_token: + params["pageToken"] = page_token + data = self._get("playlistItems", params, cost=1, allow_key=False) + for item in data.get("items", []): + vid = item.get("contentDetails", {}).get("videoId") + if vid: + yield vid + page_token = data.get("nextPageToken") + if not page_token: + break + def get_channels(self, channel_ids: list[str]) -> list[dict]: items: list[dict] = [] for batch in _chunks(channel_ids, 50):