diff --git a/backend/app/plex/overlay.py b/backend/app/plex/overlay.py new file mode 100644 index 00000000..7676a118 --- /dev/null +++ b/backend/app/plex/overlay.py @@ -0,0 +1,150 @@ +"""The HU overlay: a Hungarian Plex section written onto its English twin's rows (E30), split out of +`plex/sync.py` (E32 S2), which runs it once per paired section during the full sync. + +The pairing is admin-configured (`plex_overlay_map`, ``huKey:enKey,...``). An overlay section creates +no browsable rows of its own; it only fills the `*_hu` columns of the EN rows it matches by guid. +`plex/sync.py` and this module import each other as MODULES (attribute access at call time), which is +what lets either be imported first. +""" + +from collections.abc import Iterable + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app import logs, sysconfig +from app.models import PlexItem, PlexLibrary, PlexShow +from app.plex import sync as plex_sync +from app.plex.client import PlexClient +from app.plex.genres import hu_genres + +log = logs.get_logger(__name__) + + +def overlay_map(db: Session) -> dict[str, str]: + """The configured HU-overlay pairing: {huSectionKey: enSectionKey} from `plex_overlay_map` + ('huKey:enKey,...'). Each HU section is synced as a non-browsable overlay onto its EN twin. + Malformed pairs are skipped.""" + raw = sysconfig.get_str(db, "plex_overlay_map") + out: dict[str, str] = {} + for pair in raw.split(","): + hu, sep, en = pair.partition(":") + hu, en = hu.strip(), en.strip() + if sep and hu and en: + out[hu] = en + return out + + +def sync_overlay( + db: Session, plex: PlexClient, lib: PlexLibrary, en_lib_id: int, stats: dict +) -> None: + """Write an overlay section's Hungarian metadata onto its EN twin's rows, matched by guid. Creates + NO browsable rows — an overlay library is never listed on its own. Only title/summary/tagline/genres + differ by language; the language-neutral people_text feeds both search vectors. A show section's + episodes are overlaid too (E30): the HU section lists every episode with its Hungarian title and + summary, and the episode rows live in the EN show library. + + This is a full RECONCILE: after a COMPLETE HU listing every guid-bearing EN row's HU fields are + cleared, then the matches are repopulated — so an EN row whose HU twin vanished (or was re-guid'd) + falls back to EN instead of keeping stale Hungarian metadata. A listing cut short only adds. HU + items with no EN match are skipped.""" + if lib.kind == "movie": + passes = [ + ( + plex_sync._TYPE_MOVIE, + select(PlexItem).where( + PlexItem.library_id == en_lib_id, + PlexItem.kind == "movie", + PlexItem.guid.is_not(None), + ), + ) + ] + else: + passes = [ + ( + plex_sync._TYPE_SHOW, + select(PlexShow).where( + PlexShow.library_id == en_lib_id, PlexShow.guid.is_not(None) + ), + ), + ( + plex_sync._TYPE_EPISODE, + select(PlexItem).where( + PlexItem.library_id == en_lib_id, + PlexItem.kind == "episode", + PlexItem.guid.is_not(None), + ), + ), + ] + for item_type, rows_q in passes: + _overlay_pass(db, plex, lib, item_type, db.execute(rows_q).scalars(), stats) + db.flush() + + +def _overlay_pass( + db: Session, + plex: PlexClient, + lib: PlexLibrary, + item_type: int, + rows: Iterable[PlexItem] | Iterable[PlexShow], + stats: dict, +) -> None: + """One kind of the overlay (movies, shows or episodes) — see `sync_overlay`.""" + # guid -> the EN rows carrying it (a list: two editions of one film can share a guid; both overlay). + by_guid: dict[str, list[PlexItem | PlexShow]] = {} + for r in rows: + if r.guid: # the passes select guid-bearing rows only + by_guid.setdefault(r.guid, []).append(r) + # The HU listing is read first and applied after, so the clear can wait for its verdict. + hu_by_guid, complete = _read_hu_listing(plex, lib, item_type, by_guid) + episodes = item_type == plex_sync._TYPE_EPISODE + if complete: + _clear_hu(by_guid, episodes) + else: + stats["overlay_incomplete"] = stats.get("overlay_incomplete", 0) + 1 + log.warning( + "the HU listing ended before the server's total; existing HU text is kept", + extra=logs.event("plex.sync.overlay_incomplete", library=lib.plex_key), + ) + for g, meta in hu_by_guid.items(): + targets = by_guid[g] + title_hu = (meta.get("title") or "").strip() or None + summary_hu = meta.get("summary") + tagline_hu = (meta.get("tagline") or "").strip() or None + genres_hu = hu_genres(plex_sync._tags(meta, "Genre")) + for row in targets: + row.title_hu = title_hu + row.summary_hu = summary_hu + if not episodes: + row.tagline_hu = tagline_hu + row.genres_hu = genres_hu or row.genres_hu + stats["overlaid"] = stats.get("overlaid", 0) + 1 + + +def _read_hu_listing( + plex: PlexClient, lib: PlexLibrary, item_type: int, by_guid: dict +) -> tuple[dict[str, dict], bool]: + """The HU section's entries for the guids the EN side carries, and whether the walk reached the + server's total. The first HU entry for a guid wins; a dup HU item can't overwrite or double-count.""" + progress: dict = {} + hu_by_guid: dict[str, dict] = {} + for meta in plex_sync._paginate(plex, lib.plex_key, item_type, progress): + g = plex_sync._canonical_guid(meta) + if g and g in by_guid and g not in hu_by_guid: + hu_by_guid[g] = meta + return hu_by_guid, bool(progress.get("complete")) + + +def _clear_hu(by_guid: dict, episodes: bool) -> None: + """Reconcile step: clear HU fields on all guid-bearing rows; the matches are re-set after. Only + rows that actually held HU data become dirty (SQLAlchemy skips a no-net-change UPDATE), so this is + cheap. The genres are the exception: every row gets Hungarian ones, from its HU twin or, lacking + one, from its own English list — otherwise the HU genre list mixed "Drama" (no twin) with "Dráma". + Only after a COMPLETE listing: one a Plex hiccup cut short would clear the HU text of every title + it failed to reach, until the next full sync (E32 S2, the watch reconcile's twin).""" + for group in by_guid.values(): + for row in group: + row.title_hu = row.summary_hu = None + if not episodes: + row.tagline_hu = None + row.genres_hu = hu_genres(row.genres) diff --git a/backend/app/plex/sync.py b/backend/app/plex/sync.py index 1dbd038e..c84a053f 100644 --- a/backend/app/plex/sync.py +++ b/backend/app/plex/sync.py @@ -15,7 +15,6 @@ dead `enabled` column) is a separate follow-up. Cast + intro/credit markers are item-detail/player path (P2), not here. """ -from collections.abc import Iterable from datetime import datetime, timezone from sqlalchemy import func, select @@ -23,7 +22,7 @@ from sqlalchemy.orm import Session from app import logs, sysconfig from app.models import PlexCollection, PlexItem, PlexLibrary, PlexSeason, PlexShow -from app.plex import detail, people +from app.plex import detail, overlay, people from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.plex.genres import hu_genres from app.plex.lookups import by_rating_key @@ -194,20 +193,6 @@ def _enabled_section_keys(db: Session) -> set[str] | None: return keys or None -def _overlay_map(db: Session) -> dict[str, str]: - """The configured HU-overlay pairing: {huSectionKey: enSectionKey} from `plex_overlay_map` - ('huKey:enKey,...'). Each HU section is synced as a non-browsable overlay onto its EN twin. - Malformed pairs are skipped.""" - raw = sysconfig.get_str(db, "plex_overlay_map") - out: dict[str, str] = {} - for pair in raw.split(","): - hu, sep, en = pair.partition(":") - hu, en = hu.strip(), en.strip() - if sep and hu and en: - out[hu] = en - return out - - def sync(db: Session) -> dict: """Full reconcile of the enabled movie/show sections. Returns a small stats dict.""" if not sysconfig.get_bool(db, "plex_enabled"): @@ -223,7 +208,7 @@ def sync(db: Session) -> dict: "pruned": 0, } wanted = _enabled_section_keys(db) - overlay_map = _overlay_map(db) # {huKey: enKey} + overlay_map = overlay.overlay_map(db) # {huKey: enKey} overlay_keys = set(overlay_map) try: with PlexClient(db) as plex: @@ -267,7 +252,7 @@ def sync(db: Session) -> dict: lib.metadata_lang = "hu" lib.overlay_of = en_lib.id db.flush() - _sync_overlay(db, plex, lib, en_lib.id, stats) + overlay.sync_overlay(db, plex, lib, en_lib.id, stats) db.commit() # Rich-fetch a bounded batch of not-yet-enriched titles' FULL cast, so the whole library # becomes filterable over successive runs (the cheap listing above only stored the top ~3). @@ -661,98 +646,6 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> _prune_stale(db, shows, shows_seen, stats, complete=shows_progress.get("complete", False)) -def _sync_overlay( - db: Session, plex: PlexClient, lib: PlexLibrary, en_lib_id: int, stats: dict -) -> None: - """Write an overlay section's Hungarian metadata onto its EN twin's rows, matched by guid. Creates - NO browsable rows — an overlay library is never listed on its own. Only title/summary/tagline/genres - differ by language; the language-neutral people_text feeds both search vectors. A show section's - episodes are overlaid too (E30): the HU section lists every episode with its Hungarian title and - summary, and the episode rows live in the EN show library. - - This is a full RECONCILE: every guid-bearing EN row's HU fields are cleared first, then the HU - section repopulates the matches — so an EN row whose HU twin vanished (or was re-guid'd) falls back - to EN instead of keeping stale Hungarian metadata. HU items with no EN match are skipped.""" - if lib.kind == "movie": - passes = [ - ( - _TYPE_MOVIE, - select(PlexItem).where( - PlexItem.library_id == en_lib_id, - PlexItem.kind == "movie", - PlexItem.guid.is_not(None), - ), - ) - ] - else: - passes = [ - ( - _TYPE_SHOW, - select(PlexShow).where( - PlexShow.library_id == en_lib_id, PlexShow.guid.is_not(None) - ), - ), - ( - _TYPE_EPISODE, - select(PlexItem).where( - PlexItem.library_id == en_lib_id, - PlexItem.kind == "episode", - PlexItem.guid.is_not(None), - ), - ), - ] - for item_type, rows_q in passes: - _overlay_pass(db, plex, lib, item_type, db.execute(rows_q).scalars(), stats) - db.flush() - - -def _overlay_pass( - db: Session, - plex: PlexClient, - lib: PlexLibrary, - item_type: int, - rows: Iterable[PlexItem] | Iterable[PlexShow], - stats: dict, -) -> None: - """One kind of the overlay (movies, shows or episodes) — see `_sync_overlay`.""" - # guid -> the EN rows carrying it (a list: two editions of one film can share a guid; both overlay). - by_guid: dict[str, list[PlexItem | PlexShow]] = {} - for r in rows: - if r.guid: # the passes select guid-bearing rows only - by_guid.setdefault(r.guid, []).append(r) - # Reconcile step: clear HU fields on all guid-bearing rows. Only rows that actually held HU data - # become dirty (SQLAlchemy skips a no-net-change UPDATE), so this is cheap; matches are re-set below. - # The genres are the exception: every row gets Hungarian ones, from its HU twin or, lacking one, from - # its own English list — otherwise the HU genre list mixed "Drama" (no twin) with "Dráma". - episodes = item_type == _TYPE_EPISODE - for group in by_guid.values(): - for row in group: - row.title_hu = row.summary_hu = None - if not episodes: - row.tagline_hu = None - row.genres_hu = hu_genres(row.genres) - seen: set[str] = ( - set() - ) # first HU entry for a guid wins; a dup HU item can't overwrite or double-count - for meta in _paginate(plex, lib.plex_key, item_type): - g = _canonical_guid(meta) - targets = by_guid.get(g) if g else None - if not targets or g in seen: - continue - seen.add(g) - title_hu = (meta.get("title") or "").strip() or None - summary_hu = meta.get("summary") - tagline_hu = (meta.get("tagline") or "").strip() or None - genres_hu = hu_genres(_tags(meta, "Genre")) - for row in targets: - row.title_hu = title_hu - row.summary_hu = summary_hu - if not episodes: - row.tagline_hu = tagline_hu - row.genres_hu = genres_hu or row.genres_hu - stats["overlaid"] = stats.get("overlaid", 0) + 1 - - def _upsert_collection( db: Session, existing: PlexCollection | None, rk: str, lib_id: int, meta: dict ) -> PlexCollection: @@ -815,6 +708,15 @@ def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: di select(PlexCollection).where(PlexCollection.library_id == lib.id) ).scalars() } + if not cols and existing: + # The catalog's own rule (see the module doc): an empty listing while we hold rows is a Plex + # glitch, not a library emptied of collections. Reconciling on it dropped every collection + # (its user-set `editable` flag with it) and every member's keys (E32 S2). + log.warning( + "Plex listed no collections for a library that has some; keeping them this run", + extra=logs.event("plex.sync.collections_empty", library=lib.plex_key), + ) + return seen: set[str] = set() membership: dict[str, list[str]] = {} # member rating_key -> [collection rating_key, ...] for meta in cols: diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 47c014d3..e6043bbd 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -72,13 +72,17 @@ def _rk_to_id(db: Session) -> dict[str, int]: def _scan_plex_states( db: Session, plex: PlexClient -) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int]: +) -> 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, - prog_at)}`` for every mirrored leaf Plex holds a watch signal for, plus the scanned-leaf count. - Shared by the one-time import and the full reconcile (both need Plex's whole current picture).""" + 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 + full reconcile (both need Plex's whole current picture). 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`).""" item_id_by_rk = _rk_to_id(db) wanted = _enabled_section_keys(db) scanned = 0 + complete = True out: dict[int, tuple[str, int, datetime | None, datetime | None]] = {} for s in plex.sections(): stype = s.get("type") @@ -88,7 +92,8 @@ def _scan_plex_states( if wanted is not None and key not in wanted: continue item_type = _TYPE_MOVIE if stype == "movie" else _TYPE_EPISODE - for meta in _paginate(plex, key, item_type): + progress: dict = {} + for meta in _paginate(plex, key, item_type, progress): scanned += 1 item_id = item_id_by_rk.get(str(meta.get("ratingKey") or "")) if item_id is None: @@ -98,7 +103,8 @@ def _scan_plex_states( ) if res is not None: out[item_id] = res - return out, scanned + complete = complete and bool(progress.get("complete")) + return out, scanned, complete def import_owner_watch_state(db: Session, link: PlexLink) -> dict: @@ -111,7 +117,8 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict: stats = {"watched": 0, "in_progress": 0, "scanned": 0} try: with PlexClient(db) as plex: - scanned_states, stats["scanned"] = _scan_plex_states(db, plex) + # A partial scan is still safe here: the import only ever ADDS what Plex reports. + scanned_states, stats["scanned"], _complete = _scan_plex_states(db, plex) # One row per item (keyed by item_id → inherently de-duped) then bulk-UPSERT. An upsert is # idempotent AND concurrency-safe: a repeated/overlapping enable (the toggle firing twice # while the multi-second import runs) can't raise a duplicate-key error — it re-updates. @@ -590,6 +597,20 @@ def run_plex_watch_sync(db: Session) -> dict: return total +def _clear_unwatched(db: Session, user_id: int, plex_states: dict) -> int: + """Clear the user's rows Plex no longer has a record for — the mirrored ones (Plex un-watched + them); hidden and dirty rows stay (a dirty one is re-pushed by `_repush_dirty`). Called only + after a COMPLETE scan: a walk a Plex hiccup cut short leaves out rows it merely failed to reach, + and clearing those un-watched them here until the next complete run (E32 S2).""" + n = 0 + for st in db.execute(select(PlexState).where(PlexState.user_id == user_id)).scalars().all(): + if st.item_id in plex_states or st.status == "hidden" or not st.synced_to_plex: + continue + db.delete(st) + n += 1 + return n + + def run_plex_watch_reconcile(db: Session) -> dict: """Scheduler job (default daily) — the authoritative full reconcile. Rescans every section and, per item, uses `synced_to_plex` to settle what the incremental feed can't: @@ -608,7 +629,7 @@ def run_plex_watch_reconcile(db: Session) -> dict: try: with PlexClient(db) as plex: _resolve_account_id(db, plex, link) - plex_states, scanned = _scan_plex_states(db, plex) + plex_states, scanned, complete = _scan_plex_states(db, plex) total["scanned"] += scanned for item_id, (status, pos, watched_at, prog_at) in plex_states.items(): @@ -616,18 +637,14 @@ def run_plex_watch_reconcile(db: Session) -> dict: if ch: total["applied"] += 1 - # Siftlode rows Plex no longer has a record for: clear the ones that were mirrored - # (Plex un-watched them); leave hidden + dirty (dirty gets re-pushed just below). - for st in ( - db.execute(select(PlexState).where(PlexState.user_id == link.user_id)) - .scalars() - .all() - ): - if st.item_id in plex_states or st.status == "hidden": - continue - if st.synced_to_plex: - db.delete(st) - total["cleared"] += 1 + if complete: + total["cleared"] += _clear_unwatched(db, link.user_id, plex_states) + else: + total["incomplete"] = total.get("incomplete", 0) + 1 + log.warning( + "the Plex scan ended before the server's total; nothing is cleared this run", + extra=logs.event("plex.watch.reconcile_incomplete", user_id=link.user_id), + ) total["repushed"] += _repush_dirty(db, plex, link) link.last_watch_sync_at = datetime.now(timezone.utc) diff --git a/backend/tests/test_plex_overlay.py b/backend/tests/test_plex_overlay.py index bf70ca34..3497da53 100644 --- a/backend/tests/test_plex_overlay.py +++ b/backend/tests/test_plex_overlay.py @@ -16,18 +16,21 @@ from app import sysconfig from app.db import engine from app.models import PlexItem, PlexLibrary, PlexShow from app.plex.genres import hu_genre, hu_genres -from app.plex.sync import _apply_item, _canonical_guid, _sync_overlay +from app.plex.overlay import sync_overlay +from app.plex.sync import _apply_item, _canonical_guid from app.plex.query import content_libs class _StubPlex: - """Minimal PlexClient stand-in: _sync_overlay only calls section_items via _paginate.""" + """Minimal PlexClient stand-in: sync_overlay only calls section_items via _paginate. `total` + over the item count is a listing cut short (the server said more than it handed over).""" - def __init__(self, items: list[dict]): + def __init__(self, items: list[dict], total: int | None = None): self._items = items + self._total = len(items) if total is None else total def section_items(self, key, item_type, start, size): - return self._items[start : start + size], len(self._items) + return self._items[start : start + size], self._total def _hu_movie(rk: str, title: str, summary: str, genres: list[str], guid: str | None) -> dict: @@ -71,7 +74,7 @@ def test_sync_overlay_writes_hu_onto_matched_en_rows(db): ] ) stats = {"overlaid": 0} - _sync_overlay(db, stub, hu_lib, en_lib.id, stats) + sync_overlay(db, stub, hu_lib, en_lib.id, stats) db.commit() db.refresh(matched) @@ -185,13 +188,51 @@ def test_overlay_reconcile_clears_stale_hu_when_twin_vanishes(db): ) db.add(row) db.commit() - # The HU section no longer lists this film (empty overlay). - _sync_overlay(db, _StubPlex([]), hu_lib, en_lib.id, {"overlaid": 0}) + # The HU section, read in full, no longer lists this film (only another one). + other = _hu_movie("h9", "Más", "más", [], "tmdb://99") + sync_overlay(db, _StubPlex([other]), hu_lib, en_lib.id, {"overlaid": 0}) db.commit() db.refresh(row) assert row.title_hu is None and row.summary_hu is None and row.genres_hu is None +def test_a_listing_cut_short_keeps_the_hungarian_text_it_did_not_reach(db): + """E32 S2: the reconcile clears HU text only after a COMPLETE listing. A walk a Plex hiccup cut + short (an empty page before the server's total), or an empty section, says nothing about the + titles it did not reach — clearing them dropped their Hungarian text until the next full sync.""" + en_lib = PlexLibrary(plex_key="4", title="Movies", kind="movie") + hu_lib = PlexLibrary(plex_key="9", title="Filmek", kind="movie", metadata_lang="hu") + db.add_all([en_lib, hu_lib]) + db.flush() + hu_lib.overlay_of = en_lib.id + kept = PlexItem( + rating_key="m1", + library_id=en_lib.id, + kind="movie", + title="Heat", + guid="tmdb:1", + title_hu="Szemtől szemben", + summary_hu="régi", + genres_hu=["Akció"], + ) + fresh = PlexItem(rating_key="m2", library_id=en_lib.id, kind="movie", title="Up", guid="tmdb:2") + db.add_all([kept, fresh]) + db.commit() + stats: dict = {} + reached = _hu_movie("h2", "Fel", "fel", [], "tmdb://2") + sync_overlay(db, _StubPlex([reached], total=2), hu_lib, en_lib.id, stats) + db.commit() + db.refresh(kept) + db.refresh(fresh) + assert kept.title_hu == "Szemtől szemben" and kept.summary_hu == "régi" + assert fresh.title_hu == "Fel" # what the listing did reach is still applied + assert stats["overlay_incomplete"] == 1 + sync_overlay(db, _StubPlex([]), hu_lib, en_lib.id, {}) # an empty section: the same answer + db.commit() + db.refresh(kept) + assert kept.title_hu == "Szemtől szemben" + + def test_apply_item_keeps_prior_guid_when_listing_lacks_guid(): row = PlexItem(rating_key="m1", kind="movie", guid="tmdb:1") _apply_item(row, lib_id=7, kind="movie", meta={"title": "Heat"}) # no Guid array @@ -225,7 +266,7 @@ def test_overlay_applies_to_all_editions_sharing_a_guid(db): ] ) stats = {"overlaid": 0} - _sync_overlay(db, stub, hu_lib, en_lib.id, stats) + sync_overlay(db, stub, hu_lib, en_lib.id, stats) db.commit() db.refresh(ed1) db.refresh(ed2) @@ -291,7 +332,7 @@ def test_a_show_overlay_writes_hungarian_episode_titles_and_summaries(db): ], } ) - _sync_overlay(db, stub, hu_lib, en_lib.id, {}) + sync_overlay(db, stub, hu_lib, en_lib.id, {}) db.commit() db.refresh(ep) db.refresh(show) @@ -328,7 +369,7 @@ def test_the_overlay_gives_every_row_one_hungarian_word_per_genre(db): "h1", "Szemtől szemben", "HU.", ["Action", "Bűnügyi", "Drama", "Dráma"], "tmdb://1" ) meta["tagline"] = "Magyar mottó." - _sync_overlay(db, _StubPlex([meta]), hu_lib, en_lib.id, {}) + sync_overlay(db, _StubPlex([meta]), hu_lib, en_lib.id, {}) db.commit() db.refresh(twin) db.refresh(lone) @@ -353,7 +394,7 @@ def test_a_twin_without_genres_keeps_the_rows_own_hungarian_list(db): ) db.add(row) db.commit() - _sync_overlay( + sync_overlay( db, _StubPlex([_hu_movie("h1", "Szemtől szemben", "HU.", [], "tmdb://1")]), hu_lib, diff --git a/backend/tests/test_plex_paths.py b/backend/tests/test_plex_paths.py new file mode 100644 index 00000000..446e8bd7 --- /dev/null +++ b/backend/tests/test_plex_paths.py @@ -0,0 +1,82 @@ +"""plex/paths — the Plex-side path → local mount translation (E32 S2: 23% covered, and a wrong answer +here reads as "media file not found" for a whole library at once, with nothing else failing). + +Pure logic: the admin map is read through sysconfig, stubbed per test. +""" + +from pathlib import Path + +import pytest + +from app.plex import paths + + +@pytest.fixture +def path_map(monkeypatch): + def use(raw: str) -> None: + monkeypatch.setattr(paths.sysconfig, "get_str", lambda db, key: raw) + + return use + + +def test_the_map_parses_both_separators_drops_junk_and_puts_the_longest_prefix_first(): + raw = " /data=/mnt/a ,\n/data/movies = /mnt/m\nnot-a-pair,=/x,/y=, /data/tv=/mnt/tv " + assert paths.parse_map(raw) == [ + ("/data/movies", "/mnt/m"), + ("/data/tv", "/mnt/tv"), + ("/data", "/mnt/a"), + ] + assert paths.parse_map("") == [] + + +def test_the_longest_matching_prefix_wins(path_map): + path_map("/data=/mnt/a,/data/movies=/mnt/m") + assert paths.map_path(None, "/data/movies/Foo.mkv") == str(Path("/mnt/m") / "Foo.mkv") + assert paths.map_path(None, "/data/tv/Bar.mkv") == str(Path("/mnt/a") / "tv/Bar.mkv") + + +def test_a_prefix_matches_only_at_a_path_boundary(path_map): + # "/data/movies2" shares the characters of "/data/movies" but is another folder. + path_map("/data/movies=/mnt/m") + assert paths.map_path(None, "/data/movies2/Foo.mkv") == "/data/movies2/Foo.mkv" + + +def test_the_prefix_itself_and_a_trailing_slash_in_the_map(path_map): + path_map("/data/movies/=/mnt/m") + assert paths.map_path(None, "/data/movies/Foo.mkv") == str(Path("/mnt/m") / "Foo.mkv") + path_map("/data/movies=/mnt/m") + assert paths.map_path(None, "/data/movies") == "/mnt/m" + + +def test_no_map_leaves_the_path_as_plex_sees_it(path_map): + path_map("") + assert paths.map_path(None, "/data/movies/Foo.mkv") == "/data/movies/Foo.mkv" + + +def test_the_local_file_must_exist(path_map, tmp_path): + (tmp_path / "movies").mkdir() + (tmp_path / "movies" / "Foo.mkv").write_bytes(b"x") + path_map(f"/data={tmp_path}") + assert ( + paths.local_media_path(None, "/data/movies/Foo.mkv") + == (tmp_path / "movies" / "Foo.mkv").resolve() + ) + assert paths.local_media_path(None, "/data/movies/Missing.mkv") is None + assert paths.local_media_path(None, "/data/movies") is None # a folder is not a file + assert paths.local_media_path(None, None) is None + assert paths.local_media_path(None, "") is None + + +def test_a_path_that_cannot_be_resolved_is_logged_and_answered_none(path_map, monkeypatch, caplog): + path_map("") + + def broken(self, strict=False): + raise OSError("dead mount") + + monkeypatch.setattr(paths.Path, "resolve", broken) + assert paths.local_media_path(None, "/data/movies/Foo.mkv") is None + rec = next( + r for r in caplog.records if getattr(r, "event", None) == "plex.paths.resolve_failed" + ) + # The title stays out of the log line: the path carries it. + assert "Foo" not in rec.getMessage() and rec.attrs == {"reason": "OSError"} diff --git a/backend/tests/test_plex_watch_reconcile.py b/backend/tests/test_plex_watch_reconcile.py new file mode 100644 index 00000000..08afa814 --- /dev/null +++ b/backend/tests/test_plex_watch_reconcile.py @@ -0,0 +1,167 @@ +"""The daily full watch reconcile (`run_plex_watch_reconcile`) — the one pass that DELETES watch state: +a mirrored row Plex no longer has a record for is read as "Plex un-watched it" and cleared. E32 S2 +found it trusting a scan a Plex hiccup had cut short; these hold what it may and may not clear. +""" + +import pytest +from sqlalchemy import select + +from app import sysconfig +from app.models import PlexItem, PlexLibrary, PlexLink, PlexState, User +from app.plex import watch_sync + + +class _FakePlex: + """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.""" + + def __init__(self, leaves: list[dict], total: int | None = None): + self.leaves = leaves + self.total = len(leaves) if total is None else total + self.calls: list[tuple] = [] + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def sections(self): + return [{"type": "movie", "key": "1"}] + + def section_items(self, key, item_type, start, size): + return self.leaves[start : start + size], self.total + + def scrobble(self, rk): + self.calls.append(("scrobble", rk)) + + def unscrobble(self, rk): + self.calls.append(("unscrobble", rk)) + + def set_timeline(self, rk, time_ms, duration_ms): + self.calls.append(("timeline", rk, time_ms)) + + +@pytest.fixture +def world(db): + """Five mirrored movies and the owner's link; each test sets the local states it needs.""" + sysconfig.set_value(db, "plex_enabled", True) + lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) + user = User(email="reconcile@example.com") + db.add_all([lib, user]) + db.flush() + items = {} + for rk in ("m1", "m2", "m3", "m4", "m5"): + items[rk] = PlexItem( + rating_key=rk, library_id=lib.id, kind="movie", title=rk, duration_s=6000 + ) + db.add(items[rk]) + db.add( + PlexLink( + user_id=user.id, + uses_admin=True, + sync_enabled=True, + initial_import_done=True, + plex_account_id=1, + ) + ) + db.flush() + + def state(rk, **kw): + db.add(PlexState(user_id=user.id, item_id=items[rk].id, **kw)) + + def rows(): + db.expire_all() + by_id = {i.id: rk for rk, i in items.items()} + return { + by_id[s.item_id]: (s.status, s.position_seconds, s.synced_to_plex) + for s in db.execute(select(PlexState).where(PlexState.user_id == user.id)).scalars() + } + + return state, rows + + +def _run(db, monkeypatch, plex): + monkeypatch.setattr(watch_sync, "PlexClient", lambda _db: plex) + return watch_sync.run_plex_watch_reconcile(db) + + +def _local_states(state): + state("m1", status="watched", synced_to_plex=True) # mirrored; Plex has no record now + state("m2", status="watched", synced_to_plex=False) # local change that never reached Plex + state("m3", status="hidden", synced_to_plex=True) # Siftlode-only, never cleared + state("m5", status="new", position_seconds=300, synced_to_plex=True) # Plex says otherwise + + +def test_a_complete_scan_clears_what_plex_unwatched_and_keeps_the_rest(db, monkeypatch, world): + state, rows = world + _local_states(state) + db.commit() + plex = _FakePlex( + [ + {"ratingKey": "m3"}, # listed, no watch signal + {"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000}, + {"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000}, + ] + ) + out = _run(db, monkeypatch, plex) + assert rows() == { + "m2": ("watched", 0, True), # re-pushed, then settled + "m3": ("hidden", 0, True), + "m4": ("watched", 0, True), # pulled in + "m5": ("new", 120, True), # Plex's position wins + } + assert ("scrobble", "m2") in plex.calls + assert out["cleared"] == 1 and out["repushed"] == 1 and "incomplete" not in out + + +def test_a_scan_cut_short_clears_nothing(db, monkeypatch, world): + state, rows = world + _local_states(state) + db.commit() + # The server said 5, the walk got 3: m1 may simply not have been reached. + plex = _FakePlex( + [ + {"ratingKey": "m3"}, + {"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000}, + {"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000}, + ], + total=5, + ) + out = _run(db, monkeypatch, plex) + after = rows() + assert after["m1"] == ("watched", 0, True) # NOT un-watched on a partial picture + assert after["m4"] == ("watched", 0, True) and after["m5"] == ("new", 120, True) # pulls stand + assert after["m2"] == ("watched", 0, True) # the dirty row is still pushed up + assert out["cleared"] == 0 and out["incomplete"] == 1 + + +def test_an_empty_listing_is_not_a_library_of_unwatched_titles(db, monkeypatch, world): + state, rows = world + state("m1", status="watched", synced_to_plex=True) + db.commit() + out = _run(db, monkeypatch, _FakePlex([])) + assert rows() == {"m1": ("watched", 0, True)} + assert out["cleared"] == 0 + + +def test_a_dirty_resume_is_re_pushed_as_a_timeline_and_a_dirty_reset_as_an_unscrobble( + db, monkeypatch, world +): + state, rows = world + state("m1", status="new", position_seconds=300, synced_to_plex=False) + state("m2", status="new", position_seconds=2, synced_to_plex=False) # under the 5 s floor + db.commit() + plex = _FakePlex([{"ratingKey": rk} for rk in ("m1", "m2", "m3", "m4", "m5")]) + _run(db, monkeypatch, plex) + assert ("timeline", "m1", 300_000) in plex.calls + assert ("unscrobble", "m2") in plex.calls + assert {rk: synced for rk, (_s, _p, synced) in rows().items()} == {"m1": True, "m2": True} + + +def test_watch_fields_map_to_a_state_only_when_plex_holds_a_signal(): + to = watch_sync._plex_watch_to_state + assert to(None, None, None) is None + assert to(0, 4_000, None) is None # a stray few seconds is not "in progress" + assert to(0, 90_000, 1_700_000_000)[:2] == ("new", 90) + assert to("2", 90_000, 1_700_000_000)[:2] == ("watched", 0) # watched wins over an offset diff --git a/backend/tests/test_source_size.py b/backend/tests/test_source_size.py index 467f3e1e..f0388bcf 100644 --- a/backend/tests/test_source_size.py +++ b/backend/tests/test_source_size.py @@ -32,7 +32,6 @@ BIG_MODULES: dict[str, int] = { "app/models.py": 1607, "app/plex/stream.py": 1128, "app/plex/subtitles.py": 1092, - "app/plex/sync.py": 841, "app/routes/admin.py": 864, "app/routes/auth.py": 1353, "app/routes/downloads.py": 1237,