bc1eb62bfe
Store a SHA-256 of each playlist's name + ordered video ids as last synced from / pushed to YouTube (migration 0013, backfilled for already-clean linked playlists). 'dirty' is now recomputed on every edit by comparing the current fingerprint to that baseline instead of being stickily set true. So manually restoring the original order (or undo back to it) clears dirty on its own — the Reset to YouTube button disappears and no YouTube read is needed. Baseline is set on read-sync, repull (reset), and a clean push; a missing baseline counts as dirty (unknown YT state). Verified the migration's fingerprint matches the app's runtime fingerprint exactly.
394 lines
16 KiB
Python
394 lines
16 KiB
Python
"""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 hashlib
|
|
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, has_write_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")
|
|
|
|
# Cost of a single YouTube write op (playlists/playlistItems insert/update/delete).
|
|
WRITE_UNIT_COST = 50
|
|
|
|
|
|
def fingerprint(name: str, ids: list[str]) -> str:
|
|
"""A stable checksum of a playlist's name + ordered video ids. Stored as the synced
|
|
baseline; the current fingerprint is compared to it to derive `dirty`."""
|
|
h = hashlib.sha256()
|
|
h.update((name or "").encode())
|
|
h.update(b"\n")
|
|
h.update(",".join(ids).encode())
|
|
return h.hexdigest()
|
|
|
|
|
|
def current_fingerprint(db: Session, pl: Playlist) -> str:
|
|
return fingerprint(pl.name, _local_video_ids(db, pl))
|
|
|
|
|
|
def recompute_dirty(db: Session, pl: Playlist) -> None:
|
|
"""Set `dirty` by comparing the current state to the synced baseline. Unlinked lists
|
|
(and Watch later) are never dirty; a missing baseline counts as dirty (unknown YT state)."""
|
|
if not pl.yt_playlist_id:
|
|
pl.dirty = False
|
|
else:
|
|
pl.dirty = pl.synced_fingerprint != current_fingerprint(db, pl)
|
|
|
|
|
|
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()
|
|
}
|
|
# YouTube playlists that we already own as local (exported) playlists must not be
|
|
# re-mirrored, or every exported list would also show up as a read-only duplicate.
|
|
owned_yt_ids = set(
|
|
db.execute(
|
|
select(Playlist.yt_playlist_id).where(
|
|
Playlist.user_id == user.id,
|
|
Playlist.source == "local",
|
|
Playlist.yt_playlist_id.is_not(None),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
# 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 or ytid in owned_yt_ids:
|
|
continue
|
|
pl = existing.get(ytid)
|
|
if pl is not None and pl.dirty:
|
|
# Local edits not yet pushed: don't clobber them with YouTube's state.
|
|
# A successful 'Sync to YouTube' clears dirty and lets the mirror refresh.
|
|
continue
|
|
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))
|
|
pl.synced_fingerprint = fingerprint(pl.name, ordered)
|
|
pl.dirty = False
|
|
db.commit()
|
|
synced += 1
|
|
return {"synced": synced}
|
|
|
|
|
|
def _local_video_ids(db: Session, pl: Playlist) -> list[str]:
|
|
return list(
|
|
db.execute(
|
|
select(PlaylistItem.video_id)
|
|
.where(PlaylistItem.playlist_id == pl.id)
|
|
.order_by(PlaylistItem.position)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
|
|
def _reorder_moves(current: list[str], desired: list[str]) -> int:
|
|
"""Count the playlistItems.update moves an insertion-sort would need to turn
|
|
`current` order into `desired` order (both already the same set of video ids).
|
|
Items already in place are skipped; this matches what apply_push actually executes."""
|
|
model = list(current)
|
|
moves = 0
|
|
for i, want in enumerate(desired):
|
|
if i < len(model) and model[i] == want:
|
|
continue
|
|
j = model.index(want, i)
|
|
model.pop(j)
|
|
model.insert(i, want)
|
|
moves += 1
|
|
return moves
|
|
|
|
|
|
def plan_push(db: Session, user: User, pl: Playlist) -> dict:
|
|
"""Compute (without writing) what pushing this local playlist to YouTube would do:
|
|
create vs update, how many inserts/deletes/reorders, the quota estimate, and whether
|
|
YouTube has diverged (items present there that local will remove). Costs read units
|
|
(1/page) for a linked playlist; nothing for a fresh export."""
|
|
desired = _local_video_ids(db, pl)
|
|
if not pl.yt_playlist_id:
|
|
ops = 1 + len(desired) # create the playlist + one insert per video
|
|
return {
|
|
"action": "create",
|
|
"to_insert": len(desired),
|
|
"to_delete": 0,
|
|
"to_reorder": 0,
|
|
"rename": False,
|
|
"yt_extra": 0,
|
|
"units_estimate": ops * WRITE_UNIT_COST,
|
|
}
|
|
with YouTubeClient(db, user) as yt:
|
|
current = list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id))
|
|
snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
|
|
rename = bool(snippet) and (snippet.get("title") or "") != pl.name
|
|
cur_vids = [it["video_id"] for it in current]
|
|
cur_set = set(cur_vids)
|
|
desired_set = set(desired)
|
|
to_insert = [v for v in desired if v not in cur_set]
|
|
to_delete = [v for v in cur_vids if v not in desired_set]
|
|
# After membership reconciliation, the YT order is current-minus-deletes plus inserts
|
|
# appended; the reorder pass then moves items into the desired order.
|
|
after_membership = [v for v in cur_vids if v in desired_set] + to_insert
|
|
reorders = _reorder_moves(after_membership, desired)
|
|
ops = len(to_insert) + len(to_delete) + reorders + (1 if rename else 0)
|
|
return {
|
|
"action": "update",
|
|
"to_insert": len(to_insert),
|
|
"to_delete": len(to_delete),
|
|
"to_reorder": reorders,
|
|
"rename": rename,
|
|
"yt_extra": len(to_delete),
|
|
"units_estimate": ops * WRITE_UNIT_COST,
|
|
}
|
|
|
|
|
|
def push_playlist(db: Session, user: User, pl: Playlist) -> dict:
|
|
"""Push a local playlist to YouTube: create it if needed, then reconcile membership
|
|
and order so YouTube matches local (local wins). Marks the playlist clean on success.
|
|
Caller must have checked the write scope and the quota budget."""
|
|
desired = _local_video_ids(db, pl)
|
|
inserted = deleted = reordered = created = 0
|
|
failures: list[str] = []
|
|
with YouTubeClient(db, user) as yt:
|
|
if not pl.yt_playlist_id:
|
|
pl.yt_playlist_id = yt.create_playlist(pl.name)
|
|
db.commit()
|
|
created = 1
|
|
# Brand-new playlist: just append every video in order.
|
|
for vid in desired:
|
|
try:
|
|
yt.add_playlist_item(pl.yt_playlist_id, vid)
|
|
inserted += 1
|
|
except YouTubeError:
|
|
failures.append(vid)
|
|
if not failures:
|
|
pl.synced_fingerprint = fingerprint(pl.name, desired)
|
|
pl.dirty = False
|
|
db.commit()
|
|
return {
|
|
"created": created,
|
|
"inserted": inserted,
|
|
"deleted": deleted,
|
|
"reordered": reordered,
|
|
"failures": failures,
|
|
"yt_playlist_id": pl.yt_playlist_id,
|
|
}
|
|
|
|
# Existing link: diff against the live YouTube state.
|
|
snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
|
|
if snippet is not None and (snippet.get("title") or "") != pl.name:
|
|
try:
|
|
yt.update_playlist_title(pl.yt_playlist_id, pl.name)
|
|
except YouTubeError:
|
|
failures.append("title")
|
|
current = list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id))
|
|
item_id_by_vid = {it["video_id"]: it["item_id"] for it in current}
|
|
cur_vids = [it["video_id"] for it in current]
|
|
desired_set = set(desired)
|
|
# Delete extras (local is authoritative).
|
|
for it in current:
|
|
if it["video_id"] not in desired_set:
|
|
try:
|
|
yt.delete_playlist_item(it["item_id"])
|
|
deleted += 1
|
|
except YouTubeError:
|
|
failures.append(it["video_id"])
|
|
# Insert missing (append; the reorder pass below fixes positions).
|
|
cur_set = set(cur_vids)
|
|
for vid in desired:
|
|
if vid not in cur_set:
|
|
try:
|
|
item_id_by_vid[vid] = yt.add_playlist_item(pl.yt_playlist_id, vid)
|
|
inserted += 1
|
|
except YouTubeError:
|
|
failures.append(vid)
|
|
# Reorder to match `desired` via insertion-sort moves over a local model.
|
|
model = [v for v in cur_vids if v in desired_set] + [
|
|
v for v in desired if v not in cur_set and v in item_id_by_vid
|
|
]
|
|
for i, want in enumerate(desired):
|
|
if want not in item_id_by_vid:
|
|
continue # failed insert; skip
|
|
if i < len(model) and model[i] == want:
|
|
continue
|
|
try:
|
|
yt.move_playlist_item(item_id_by_vid[want], pl.yt_playlist_id, want, i)
|
|
reordered += 1
|
|
except YouTubeError:
|
|
failures.append(want)
|
|
continue
|
|
j = model.index(want)
|
|
model.pop(j)
|
|
model.insert(i, want)
|
|
if not failures:
|
|
pl.synced_fingerprint = fingerprint(pl.name, desired)
|
|
pl.dirty = False
|
|
db.commit()
|
|
return {
|
|
"created": created,
|
|
"inserted": inserted,
|
|
"deleted": deleted,
|
|
"reordered": reordered,
|
|
"failures": failures,
|
|
"yt_playlist_id": pl.yt_playlist_id,
|
|
}
|
|
|
|
|
|
def repull_playlist(db: Session, user: User, pl: Playlist) -> dict:
|
|
"""Force-refresh ONE YouTube-linked playlist from YouTube, discarding local edits and
|
|
clearing dirty. This is the per-playlist 'reset to YouTube' — unlike the bulk read-sync
|
|
it deliberately does NOT skip a dirty playlist (overwriting local changes is the point).
|
|
Works for both mirrors (source='youtube') and exported local playlists (yt_playlist_id)."""
|
|
if not pl.yt_playlist_id:
|
|
raise YouTubeError("Playlist is not linked to YouTube")
|
|
with YouTubeClient(db, user) as yt:
|
|
snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
|
|
if snippet is None:
|
|
raise YouTubeError("Playlist no longer exists on YouTube")
|
|
ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id))
|
|
_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)
|
|
pl.name = snippet.get("title") or pl.name
|
|
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))
|
|
pl.synced_fingerprint = fingerprint(pl.name, ordered)
|
|
pl.dirty = False
|
|
db.commit()
|
|
return {"items": len(ordered)}
|
|
|
|
|
|
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}
|