52 lines
2.2 KiB
Python
52 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
|