"""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}