fix(plex): a listing cut short no longer clears watch state or Hungarian text
The daily watch reconcile reads "absent from Plex's scan" as "Plex un-watched it" and deletes the row — but it walked the sections without asking _paginate whether the walk reached the server's total, so a Plex hiccup mid-walk (an empty page early) un-watched every title it failed to reach until the next complete run. The scan now reports completeness; an incomplete one still pulls in what it saw, re-pushes dirty rows, and clears nothing (logged as plex.watch.reconcile_incomplete). The HU overlay had the same shape: clear every row's HU text, then refill from the listing. It now reads the listing first and clears only after a complete one. The overlay moved to plex/overlay.py (plex/sync.py drops under the 800-line limit and off the size ratchet's list). Tests: plex/paths (23% -> 100%: prefix boundary, longest prefix, dead mount), the reconcile's clear/keep/re-push rules, the cut-short scan and listing (each seen red with the guard removed).
This commit is contained in:
@@ -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)
|
||||
+12
-110
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"}
|
||||
@@ -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
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user