a88254c841
Optional Plex module groundwork (design locked 2026-07-05): plays the LOCAL physical file; Plex is metadata + optional watch-sync only. - models + migration 0044: plex_libraries/plex_shows/plex_seasons/plex_items (playable leaf, ffprobe playability class, markers, weighted FTS search_vector) + plex_states (per-user watch state, mirrors video_states) - sysconfig 'plex' group + config.py env defaults (server url/token[secret]/ path-map/libraries/sync-interval/max-transcodes) - app/plex/client.py (PlexClient httpx: server info, sections, items, metadata +markers, image) + app/plex/paths.py (Plex→local path map + file resolve) - routes/plex.py admin POST /api/plex/test (verify connection + list sections)
51 lines
2.2 KiB
Python
51 lines
2.2 KiB
Python
"""Map a Plex-side media path to Siftlode's local mount, so the player can open the physical file.
|
|
|
|
The Plex metadata gives each media Part an absolute path as the PLEX server sees it (e.g.
|
|
``/data/movies/Foo.mkv``). Siftlode plays the file directly, so it must translate that to the path
|
|
as SIFTLODE sees the same storage on a local mount (e.g. ``/plex-media/movies/Foo.mkv``). The map
|
|
is admin-configured (sysconfig ``plex_path_map``) as newline/comma-separated ``plexPrefix=localPrefix``
|
|
pairs; the longest matching Plex prefix wins. An empty map means the paths are identical on both
|
|
sides (Plex and Siftlode see the same filesystem).
|
|
"""
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import sysconfig
|
|
|
|
|
|
def parse_map(raw: str) -> list[tuple[str, str]]:
|
|
"""Parse the ``plexPrefix=localPrefix`` pairs, longest Plex prefix first."""
|
|
pairs: list[tuple[str, str]] = []
|
|
for chunk in (raw or "").replace("\n", ",").split(","):
|
|
chunk = chunk.strip()
|
|
if not chunk or "=" not in chunk:
|
|
continue
|
|
plex, local = chunk.split("=", 1)
|
|
plex, local = plex.strip(), local.strip()
|
|
if plex and local:
|
|
pairs.append((plex, local))
|
|
pairs.sort(key=lambda p: len(p[0]), reverse=True)
|
|
return pairs
|
|
|
|
|
|
def map_path(db: Session, plex_path: str) -> str:
|
|
"""Translate a Plex-side absolute path to the local mount path (no mapping = unchanged)."""
|
|
for plex_prefix, local_prefix in parse_map(sysconfig.get_str(db, "plex_path_map")):
|
|
if plex_path == plex_prefix or plex_path.startswith(plex_prefix.rstrip("/") + "/"):
|
|
rest = plex_path[len(plex_prefix):].lstrip("/")
|
|
return str(Path(local_prefix) / rest) if rest else local_prefix
|
|
return plex_path
|
|
|
|
|
|
def local_media_path(db: Session, plex_path: str | None) -> Path | None:
|
|
"""The resolved, existing local file for a Plex media path, or None if it can't be read.
|
|
The path originates from the trusted Plex server + an admin-set map (not user input), so no
|
|
traversal check is needed beyond requiring a real regular file to exist."""
|
|
if not plex_path:
|
|
return None
|
|
try:
|
|
p = Path(map_path(db, plex_path)).resolve()
|
|
except OSError:
|
|
return None
|
|
return p if p.is_file() else None
|