peter
c0db80d333
- the sweep guard anchors the rating_key to digits: the round-2 widening had degenerated to "anything ending in _<digits>" (backup_2024, release_10) - RSS counts apply-side failures too, so a read-only database no longer reports a wholly failed poll as "ok — new=0, failed=0" - the player reads/writes its volume directly (readAccount/writeAccount): the persisted-object hook rendered nothing and cost a second write per settle - new tests: the sweep guard's accept/reject table, and run_ffmpeg's threading (drain, cancel while silent, stall watchdog, stderr tail on a nonzero exit) - the breaker sentinel is checked one way (isinstance) at all four sites - run_rss_poll is typed dict[str, int] All three new test groups were mutation-checked: reverting each fix turns them red (or, for the cancel/stall pair, hangs — which is the bug itself).
476 lines
19 KiB
Python
476 lines
19 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 collections.abc import Callable
|
|
from datetime import datetime, timezone
|
|
from typing import TypeVar
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import progress, quota
|
|
from app.auth import has_read_scope
|
|
from app.models import Channel, Playlist, PlaylistItem, User, Video
|
|
from app.sync.runner import token_users
|
|
from app.sync.videos import apply_video_details, parse_dt
|
|
from app.youtube.client import YouTubeClient, YouTubeError
|
|
|
|
log = logging.getLogger("siftlode.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 _replace_items_from_youtube(
|
|
db: Session, yt: YouTubeClient, pl: Playlist, ids: list[str]
|
|
) -> list[str]:
|
|
"""Mirror a YouTube playlist's ordered video ids into `pl`'s local items (authoritative
|
|
replace): ensure the videos exist in the catalog, keep only those present in their YT
|
|
order (de-duplicated), swap the items out, and reset the synced fingerprint + dirty flag.
|
|
`pl.name` must already be set to the desired name. The caller commits. Returns the final
|
|
ordered ids."""
|
|
_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)
|
|
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
|
|
return ordered
|
|
|
|
|
|
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
|
|
# Mirror is authoritative: replace the items to match YouTube's order.
|
|
ids = list(yt.iter_my_playlist_video_ids(ytid))
|
|
_replace_items_from_youtube(db, yt, pl, ids)
|
|
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, *, live=None) -> 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. `live` = a pre-fetched
|
|
(current_items, snippet) tuple so the caller can share one read with push_playlist."""
|
|
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,
|
|
}
|
|
if live is not None:
|
|
current, snippet = live
|
|
else:
|
|
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,
|
|
}
|
|
|
|
|
|
# After this many consecutive write failures we stop calling YouTube for the rest of the push.
|
|
# The failure modes that matter here are all account-wide, not per-video: quota exhausted, the
|
|
# write scope revoked, the token dead. Every remaining write then fails identically — but each
|
|
# one still SPENDS 50 units, so a long playlist burns through the whole (already dead) budget,
|
|
# and tomorrow's too, to produce a failure list we knew after the fifth item.
|
|
_MAX_CONSECUTIVE_FAILURES = 5
|
|
|
|
|
|
class _Failed:
|
|
"""Type of the `FAILED` sentinel, so `run()` can stay generic in the op's return type: a call
|
|
site gets back `str | _Failed` (not a bare `object`) and keeps its item id typed."""
|
|
|
|
|
|
#: Returned by `_Breaker.run` when the write did not happen (refused or failed).
|
|
FAILED = _Failed()
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class _Breaker:
|
|
"""Consecutive-failure circuit breaker for a single push run. Any success re-arms it.
|
|
|
|
Drive it with `run()` — it owns the whole open-check / call / record-outcome sequence, so a
|
|
write site can't accidentally implement two thirds of the protocol (forgetting to record a
|
|
failure would silently disable the breaker for that operation)."""
|
|
|
|
def __init__(self, limit: int = _MAX_CONSECUTIVE_FAILURES) -> None:
|
|
self.limit = limit
|
|
self.streak = 0
|
|
self.tripped = False
|
|
|
|
def open(self) -> bool:
|
|
"""True while writes may still be attempted."""
|
|
return not self.tripped
|
|
|
|
def hit(self, ok: bool) -> None:
|
|
if ok:
|
|
self.streak = 0
|
|
return
|
|
self.streak += 1
|
|
if self.streak >= self.limit and not self.tripped:
|
|
self.tripped = True
|
|
log.warning(
|
|
"playlist push: %d consecutive YouTube write failures — stopping this run "
|
|
"(quota, scope or token); the remaining items are reported as failures",
|
|
self.limit,
|
|
)
|
|
|
|
def run(self, op: Callable[[], T], on_fail: Callable[[], None]) -> T | _Failed:
|
|
"""Attempt one YouTube write. Returns the op's result, or `FAILED` when the breaker is
|
|
already tripped (op not called) or the op raised YouTubeError. `on_fail` records the item
|
|
in the caller's failure list either way."""
|
|
if self.tripped:
|
|
on_fail()
|
|
return FAILED
|
|
try:
|
|
out = op()
|
|
except YouTubeError:
|
|
on_fail()
|
|
self.hit(False)
|
|
return FAILED
|
|
self.hit(True)
|
|
return out
|
|
|
|
|
|
def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> 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. `live` = a pre-fetched
|
|
(current_items, snippet) tuple (from plan_push's read) to avoid a second live page-through."""
|
|
desired = _local_video_ids(db, pl)
|
|
inserted = deleted = reordered = created = 0
|
|
failures: list[str] = []
|
|
breaker = _Breaker()
|
|
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:
|
|
added = breaker.run(
|
|
lambda v=vid: yt.add_playlist_item(pl.yt_playlist_id, v),
|
|
lambda v=vid: failures.append(v),
|
|
)
|
|
if not isinstance(added, _Failed):
|
|
inserted += 1
|
|
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,
|
|
"aborted": breaker.tripped,
|
|
"yt_playlist_id": pl.yt_playlist_id,
|
|
}
|
|
|
|
# Existing link: diff against the live YouTube state (reuse plan_push's read if given).
|
|
if live is not None:
|
|
current, snippet = live
|
|
else:
|
|
snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
|
|
current = list(yt.iter_playlist_items_with_ids(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")
|
|
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:
|
|
dropped = breaker.run(
|
|
lambda i=it: yt.delete_playlist_item(i["item_id"]),
|
|
lambda i=it: failures.append(i["video_id"]),
|
|
)
|
|
if not isinstance(dropped, _Failed):
|
|
deleted += 1
|
|
# Insert missing (append; the reorder pass below fixes positions).
|
|
cur_set = set(cur_vids)
|
|
for vid in desired:
|
|
if vid not in cur_set:
|
|
item_id = breaker.run(
|
|
lambda v=vid: yt.add_playlist_item(pl.yt_playlist_id, v),
|
|
lambda v=vid: failures.append(v),
|
|
)
|
|
if not isinstance(item_id, _Failed):
|
|
item_id_by_vid[vid] = item_id
|
|
inserted += 1
|
|
# 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
|
|
moved = breaker.run(
|
|
lambda w=want, pos=i: yt.move_playlist_item(
|
|
item_id_by_vid[w], pl.yt_playlist_id, w, pos
|
|
),
|
|
lambda w=want: failures.append(w),
|
|
)
|
|
if isinstance(moved, _Failed):
|
|
continue
|
|
reordered += 1
|
|
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,
|
|
"aborted": breaker.tripped,
|
|
"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")
|
|
pl.name = snippet.get("title") or pl.name
|
|
ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id))
|
|
ordered = _replace_items_from_youtube(db, yt, pl, ids)
|
|
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 = token_users(db)
|
|
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}
|