fix(sync): an empty answer the remote's own count contradicts is not believed, one level down

E32 review round 1 (B1, B2, B3, B8).

- Plex watch reconcile: a section the server SAYS is empty (totalSize 0, nothing listed) is a
  complete listing. It read as "cut short" forever, so one empty library stopped un-watches from
  ever syncing. A listing with no total at all stays incomplete (the client now returns None for
  it instead of 0), and a whole scan that listed nothing still clears nothing. Tests now cover
  completeness across two sections, both orders.
- Plex collections: no members from a collection whose listing counts some (childCount) keeps
  the membership we hold, in the full sync and in the targeted resync after a write.
- YouTube playlists: no items for a playlist whose listing counts some (itemCount) keeps the
  mirror in the bulk pull, and is refused as retryable in "reset to YouTube" and in the push
  plan/merge, where it read as "YouTube deleted them all". The rule sits beside vanished().
- A playlist deleted on YouTube answered "reconnect your account"; it now says it is gone
  (errors.youtube.gone, HU+EN).
- The pure 3-way merge moved to sync/playlist_merge.py; playlists.py is back under 800 lines and
  off the size list.
This commit is contained in:
2026-10-04 02:13:53 +02:00
parent 5d10a521a8
commit df4a79a1f1
15 changed files with 381 additions and 127 deletions
+5 -3
View File
@@ -78,9 +78,10 @@ class PlexClient:
# --- Catalog mirror (P1) --- # --- Catalog mirror (P1) ---
def section_items( def section_items(
self, key: str, item_type: int, start: int = 0, size: int = 200 self, key: str, item_type: int, start: int = 0, size: int = 200
) -> tuple[list[dict], int]: ) -> tuple[list[dict], int | None]:
"""A page of a section's top-level items. type 1 = movie, 2 = show. Returns """A page of a section's top-level items. type 1 = movie, 2 = show. Returns
(items, total_size).""" (items, total_size); the total is None when the answer carries none, so a container that
says "0" (a genuinely empty section) stays distinguishable from one that says nothing."""
mc = self._get( mc = self._get(
f"/library/sections/{key}/all", f"/library/sections/{key}/all",
params={ params={
@@ -93,7 +94,8 @@ class PlexClient:
"X-Plex-Container-Size": size, "X-Plex-Container-Size": size,
}, },
) )
return mc.get("Metadata", []) or [], int(mc.get("totalSize", mc.get("size", 0)) or 0) total = mc.get("totalSize", mc.get("size"))
return mc.get("Metadata", []) or [], None if total is None else int(total)
def children(self, rating_key: str) -> list[dict]: def children(self, rating_key: str) -> list[dict]:
"""Direct children (show → seasons, season → episodes).""" """Direct children (show → seasons, season → episodes)."""
+43 -10
View File
@@ -418,7 +418,10 @@ def _paginate(plex: PlexClient, key: str, item_type: int, progress: dict | None
if progress is not None: if progress is not None:
# A truthy `total` we reached = a trustworthy complete listing. `total` unknown/zero-with-rows # A truthy `total` we reached = a trustworthy complete listing. `total` unknown/zero-with-rows
# (a Plex that doesn't report totalSize) is treated as NOT complete, so prune stays conservative. # (a Plex that doesn't report totalSize) is treated as NOT complete, so prune stays conservative.
# `total` itself is kept for a caller to whom a server that SAID 0 is news it can trust (the
# watch reconcile); an empty mirror or overlay must not read it so (E32 S2).
progress["complete"] = bool(total) and start >= total progress["complete"] = bool(total) and start >= total
progress["total"] = total
# The section listing exposes at most the top ~20 cast (Plex's own cap); a stored cast longer than this # The section listing exposes at most the top ~20 cast (Plex's own cap); a stored cast longer than this
@@ -671,9 +674,11 @@ def resync_collection(db: Session, plex: PlexClient, lib: PlexLibrary, rk: str)
meta = plex.metadata(rk) or {} meta = plex.metadata(rk) or {}
col = _upsert_collection(db, by_rating_key(db, PlexCollection, rk), rk, lib.id, meta) col = _upsert_collection(db, by_rating_key(db, PlexCollection, rk), rk, lib.id, meta)
Model = PlexItem if lib.kind == "movie" else PlexShow Model = PlexItem if lib.kind == "movie" else PlexShow
new_members = { children = _believed_children(plex, rk, meta)
str(c.get("ratingKey")) for c in plex.collection_children(rk) if c.get("ratingKey") if children is None:
} db.flush()
return col
new_members = {str(c.get("ratingKey")) for c in children if c.get("ratingKey")}
current = db.execute(select(Model).where(Model.collection_keys.contains([rk]))).scalars().all() current = db.execute(select(Model).where(Model.collection_keys.contains([rk]))).scalars().all()
for row in current: # drop rk from rows that are no longer members for row in current: # drop rk from rows that are no longer members
if row.rating_key not in new_members: if row.rating_key not in new_members:
@@ -686,6 +691,22 @@ def resync_collection(db: Session, plex: PlexClient, lib: PlexLibrary, rk: str)
return col return col
def _believed_children(plex: PlexClient, rk: str, meta: dict) -> list[dict] | None:
"""A collection's members, or None when the answer is not believed: no members from a collection
whose own listing counts some (`childCount`) is a glitch, not a collection emptied — the same
empty-answer rule as the collection listing's (E32 review). Rebuilding on it showed the
collection empty until the next sync."""
children = plex.collection_children(rk)
count = str(meta.get("childCount") or "0")
if children or not count.isdigit() or int(count) == 0:
return children
log.warning(
"Plex listed no members for a non-empty collection; keeping its membership",
extra=logs.event("plex.sync.collection_children_empty", collection=rk),
)
return None
def delete_collection_local(db: Session, lib: PlexLibrary, rk: str) -> None: def delete_collection_local(db: Session, lib: PlexLibrary, rk: str) -> None:
"""Local cleanup after a collection is deleted from Plex: strip rk from members + drop the row.""" """Local cleanup after a collection is deleted from Plex: strip rk from members + drop the row."""
Model = PlexItem if lib.kind == "movie" else PlexShow Model = PlexItem if lib.kind == "movie" else PlexShow
@@ -697,6 +718,19 @@ def delete_collection_local(db: Session, lib: PlexLibrary, rk: str) -> None:
db.flush() db.flush()
def _rebuild_membership(
db: Session, lib: PlexLibrary, membership: dict[str, list[str]], kept: set[str]
) -> None:
"""Rebuild collection_keys on every member row of this library (clear + set), carrying over the
keys of a collection whose empty answer was not believed (`_believed_children`)."""
for Model in (PlexItem, PlexShow):
for row in db.execute(select(Model).where(Model.library_id == lib.id)).scalars():
held = [k for k in (row.collection_keys or []) if k in kept]
keys = [*membership.get(row.rating_key, []), *held]
row.collection_keys = list(dict.fromkeys(keys)) or None
db.flush()
def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None: def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
"""Mirror a library's collections + membership. Reads never touch Plex afterwards. Membership is """Mirror a library's collections + membership. Reads never touch Plex afterwards. Membership is
written as `collection_keys` on member movies (plex_items) / shows (plex_shows). A full reconcile: written as `collection_keys` on member movies (plex_items) / shows (plex_shows). A full reconcile:
@@ -718,6 +752,7 @@ def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: di
) )
return return
seen: set[str] = set() seen: set[str] = set()
kept: set[str] = set() # collections whose empty children answer is not believed
membership: dict[str, list[str]] = {} # member rating_key -> [collection rating_key, ...] membership: dict[str, list[str]] = {} # member rating_key -> [collection rating_key, ...]
for meta in cols: for meta in cols:
rk = str(meta.get("ratingKey") or "") rk = str(meta.get("ratingKey") or "")
@@ -725,7 +760,10 @@ def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: di
continue continue
_upsert_collection(db, existing.get(rk), rk, lib.id, meta) _upsert_collection(db, existing.get(rk), rk, lib.id, meta)
seen.add(rk) seen.add(rk)
for child in plex.collection_children(rk): children = _believed_children(plex, rk, meta)
if children is None:
kept.add(rk)
for child in children or []:
crk = str(child.get("ratingKey") or "") crk = str(child.get("ratingKey") or "")
if crk: if crk:
membership.setdefault(crk, []).append(rk) membership.setdefault(crk, []).append(rk)
@@ -734,9 +772,4 @@ def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: di
for rk, col in existing.items(): for rk, col in existing.items():
if rk not in seen: if rk not in seen:
db.delete(col) db.delete(col)
# Rebuild collection_keys on every member row of this library (clear + set). _rebuild_membership(db, lib, membership, kept)
for row in db.execute(select(PlexItem).where(PlexItem.library_id == lib.id)).scalars():
row.collection_keys = membership.get(row.rating_key) or None
for row in db.execute(select(PlexShow).where(PlexShow.library_id == lib.id)).scalars():
row.collection_keys = membership.get(row.rating_key) or None
db.flush()
+12 -6
View File
@@ -75,9 +75,9 @@ def _scan_plex_states(
) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int, bool]: ) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int, bool]:
"""Scan the enabled movie/show sections and return ``{item_id: (status, pos, watched_at, """Scan the enabled movie/show sections and return ``{item_id: (status, pos, watched_at,
prog_at)}`` for every mirrored leaf Plex holds a watch signal for, the scanned-leaf count, and prog_at)}`` for every mirrored leaf Plex holds a watch signal for, the scanned-leaf count, and
whether EVERY section's walk reached the server's total. Shared by the one-time import and the whether EVERY section's walk reached the server's total (or the server said it holds none).
full reconcile (both need Plex's whole current picture). The last one matters to the reconcile Shared by the one-time import and the full reconcile (both need Plex's whole current picture).
alone: it reads "absent from the scan" as "Plex un-watched it", which a walk cut short by a The last one matters to the reconcile alone: it reads "absent from the scan" as "Plex un-watched it", which a walk cut short by a
Plex hiccup does not mean (see `_paginate`).""" Plex hiccup does not mean (see `_paginate`)."""
item_id_by_rk = _rk_to_id(db) item_id_by_rk = _rk_to_id(db)
wanted = _enabled_section_keys(db) wanted = _enabled_section_keys(db)
@@ -93,6 +93,7 @@ def _scan_plex_states(
continue continue
item_type = _TYPE_MOVIE if stype == "movie" else _TYPE_EPISODE item_type = _TYPE_MOVIE if stype == "movie" else _TYPE_EPISODE
progress: dict = {} progress: dict = {}
before = scanned
for meta in _paginate(plex, key, item_type, progress): for meta in _paginate(plex, key, item_type, progress):
scanned += 1 scanned += 1
item_id = item_id_by_rk.get(str(meta.get("ratingKey") or "")) item_id = item_id_by_rk.get(str(meta.get("ratingKey") or ""))
@@ -103,7 +104,10 @@ def _scan_plex_states(
) )
if res is not None: if res is not None:
out[item_id] = res out[item_id] = res
complete = complete and bool(progress.get("complete")) # A section the server SAYS is empty (total 0, nothing listed) is complete too: without this
# one genuinely empty library kept every run "incomplete" and un-watches never synced again.
said_empty = progress.get("total") == 0 and scanned == before
complete = complete and (bool(progress.get("complete")) or said_empty)
return out, scanned, complete return out, scanned, complete
@@ -637,12 +641,14 @@ def run_plex_watch_reconcile(db: Session) -> dict:
if ch: if ch:
total["applied"] += 1 total["applied"] += 1
if complete: # A whole scan that listed nothing at all is not "Plex un-watched everything" either:
# every section can say 0 only when the server lost its libraries, never as news.
if complete and scanned:
total["cleared"] += _clear_unwatched(db, link.user_id, plex_states) total["cleared"] += _clear_unwatched(db, link.user_id, plex_states)
else: else:
total["incomplete"] = total.get("incomplete", 0) + 1 total["incomplete"] = total.get("incomplete", 0) + 1
log.warning( log.warning(
"the Plex scan ended before the server's total; nothing is cleared this run", "the Plex scan was cut short or listed nothing; nothing is cleared this run",
extra=logs.event("plex.watch.reconcile_incomplete", user_id=link.user_id), extra=logs.event("plex.watch.reconcile_incomplete", user_id=link.user_id),
) )
+89
View File
@@ -0,0 +1,89 @@
"""The 3-way merge of a YouTube-linked playlist, as a pure function: what each side gains and loses
so neither loses an item, the merged order, and the reorder moves YouTube needs to reach it. No I/O —
`sync/playlists.py` reads both sides, calls `merge_plan`, and writes the outcome."""
from dataclasses import dataclass
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
@dataclass
class MergePlan:
"""The 3-way merge of a linked playlist: what each side gains and loses so neither side
loses an item, plus the merged order. `final` is the id set both sides end up with."""
final: list[str] # merged order (local order, YouTube-only adds appended)
add_to_yt: list[str] # local-adds to push out to YouTube
add_to_local: list[str] # YouTube-adds to pull into Siftlode
remove_from_yt: list[str] # items deleted locally that are still on YouTube
remove_from_local: list[str] # items deleted on YouTube that are still local
reorders: int # estimated YouTube playlistItems.update moves to reach `final`
baseline_known: bool
def merge_plan(
baseline: list[str] | None,
local: list[str],
youtube: list[str],
unavailable: set[str] | None = None,
) -> MergePlan:
"""Pure 3-way merge. Classify each id against the last-synced `baseline`:
- in local but not youtube, not deleted → add to YouTube (a local-add)
- in youtube but not local, not deleted → add to Siftlode (a YouTube-add)
- in baseline but gone from local → a genuine local delete → propagate to YouTube
- in baseline but gone from youtube → a genuine YouTube delete → propagate to Siftlode
With NO baseline (first-ever merge) we can't tell an add from a delete, so we UNION both
sides (never delete) — the lossless default. Merged order: local order first, YouTube-only
additions appended in their YouTube order.
`unavailable` = known un-ingestable-but-live YouTube ids: dropped from the YouTube side up
front so they're never re-classified as a YouTube-add (they can't ingest) nor as a local
delete (they were never local) — otherwise such a video would make the playlist never
converge. They stay in place on YouTube, untouched by this merge."""
un = set(unavailable or ())
L = list(dict.fromkeys(local))
Y = [v for v in dict.fromkeys(youtube) if v not in un]
Lset, Yset = set(L), set(Y)
if baseline is None:
add_to_yt = [v for v in L if v not in Yset]
add_to_local = [v for v in Y if v not in Lset]
remove_from_yt: list[str] = []
remove_from_local: list[str] = []
final = L + [v for v in Y if v not in Lset]
else:
B = set(baseline)
deleted_local = B - Lset # was synced, removed on our side
deleted_yt = B - Yset # was synced, removed on YouTube's side
gone = deleted_local | deleted_yt
remove_from_yt = [v for v in Y if v in deleted_local]
remove_from_local = [v for v in L if v in deleted_yt]
add_to_yt = [v for v in L if v not in Yset and v not in gone]
add_to_local = [v for v in Y if v not in Lset and v not in gone]
final = [v for v in L if v not in gone] + [v for v in Y if v not in Lset and v not in gone]
final_set = set(final)
# On YouTube, after membership reconciliation the order is current-kept + inserts appended;
# the reorder pass then moves items into `final`.
after_membership = [v for v in Y if v in final_set] + add_to_yt
reorders = _reorder_moves(after_membership, final)
return MergePlan(
final=final,
add_to_yt=add_to_yt,
add_to_local=add_to_local,
remove_from_yt=remove_from_yt,
remove_from_local=remove_from_local,
reorders=reorders,
baseline_known=baseline is not None,
)
+15 -88
View File
@@ -1,6 +1,7 @@
"""YouTube playlist sync, both directions. Read: mirror each user's own YouTube playlists """YouTube playlist sync, both directions. Read: mirror each user's own YouTube playlists
into local `source='youtube'` playlists (kept fresh by the scheduler). Write-back: two-way sync into local `source='youtube'` playlists (kept fresh by the scheduler). Write-back: two-way sync
local edits with YouTube (`plan_merge` / `merge_playlist`), behind the optional write scope. local edits with YouTube (`plan_merge` / `merge_playlist`), behind the optional write scope; the
pure 3-way merge they run is `sync/playlist_merge.py`.
YouTube's special Watch Later / History playlists are not exposed by the Data API and are 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 never synced. Videos in a playlist that aren't in our shared catalog yet are fetched and
@@ -10,7 +11,7 @@ import hashlib
import json import json
from collections.abc import Callable, Iterator from collections.abc import Callable, Iterator
from contextlib import AbstractContextManager, contextmanager from contextlib import AbstractContextManager, contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import TypeVar from typing import TypeVar
@@ -21,8 +22,9 @@ from sqlalchemy.pool import NullPool
from app import logs, progress, quota from app import logs, progress, quota
from app.auth import has_read_scope from app.auth import has_read_scope
from app.models import Channel, Playlist, PlaylistItem, User, Video from app.models import Channel, Playlist, PlaylistItem, User, Video
from app.sync.playlist_merge import merge_plan
from app.sync.runner import token_users from app.sync.runner import token_users
from app.sync.subscriptions import vanished from app.sync.subscriptions import blank_listing, refuse_blank_listing, vanished
from app.sync.videos import apply_video_details, parse_dt from app.sync.videos import apply_video_details, parse_dt
from app.youtube.client import YouTubeClient, YouTubeError from app.youtube.client import YouTubeClient, YouTubeError
@@ -376,6 +378,9 @@ def pull_user_playlists(db: Session, user: User) -> dict:
# The next scheduled run refreshes it. # The next scheduled run refreshes it.
continue continue
ids = list(yt.iter_my_playlist_video_ids(ytid)) ids = list(yt.iter_my_playlist_video_ids(ytid))
if blank_listing(ids, p.get("item_count")):
log.warning("YouTube listed no items for non-empty playlist %s; kept", pl.id)
continue
# Ingest first — it commits — so the row lock below is held from the dirty check # Ingest first — it commits — so the row lock below is held from the dirty check
# to the commit that replaces the items. # to the commit that replaces the items.
_ensure_videos(db, yt, ids) _ensure_videos(db, yt, ids)
@@ -407,90 +412,6 @@ def _local_video_ids(db: Session, pl: Playlist) -> list[str]:
) )
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
@dataclass
class MergePlan:
"""The 3-way merge of a linked playlist: what each side gains and loses so neither side
loses an item, plus the merged order. `final` is the id set both sides end up with."""
final: list[str] # merged order (local order, YouTube-only adds appended)
add_to_yt: list[str] # local-adds to push out to YouTube
add_to_local: list[str] # YouTube-adds to pull into Siftlode
remove_from_yt: list[str] # items deleted locally that are still on YouTube
remove_from_local: list[str] # items deleted on YouTube that are still local
reorders: int # estimated YouTube playlistItems.update moves to reach `final`
baseline_known: bool
def merge_plan(
baseline: list[str] | None,
local: list[str],
youtube: list[str],
unavailable: set[str] | None = None,
) -> MergePlan:
"""Pure 3-way merge. Classify each id against the last-synced `baseline`:
- in local but not youtube, not deleted → add to YouTube (a local-add)
- in youtube but not local, not deleted → add to Siftlode (a YouTube-add)
- in baseline but gone from local → a genuine local delete → propagate to YouTube
- in baseline but gone from youtube → a genuine YouTube delete → propagate to Siftlode
With NO baseline (first-ever merge) we can't tell an add from a delete, so we UNION both
sides (never delete) — the lossless default. Merged order: local order first, YouTube-only
additions appended in their YouTube order.
`unavailable` = known un-ingestable-but-live YouTube ids: dropped from the YouTube side up
front so they're never re-classified as a YouTube-add (they can't ingest) nor as a local
delete (they were never local) — otherwise such a video would make the playlist never
converge. They stay in place on YouTube, untouched by this merge."""
un = set(unavailable or ())
L = list(dict.fromkeys(local))
Y = [v for v in dict.fromkeys(youtube) if v not in un]
Lset, Yset = set(L), set(Y)
if baseline is None:
add_to_yt = [v for v in L if v not in Yset]
add_to_local = [v for v in Y if v not in Lset]
remove_from_yt: list[str] = []
remove_from_local: list[str] = []
final = L + [v for v in Y if v not in Lset]
else:
B = set(baseline)
deleted_local = B - Lset # was synced, removed on our side
deleted_yt = B - Yset # was synced, removed on YouTube's side
gone = deleted_local | deleted_yt
remove_from_yt = [v for v in Y if v in deleted_local]
remove_from_local = [v for v in L if v in deleted_yt]
add_to_yt = [v for v in L if v not in Yset and v not in gone]
add_to_local = [v for v in Y if v not in Lset and v not in gone]
final = [v for v in L if v not in gone] + [v for v in Y if v not in Lset and v not in gone]
final_set = set(final)
# On YouTube, after membership reconciliation the order is current-kept + inserts appended;
# the reorder pass then moves items into `final`.
after_membership = [v for v in Y if v in final_set] + add_to_yt
reorders = _reorder_moves(after_membership, final)
return MergePlan(
final=final,
add_to_yt=add_to_yt,
add_to_local=add_to_local,
remove_from_yt=remove_from_yt,
remove_from_local=remove_from_local,
reorders=reorders,
baseline_known=baseline is not None,
)
def plan_merge(db: Session, user: User, pl: Playlist, *, live=None) -> dict: def plan_merge(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
"""Compute (without writing) what a two-way sync of this playlist would do: create vs """Compute (without writing) what a two-way sync of this playlist would do: create vs
update, how many items each side gains/loses, reorders, and the quota estimate (YouTube update, how many items each side gains/loses, reorders, and the quota estimate (YouTube
@@ -518,6 +439,7 @@ def plan_merge(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
snippet = yt.get_playlist_snippet(pl.yt_playlist_id) snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
rename = bool(snippet) and (snippet.get("title") or "") != pl.name rename = bool(snippet) and (snippet.get("title") or "") != pl.name
yt_ids = [it["video_id"] for it in current] yt_ids = [it["video_id"] for it in current]
refuse_blank_listing(yt_ids, (snippet or {}).get("item_count"))
mp = merge_plan(parse_baseline(pl), local, yt_ids, parse_unavailable(pl)) mp = merge_plan(parse_baseline(pl), local, yt_ids, parse_unavailable(pl))
ops = len(mp.add_to_yt) + len(mp.remove_from_yt) + mp.reorders + (1 if rename else 0) ops = len(mp.add_to_yt) + len(mp.remove_from_yt) + mp.reorders + (1 if rename else 0)
return { return {
@@ -664,6 +586,8 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
current = list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id)) 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} item_id_by_vid = {it["video_id"]: it["item_id"] for it in current}
yt_ids = [it["video_id"] for it in current] yt_ids = [it["video_id"] for it in current]
# A merge would read "YouTube lists none of them" as "YouTube deleted them all".
refuse_blank_listing(yt_ids, (snippet or {}).get("item_count"))
# Re-attempt ingestion of ids previously marked un-ingestable that are still on YouTube: a # Re-attempt ingestion of ids previously marked un-ingestable that are still on YouTube: a
# video that has since become available (e.g. private→public) should re-enter the merge, not # video that has since become available (e.g. private→public) should re-enter the merge, not
# stay skipped forever. Those that still won't ingest remain filtered (`effective_unavailable`). # stay skipped forever. Those that still won't ingest remain filtered (`effective_unavailable`).
@@ -802,8 +726,11 @@ def repull_playlist(db: Session, user: User, pl: Playlist) -> dict:
with YouTubeClient(db, user) as yt: with YouTubeClient(db, user) as yt:
snippet = yt.get_playlist_snippet(pl.yt_playlist_id) snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
if snippet is None: if snippet is None:
raise YouTubeError("Playlist no longer exists on YouTube") raise YouTubeError(
"Playlist no longer exists on YouTube", reason="playlistNotFound", path="playlists"
)
ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id)) ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id))
refuse_blank_listing(ids, snippet.get("item_count"))
_ensure_videos(db, yt, ids) _ensure_videos(db, yt, ids)
_lock_or_gone(db, pl) _lock_or_gone(db, pl)
pl.name = snippet.get("title") or pl.name pl.name = snippet.get("title") or pl.name
+20 -2
View File
@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
from app import logs from app import logs
from app.models import Channel, Subscription, User from app.models import Channel, Subscription, User
from app.utils import normalize_description from app.utils import normalize_description
from app.youtube.client import YouTubeClient, best_thumbnail from app.youtube.client import YouTubeClient, YouTubeError, best_thumbnail
log = logs.get_logger(__name__) log = logs.get_logger(__name__)
@@ -115,7 +115,8 @@ def should_prune(fetched_count: int, existing_count: int) -> bool:
has — taking each row's `deep_requested` flag with it — in a single run. Someone who genuinely has — taking each row's `deep_requested` flag with it — in a single run. Someone who genuinely
unsubscribed from everything simply keeps their rows until a sync that returns something. unsubscribed from everything simply keeps their rows until a sync that returns something.
Pure so it can be tested without a database; `import_subscriptions` is the only caller.""" Pure so it can be tested without a database; its callers are `import_subscriptions` and, through
`vanished`, every mirror that drops what the remote no longer lists."""
return fetched_count > 0 or existing_count == 0 return fetched_count > 0 or existing_count == 0
@@ -128,6 +129,23 @@ def vanished[K, V](existing: dict[K, V], listed: Collection[K]) -> list[V]:
return [row for key, row in existing.items() if key not in listed] return [row for key, row in existing.items() if key not in listed]
def blank_listing(listed: Collection, count: object) -> bool:
"""The same rule one level down: no members from a container whose own listing counts some (a
playlist's `itemCount`) is a blip, not a list emptied. Replacing or merging on it deleted every
mirrored item and reset the merge baseline (E32 review)."""
return not listed and isinstance(count, int) and count > 0
def refuse_blank_listing(listed: Collection, count: object) -> None:
"""`blank_listing` for a user-started pull or push: refused as retryable ("try again")."""
if blank_listing(listed, count):
raise YouTubeError(
"YouTube listed no items for a playlist it counts as non-empty",
reason="emptyAnswer",
path="playlistItems",
)
def import_subscriptions(db: Session, user: User) -> dict: def import_subscriptions(db: Session, user: User) -> dict:
"""Pull all subscriptions for `user`, upsert channels + subscription links, prune """Pull all subscriptions for `user`, upsert channels + subscription links, prune
channels the user has unsubscribed from on YouTube, and fetch channel details for channels the user has unsubscribed from on YouTube, and fetch channel details for
+14 -5
View File
@@ -114,7 +114,7 @@ _DEAD_GRANT = "invalid_grant"
# What a refused YouTube action means to the person who asked for it — the three things they can # What a refused YouTube action means to the person who asked for it — the three things they can
# act on differently: wait for tomorrow's budget, reconnect the account, or try again later. # act on differently: wait for tomorrow's budget, reconnect the account, or try again later.
YouTubeRefusal = Literal["quota", "reconnect", "unavailable"] YouTubeRefusal = Literal["quota", "reconnect", "unavailable", "gone"]
def refusal_reason(exc: YouTubeError) -> YouTubeRefusal: def refusal_reason(exc: YouTubeError) -> YouTubeRefusal:
@@ -126,6 +126,9 @@ def refusal_reason(exc: YouTubeError) -> YouTubeRefusal:
the account would still read as connected, with no connect button to press.""" the account would still read as connected, with no connect button to press."""
if is_quota_refusal(exc): if is_quota_refusal(exc):
return "quota" return "quota"
if exc.reason == "playlistNotFound":
# The list itself is gone (deleted on YouTube): neither a retry nor a reconnect helps.
return "gone"
if exc.path == "oauth2/token": if exc.path == "oauth2/token":
return "reconnect" if exc.reason == _DEAD_GRANT else "unavailable" return "reconnect" if exc.reason == _DEAD_GRANT else "unavailable"
if not exc.path and exc.reason != "transport": if not exc.path and exc.reason != "transport":
@@ -423,16 +426,22 @@ class YouTubeClient:
return resp.json() if resp.content else {} return resp.json() if resp.content else {}
def get_playlist_snippet(self, playlist_id: str) -> dict | None: def get_playlist_snippet(self, playlist_id: str) -> dict | None:
"""The snippet (title/description) of one playlist, or None if it no longer """The snippet (title/description) of one playlist plus its `item_count` (what an empty
exists. 1 unit (OAuth, so private playlists work).""" item listing is checked against), or None if it no longer exists. 1 unit (OAuth, so
private playlists work)."""
data = self._get( data = self._get(
"playlists", "playlists",
{"part": "snippet", "id": playlist_id, "maxResults": 1}, {"part": "snippet,contentDetails", "id": playlist_id, "maxResults": 1},
cost=1, cost=1,
allow_key=False, allow_key=False,
) )
items = data.get("items", []) items = data.get("items", [])
return items[0].get("snippet") if items else None if not items:
return None
return {
**(items[0].get("snippet") or {}),
"item_count": items[0].get("contentDetails", {}).get("itemCount"),
}
def update_playlist_title(self, playlist_id: str, title: str) -> None: def update_playlist_title(self, playlist_id: str, title: str) -> None:
"""Rename a playlist on YouTube. 50 units.""" """Rename a playlist on YouTube. 50 units."""
+124 -3
View File
@@ -4,14 +4,26 @@ Every mirror here reconciles: what the remote no longer lists is removed locally
(`_prune_stale`) and the subscriptions (`should_prune`) had long refused to do that on an empty (`_prune_stale`) and the subscriptions (`should_prune`) had long refused to do that on an empty
listing that contradicts what we hold — a 200 with no items is a blip far more often than a user listing that contradicts what we hold — a 200 with no items is a blip far more often than a user
who removed everything. Two mirrors did not: the Plex collections (a row carries the user-set who removed everything. Two mirrors did not: the Plex collections (a row carries the user-set
`editable` flag, and every member its keys) and the YouTube playlist mirrors. These hold both. `editable` flag, and every member its keys) and the YouTube playlist mirrors. These hold both — and,
one level down (E32 review), a collection's members and a playlist's items: an empty answer that
contradicts the remote's own count is not believed either.
""" """
import pytest
from sqlalchemy import select from sqlalchemy import select
from app.models import Playlist, PlexCollection, PlexItem, PlexLibrary from app.models import (
Channel,
Playlist,
PlaylistItem,
PlexCollection,
PlexItem,
PlexLibrary,
Video,
)
from app.plex import sync as plex_sync from app.plex import sync as plex_sync
from app.sync import playlists as plsync from app.sync import playlists as plsync
from app.youtube.client import YouTubeError
class _Plex: class _Plex:
@@ -25,6 +37,9 @@ class _Plex:
def collection_children(self, rk): def collection_children(self, rk):
return self._children.get(rk, []) return self._children.get(rk, [])
def metadata(self, rk):
return next((c for c in self._collections if c.get("ratingKey") == rk), {})
def _library_with_a_collection(db): def _library_with_a_collection(db):
lib = PlexLibrary(plex_key="4", title="Movies", kind="movie") lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
@@ -62,9 +77,40 @@ def test_a_listing_with_something_in_it_still_prunes_what_it_no_longer_has(db):
assert db.get(PlexItem, movie.id).collection_keys is None assert db.get(PlexItem, movie.id).collection_keys is None
def test_no_members_from_a_collection_plex_counts_as_non_empty_keeps_its_members(db):
lib, movie = _library_with_a_collection(db)
stats = {"collections": 0}
bond = {"ratingKey": "c1", "title": "Bond", "childCount": 2}
plex_sync._sync_collections(db, _Plex([bond], {"c1": []}), lib, stats)
db.commit()
db.expire_all()
assert db.get(PlexItem, movie.id).collection_keys == ["c1"]
def test_a_collection_plex_counts_as_empty_does_lose_its_members(db):
lib, movie = _library_with_a_collection(db)
stats = {"collections": 0}
bond = {"ratingKey": "c1", "title": "Bond", "childCount": 0}
plex_sync._sync_collections(db, _Plex([bond], {"c1": []}), lib, stats)
db.commit()
db.expire_all()
assert db.get(PlexItem, movie.id).collection_keys is None
@pytest.mark.parametrize(("count", "kept"), [(2, ["c1"]), (0, None)])
def test_a_collection_resync_believes_an_empty_member_list_only_with_a_zero_count(db, count, kept):
lib, movie = _library_with_a_collection(db)
bond = {"ratingKey": "c1", "title": "Bond", "childCount": count}
plex_sync.resync_collection(db, _Plex([bond], {"c1": []}), lib, "c1")
db.commit()
db.expire_all()
assert db.get(PlexItem, movie.id).collection_keys == kept
class _YT: class _YT:
def __init__(self, playlists: list[dict]): def __init__(self, playlists: list[dict], snippet: dict | None = None):
self._playlists = playlists self._playlists = playlists
self._snippet = snippet
def __call__(self, *_): def __call__(self, *_):
return self return self
@@ -84,6 +130,12 @@ class _YT:
def iter_my_playlist_video_ids(self, _ytid): def iter_my_playlist_video_ids(self, _ytid):
return [] return []
def get_playlist_snippet(self, _ytid):
return self._snippet
def get_videos(self, _ids):
return []
def test_an_empty_playlist_answer_keeps_the_youtube_mirrors(db, make_user, monkeypatch): def test_an_empty_playlist_answer_keeps_the_youtube_mirrors(db, make_user, monkeypatch):
user = make_user() user = make_user()
@@ -109,3 +161,72 @@ def test_a_playlist_answer_with_something_in_it_still_drops_a_vanished_mirror(
db.execute(select(Playlist.yt_playlist_id).where(Playlist.source == "youtube")).scalars() db.execute(select(Playlist.yt_playlist_id).where(Playlist.source == "youtube")).scalars()
) )
assert "PL1" not in ytids assert "PL1" not in ytids
def _mirror_with_an_item(db, user, **kw):
db.add(Channel(id="UC_pl", title="C"))
db.flush()
db.add(Video(id="v1", channel_id="UC_pl", title="One"))
pl = Playlist(user_id=user.id, name="Mixes", yt_playlist_id="PL1", **kw)
db.add(pl)
db.flush()
db.add(PlaylistItem(playlist_id=pl.id, video_id="v1", position=0))
db.commit()
return pl
def _items(db, pl):
db.expire_all()
return (
db.execute(select(PlaylistItem.video_id).where(PlaylistItem.playlist_id == pl.id))
.scalars()
.all()
)
def test_no_items_from_a_playlist_youtube_counts_as_non_empty_keeps_the_mirror_items(
db, make_user, monkeypatch
):
user = make_user()
pl = _mirror_with_an_item(db, user, source="youtube")
listing = [{"id": "PL1", "title": "Mixes", "item_count": 1}]
monkeypatch.setattr(plsync, "YouTubeClient", _YT(listing))
plsync.pull_user_playlists(db, user)
assert _items(db, pl) == ["v1"]
def test_a_playlist_youtube_counts_as_empty_is_emptied(db, make_user, monkeypatch):
user = make_user()
pl = _mirror_with_an_item(db, user, source="youtube")
listing = [{"id": "PL1", "title": "Mixes", "item_count": 0}]
monkeypatch.setattr(plsync, "YouTubeClient", _YT(listing))
plsync.pull_user_playlists(db, user)
assert _items(db, pl) == []
def test_a_reset_to_youtube_refuses_a_blank_listing_and_keeps_the_items(db, make_user, monkeypatch):
user = make_user()
pl = _mirror_with_an_item(db, user, source="youtube")
monkeypatch.setattr(plsync, "YouTubeClient", _YT([], {"title": "Mixes", "item_count": 1}))
with pytest.raises(YouTubeError) as err:
plsync.repull_playlist(db, user, pl)
db.rollback()
assert err.value.reason == "emptyAnswer"
assert _items(db, pl) == ["v1"]
def test_a_reset_of_a_deleted_playlist_says_it_is_gone(db, make_user, monkeypatch):
user = make_user()
pl = _mirror_with_an_item(db, user, source="youtube")
monkeypatch.setattr(plsync, "YouTubeClient", _YT([], None))
with pytest.raises(YouTubeError) as err:
plsync.repull_playlist(db, user, pl)
assert err.value.reason == "playlistNotFound"
def test_a_push_plan_refuses_a_blank_listing_rather_than_reading_it_as_all_deleted(db, make_user):
user = make_user()
pl = _mirror_with_an_item(db, user, source="local")
with pytest.raises(YouTubeError) as err:
plsync.plan_merge(db, user, pl, live=([], {"title": "Mixes", "item_count": 1}))
assert err.value.reason == "emptyAnswer"
+1 -1
View File
@@ -12,8 +12,8 @@ import pytest
from app.models import Channel, Playlist, PlaylistItem, Video from app.models import Channel, Playlist, PlaylistItem, Video
from app.sync import playlists as pl_mod from app.sync import playlists as pl_mod
from app.sync.playlist_merge import merge_plan
from app.sync.playlists import ( from app.sync.playlists import (
merge_plan,
merge_playlist, merge_playlist,
parse_baseline, parse_baseline,
parse_unavailable, parse_unavailable,
+3
View File
@@ -81,6 +81,9 @@ def test_a_section_page_asks_for_guids_and_reports_the_server_total(plex):
plex.routes["/library/sections/7/all"] = {"Metadata": [], "size": 0} plex.routes["/library/sections/7/all"] = {"Metadata": [], "size": 0}
assert plex.section_items("7", 2) == ([], 0) assert plex.section_items("7", 2) == ([], 0)
# No total at all is not "0": the caller must not read it as a complete, empty section.
plex.routes["/library/sections/7/all"] = {"Metadata": []}
assert plex.section_items("7", 2) == ([], None)
def test_related_flattens_the_hubs_and_degrades_to_empty(plex): def test_related_flattens_the_hubs_and_degrades_to_empty(plex):
+48 -6
View File
@@ -13,11 +13,13 @@ from app.plex import watch_sync
class _FakePlex: class _FakePlex:
"""A PlexClient stand-in: one movie section whose listing reports `total` (more than it hands """A PlexClient stand-in: one movie section whose listing reports `total` (more than it hands
over = a walk cut short), and a record of every write the reconcile sends back to Plex.""" over = a walk cut short), plus any `more` sections as `(type, leaves, total)` — listed AFTER
the movies — and a record of every write the reconcile sends back to Plex."""
def __init__(self, leaves: list[dict], total: int | None = None): def __init__(self, leaves: list[dict], total: int | None = None, more=()):
self.leaves = leaves self.listings = {"1": ("movie", leaves, len(leaves) if total is None else total)}
self.total = len(leaves) if total is None else total for i, (stype, extra, extra_total) in enumerate(more, start=2):
self.listings[str(i)] = (stype, extra, extra_total)
self.calls: list[tuple] = [] self.calls: list[tuple] = []
def __enter__(self): def __enter__(self):
@@ -27,10 +29,11 @@ class _FakePlex:
return False return False
def sections(self): def sections(self):
return [{"type": "movie", "key": "1"}] return [{"type": stype, "key": key} for key, (stype, _l, _t) in self.listings.items()]
def section_items(self, key, item_type, start, size): def section_items(self, key, item_type, start, size):
return self.leaves[start : start + size], self.total _stype, leaves, total = self.listings[key]
return leaves[start : start + size], total
def scrobble(self, rk): def scrobble(self, rk):
self.calls.append(("scrobble", rk)) self.calls.append(("scrobble", rk))
@@ -136,6 +139,45 @@ def test_a_scan_cut_short_clears_nothing(db, monkeypatch, world):
assert out["cleared"] == 0 and out["incomplete"] == 1 assert out["cleared"] == 0 and out["incomplete"] == 1
_LISTED = [
{"ratingKey": "m3"},
{"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
{"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
]
def test_a_genuinely_empty_second_section_still_lets_the_reconcile_clear(db, monkeypatch, world):
# The server SAYS the shows section holds 0: that is a complete listing, not a hiccup — before
# the E32 review it read as "incomplete" forever and un-watches never synced again.
state, rows = world
_local_states(state)
db.commit()
out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
assert "m1" not in rows()
assert out["cleared"] == 1 and "incomplete" not in out
def test_a_cut_short_section_is_not_rescued_by_a_complete_one_after_it(db, monkeypatch, world):
# Completeness is every section's, not the last one's: movies cut short (5 said, 3 handed),
# then a complete shows listing — m1 may simply not have been reached.
state, rows = world
_local_states(state)
db.commit()
shows = [{"ratingKey": "s1"}]
out = _run(db, monkeypatch, _FakePlex(_LISTED, total=5, more=[("show", shows, 1)]))
assert rows()["m1"] == ("watched", 0, True)
assert out["cleared"] == 0 and out["incomplete"] == 1
def test_a_listing_without_a_total_is_never_complete(db, monkeypatch, world):
state, rows = world
_local_states(state)
db.commit()
out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], None)]))
assert rows()["m1"] == ("watched", 0, True)
assert out["cleared"] == 0 and out["incomplete"] == 1
def test_an_empty_listing_is_not_a_library_of_unwatched_titles(db, monkeypatch, world): def test_an_empty_listing_is_not_a_library_of_unwatched_titles(db, monkeypatch, world):
state, rows = world state, rows = world
state("m1", status="watched", synced_to_plex=True) state("m1", status="watched", synced_to_plex=True)
-1
View File
@@ -35,7 +35,6 @@ BIG_MODULES: dict[str, int] = {
"app/routes/admin.py": 864, "app/routes/admin.py": 864,
"app/routes/auth.py": 1353, "app/routes/auth.py": 1353,
"app/routes/downloads.py": 1237, "app/routes/downloads.py": 1237,
"app/sync/playlists.py": 843,
} }
COMPLEX: dict[str, list[int]] = { COMPLEX: dict[str, list[int]] = {
+3
View File
@@ -27,6 +27,9 @@ from app.youtube.client import YouTubeError, refusal_reason
(YouTubeError("No refresh token; user must re-authenticate"), "reconnect"), (YouTubeError("No refresh token; user must re-authenticate"), "reconnect"),
(YouTubeError("m", reason="transport"), "unavailable"), (YouTubeError("m", reason="transport"), "unavailable"),
(YouTubeError("m", reason="backendError", path="subscriptions"), "unavailable"), (YouTubeError("m", reason="backendError", path="subscriptions"), "unavailable"),
# A playlist deleted on YouTube: neither a retry nor a reconnect helps (it said "reconnect").
(YouTubeError("m", reason="playlistNotFound", path="playlists"), "gone"),
(YouTubeError("m", reason="emptyAnswer", path="playlistItems"), "unavailable"),
], ],
) )
def test_a_failure_is_sorted_by_what_the_user_can_do(exc, reason): def test_a_failure_is_sorted_by_what_the_user_can_do(exc, reason):
+2 -1
View File
@@ -115,7 +115,8 @@
"youtube": { "youtube": {
"quota": "Today's YouTube budget for this instance is spent. It resets at midnight Pacific time — try again then.", "quota": "Today's YouTube budget for this instance is spent. It resets at midnight Pacific time — try again then.",
"reconnect": "Google no longer accepts this account's YouTube connection. Sign out and sign in with Google again to reconnect it.", "reconnect": "Google no longer accepts this account's YouTube connection. Sign out and sign in with Google again to reconnect it.",
"unavailable": "YouTube didn't accept that just now. Please try again." "unavailable": "YouTube didn't accept that just now. Please try again.",
"gone": "This playlist no longer exists on YouTube"
}, },
"refused": { "refused": {
"account_no_access": "That account no longer has access", "account_no_access": "That account no longer has access",
+2 -1
View File
@@ -115,7 +115,8 @@
"youtube": { "youtube": {
"quota": "Ennek a példánynak elfogyott a mai YouTube-kerete. Csendes-óceáni idő szerint éjfélkor újraindul — akkor próbáld újra.", "quota": "Ennek a példánynak elfogyott a mai YouTube-kerete. Csendes-óceáni idő szerint éjfélkor újraindul — akkor próbáld újra.",
"reconnect": "A Google már nem fogadja el ennek a fióknak a YouTube-kapcsolatát. Jelentkezz ki, majd lépj be újra Google-lel az összekötéshez.", "reconnect": "A Google már nem fogadja el ennek a fióknak a YouTube-kapcsolatát. Jelentkezz ki, majd lépj be újra Google-lel az összekötéshez.",
"unavailable": "A YouTube most nem fogadta el. Próbáld újra." "unavailable": "A YouTube most nem fogadta el. Próbáld újra.",
"gone": "Ez a lejátszási lista már nem létezik a YouTube-on"
}, },
"refused": { "refused": {
"account_no_access": "Ennek a fióknak már nincs hozzáférése", "account_no_access": "Ennek a fióknak már nincs hozzáférése",