siftlode/backend/app/sync/playlists.py

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