feat(plex): Test-connection media mount probe

Extend admin POST /api/plex/test to resolve one real media file through the
local mount (a mirrored sample, else a single leaf fetched from Plex) and report
whether it is readable. A missing media bind-mount or a wrong plex_path_map is
now caught at Test time rather than only surfacing later as a dead player.
This commit is contained in:
npeter83 2026-07-05 06:58:34 +02:00
parent 3e1460759c
commit 2138f3c357

View file

@ -37,11 +37,14 @@ _TS_CONFIG = "public.unaccent_simple"
@router.post("/test") @router.post("/test")
def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
"""Admin: verify the configured Plex server URL + token, returning the server name and the """Admin: verify the configured Plex server URL + token, returning the server name and the
movie/show sections (for the config library-picker).""" movie/show sections (for the config library-picker). Also probes ONE real media file through
the local mount, so a missing bind-mount / wrong path-map is caught here rather than only
surfacing as a dead player later."""
try: try:
with PlexClient(db) as plex: with PlexClient(db) as plex:
info = plex.server_info() info = plex.server_info()
sections = plex.sections() sections = plex.sections()
media_check = _media_check(db, plex, sections)
except PlexNotConfigured as e: except PlexNotConfigured as e:
raise HTTPException(status_code=400, detail=str(e)) raise HTTPException(status_code=400, detail=str(e))
except PlexError as e: except PlexError as e:
@ -55,9 +58,44 @@ def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db)
for s in sections for s in sections
if s.get("type") in ("movie", "show") if s.get("type") in ("movie", "show")
], ],
"media_check": media_check,
} }
def _sample_plex_path(db: Session, plex: PlexClient, sections: list[dict]) -> str | None:
"""A real Plex-side media file path to validate the mount + path-map against. Prefers the
already-mirrored catalog (cheap); before the first sync, pulls one movie leaf from an enabled
section straight from Plex."""
row = db.query(PlexItem.file_path).filter(PlexItem.file_path.isnot(None)).first()
if row and row[0]:
return row[0]
wanted = plex_sync._enabled_section_keys(db)
for s in sections:
if s.get("type") != "movie":
continue
key = str(s.get("key"))
if wanted is not None and key not in wanted:
continue
items, _ = plex.section_items(key, 1, 0, 1)
if items:
fp = plex_sync._media_facts(items[0]).get("file_path")
if fp:
return fp
return None
def _media_check(db: Session, plex: PlexClient, sections: list[dict]) -> dict:
"""Resolve a sample Plex file path to its local mount path and confirm it's readable, so the
admin sees at Test time whether the physical media is actually reachable (bind-mount +
path-map correct). ``checked=False`` when no sample is available (empty library)."""
plex_path = _sample_plex_path(db, plex, sections)
if not plex_path:
return {"checked": False}
local_path = plex_paths.map_path(db, plex_path)
ok = plex_paths.local_media_path(db, plex_path) is not None
return {"checked": True, "ok": ok, "plex_path": plex_path, "local_path": local_path}
@router.post("/sync") @router.post("/sync")
def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
"""Admin: mirror the enabled Plex sections into the local catalog. Synchronous (can take a """Admin: mirror the enabled Plex sections into the local catalog. Synchronous (can take a