b23f805533
The quota-attribution action keys were inconsistent (mixed verbs, ad-hoc names, English-only labels). Introduce a QuotaAction constants holder (one source of truth), rename every attribution call site to it, and add migration 0020 to rename historical quota_events.action values. The display label now resolves from i18n (quotaActions.<key>, EN/HU/DE) instead of a hard-coded English map. Scheduler job ids and progress phase labels are a separate namespace and are left untouched.
399 lines
16 KiB
Python
399 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 progress, 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
|
|
return
|
|
# The session has autoflush off, so the caller's pending item/order/name edits aren't
|
|
# visible to current_fingerprint's query yet — flush first or dirty would lag one edit.
|
|
db.flush()
|
|
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 i, user in enumerate(users, 1):
|
|
if not has_read_scope(user):
|
|
progress.report(i, len(users), "playlist_sync")
|
|
continue
|
|
try:
|
|
with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL):
|
|
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)
|
|
progress.report(i, len(users), "playlist_sync")
|
|
return {"users": len(users), "playlists_synced": total}
|