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.
This commit is contained in:
parent
cbf5037374
commit
240b6fb611
5 changed files with 228 additions and 1 deletions
|
|
@ -89,6 +89,7 @@ class Settings(BaseSettings):
|
||||||
autotag_title_sample: int = 40
|
autotag_title_sample: int = 40
|
||||||
autotag_interval_minutes: int = 30
|
autotag_interval_minutes: int = 30
|
||||||
subscriptions_resync_minutes: int = 360
|
subscriptions_resync_minutes: int = 360
|
||||||
|
playlist_sync_minutes: int = 360
|
||||||
# live_status values hidden from the feed by default. Completed-stream VODs
|
# live_status values hidden from the feed by default. Completed-stream VODs
|
||||||
# ("was_live") are real watchable content and stay visible.
|
# ("was_live") are real watchable content and stay visible.
|
||||||
feed_default_hidden_live: str = "live,upcoming"
|
feed_default_hidden_live: str = "live,upcoming"
|
||||||
|
|
|
||||||
|
|
@ -5,10 +5,12 @@ from fastapi import APIRouter, Depends, HTTPException
|
||||||
from sqlalchemy import func, select, and_
|
from sqlalchemy import func, select, and_
|
||||||
from sqlalchemy.orm import Session, aliased
|
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.db import get_db
|
||||||
from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState
|
from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState
|
||||||
from app.routes.feed import _saved_expr, _serialize
|
from app.routes.feed import _saved_expr, _serialize
|
||||||
|
from app.sync.playlists import sync_user_playlists
|
||||||
|
|
||||||
router = APIRouter(prefix="/api/playlists", tags=["playlists"])
|
router = APIRouter(prefix="/api/playlists", tags=["playlists"])
|
||||||
|
|
||||||
|
|
@ -187,6 +189,22 @@ def remove_watch_later(
|
||||||
return {"saved": False}
|
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}")
|
@router.get("/{playlist_id}")
|
||||||
def get_playlist(
|
def get_playlist(
|
||||||
playlist_id: int,
|
playlist_id: int,
|
||||||
|
|
|
||||||
|
|
@ -9,6 +9,7 @@ from app.config import settings
|
||||||
from app.db import SessionLocal
|
from app.db import SessionLocal
|
||||||
from app.state import is_sync_paused
|
from app.state import is_sync_paused
|
||||||
from app.sync.autotag import run_autotag_all
|
from app.sync.autotag import run_autotag_all
|
||||||
|
from app.sync.playlists import sync_all_playlists
|
||||||
from app.sync.runner import (
|
from app.sync.runner import (
|
||||||
run_deep_backfill,
|
run_deep_backfill,
|
||||||
run_enrich,
|
run_enrich,
|
||||||
|
|
@ -79,6 +80,12 @@ def _subscriptions_job() -> None:
|
||||||
_job("subscriptions", work)
|
_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:
|
def start_scheduler() -> None:
|
||||||
global _scheduler
|
global _scheduler
|
||||||
if not settings.scheduler_enabled or _scheduler is not None:
|
if not settings.scheduler_enabled or _scheduler is not None:
|
||||||
|
|
@ -114,6 +121,12 @@ def start_scheduler() -> None:
|
||||||
minutes=settings.subscriptions_resync_minutes,
|
minutes=settings.subscriptions_resync_minutes,
|
||||||
id="subscriptions",
|
id="subscriptions",
|
||||||
)
|
)
|
||||||
|
scheduler.add_job(
|
||||||
|
_playlist_sync_job,
|
||||||
|
"interval",
|
||||||
|
minutes=settings.playlist_sync_minutes,
|
||||||
|
id="playlist_sync",
|
||||||
|
)
|
||||||
scheduler.start()
|
scheduler.start()
|
||||||
_scheduler = scheduler
|
_scheduler = scheduler
|
||||||
logger.info("scheduler started")
|
logger.info("scheduler started")
|
||||||
|
|
|
||||||
152
backend/app/sync/playlists.py
Normal file
152
backend/app/sync/playlists.py
Normal file
|
|
@ -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}
|
||||||
|
|
@ -125,6 +125,49 @@ class YouTubeClient:
|
||||||
if not page_token:
|
if not page_token:
|
||||||
break
|
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]:
|
def get_channels(self, channel_ids: list[str]) -> list[dict]:
|
||||||
items: list[dict] = []
|
items: list[dict] = []
|
||||||
for batch in _chunks(channel_ids, 50):
|
for batch in _chunks(channel_ids, 50):
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue