peter
b1832afd18
Mechanical, whole-tree `ruff format` pass — no behaviour change. Recorded in .git-blame-ignore-revs (next commit) so git blame skips it.
241 lines
11 KiB
Python
241 lines
11 KiB
Python
"""On-disk layout for downloaded media — Plex-friendly tree + .nfo/poster sidecars.
|
||
|
||
The physical filename is system-decided and bound to the source id (never renamed; the user's
|
||
custom name is display-only, applied at local-download time). Layout is admin-configurable:
|
||
|
||
plex: {Channel}/Season {Year}/{Channel} - {YYYY-MM-DD} - {title} [{id}].{ext}
|
||
flat: {title} [{id}].{ext}
|
||
|
||
Plus a Kodi/Plex-readable `<basename>.nfo` (episodedetails) and `<basename>.jpg` thumbnail, so
|
||
a Plex library using a metadata agent picks up rich info. Channel = "show", video = episode,
|
||
date-based — the community-standard mapping for YouTube-in-Plex.
|
||
"""
|
||
|
||
import re
|
||
import shutil
|
||
import unicodedata
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from xml.sax.saxutils import escape
|
||
|
||
# Reserve room for the "_id.ext" suffix + parent dirs within the 255-char component limit.
|
||
_MAX_TITLE = 120
|
||
# User-facing (Content-Disposition) names can be longer than an on-disk component.
|
||
_MAX_DISPLAY = 150
|
||
# The ONLY characters allowed in a generated name — letters, digits, and `._-`. Everything else
|
||
# (spaces, and every shell/filesystem-unpleasant metacharacter: ! ? * & $ ( ) [ ] { } ' " ; | < >
|
||
# etc.) collapses to a single underscore, so names are safe to handle in any shell or tool.
|
||
_UNSAFE = re.compile(r"[^A-Za-z0-9._-]+")
|
||
# Unicode categories we drop entirely: emoji/pictographs (So), modifier symbols/skin-tones
|
||
# (Sk), and every control/format/private/surrogate/unassigned char (C*).
|
||
_DROP_CATEGORIES = {"So", "Sk", "Cc", "Cf", "Cs", "Co", "Cn"}
|
||
# Latin letters NFKD can't decompose to an ASCII base — fold them explicitly (trilingual HU/DE
|
||
# plus common European). Everything else with a diacritic (á/é/ö/ő/ü/ű, ç, ñ…) NFKD-decomposes
|
||
# into base + a combining mark we then drop, and any remaining non-ASCII is stripped, so every
|
||
# generated name is pure ASCII: no spaces, no accents — portable and free of the `?` mojibake
|
||
# that accented names show in non-UTF-8 tools.
|
||
_FOLD_MAP = str.maketrans(
|
||
{
|
||
"ß": "ss",
|
||
"ẞ": "SS",
|
||
"ø": "o",
|
||
"Ø": "O",
|
||
"ł": "l",
|
||
"Ł": "L",
|
||
"đ": "d",
|
||
"Đ": "D",
|
||
"ð": "d",
|
||
"Ð": "D",
|
||
"þ": "th",
|
||
"Þ": "Th",
|
||
"æ": "ae",
|
||
"Æ": "AE",
|
||
"œ": "oe",
|
||
"Œ": "OE",
|
||
"ı": "i",
|
||
"İ": "I",
|
||
"·": " ",
|
||
"–": "-",
|
||
"—": "-",
|
||
}
|
||
)
|
||
|
||
|
||
def _ascii_fold(text: str) -> str:
|
||
"""Transliterate to pure ASCII: apply the explicit map, NFKD-split accented letters and drop
|
||
the combining marks (á→a, ő→o), then strip any leftover non-ASCII (non-Latin scripts that have
|
||
no ASCII base)."""
|
||
text = text.translate(_FOLD_MAP)
|
||
text = unicodedata.normalize("NFKD", text)
|
||
text = "".join(ch for ch in text if unicodedata.category(ch) != "Mn")
|
||
return text.encode("ascii", "ignore").decode("ascii")
|
||
|
||
|
||
@dataclass
|
||
class MediaMeta:
|
||
video_id: str
|
||
title: str
|
||
uploader: str
|
||
upload_date: date | None
|
||
# The .nfo <uniqueid type=...> — the source's scraper type (youtube / facebook / …), set from
|
||
# the SourceAdapter. Defaults to the source-neutral "generic" so an un-updated caller is at worst
|
||
# uninformative, never actively mislabelling a non-YouTube download as "youtube".
|
||
uniqueid_type: str = "generic"
|
||
duration_s: int | None = None
|
||
description: str | None = None
|
||
thumbnail_url: str | None = None
|
||
|
||
|
||
def _slug(name: str, limit: int, fallback: str) -> str:
|
||
"""Reduce an arbitrary (emoji-laden, clickbait, accented) title to a portable, shell-safe token.
|
||
|
||
NFKC-normalize → drop emoji/symbols/control chars → ASCII-fold accents (á→a, ő→o, ß→ss) →
|
||
replace every run of unsafe characters (spaces + shell/filesystem metacharacters) with a single
|
||
underscore → collapse separator pile-ups → trim junk from the ends → length-cap. The result is
|
||
pure ASCII containing only `[A-Za-z0-9._-]` — no spaces, accents, or characters that need
|
||
quoting in a shell."""
|
||
text = unicodedata.normalize("NFKC", name or "")
|
||
text = "".join(ch for ch in text if unicodedata.category(ch) not in _DROP_CATEGORIES)
|
||
text = _ascii_fold(text)
|
||
text = _UNSAFE.sub("_", text)
|
||
text = re.sub(r"[-_]{2,}", "_", text).strip("_.-")
|
||
if len(text) > limit: # pure ASCII → 1 char == 1 byte, so a char cap is also the byte cap
|
||
text = text[:limit].rstrip("_.-")
|
||
return text or fallback
|
||
|
||
|
||
def sanitize(name: str, limit: int = _MAX_TITLE) -> str:
|
||
"""On-disk path token: portable, shell-safe, pure ASCII (see `_slug`)."""
|
||
return _slug(name, limit, "untitled")
|
||
|
||
|
||
def display_filename(name: str) -> str:
|
||
"""Content-Disposition filename base — the same shell-safe slug policy as the on-disk name, just
|
||
with a larger length cap (a served name may be longer than a path component)."""
|
||
return _slug(name, _MAX_DISPLAY, "download")
|
||
|
||
|
||
def rel_path(meta: MediaMeta, ext: str, layout: str = "plex", asset_id: int | None = None) -> str:
|
||
"""Path of the media file relative to DOWNLOAD_ROOT (forward slashes).
|
||
|
||
`asset_id` disambiguates the file. A MediaAsset is unique on (source_kind, source_ref,
|
||
format_sig), so the SAME video downloaded in two formats is two distinct assets — but without
|
||
the id they compute the identical path, so the second silently overwrites the first and deleting
|
||
either removes the shared file. The `_a{id}` suffix (which Plex ignores — the .nfo sidecar carries
|
||
the metadata) keeps them apart. Optional so callers/tests that don't need it keep the plain name."""
|
||
title = sanitize(meta.title)
|
||
# The id goes into the name bare (no brackets — `[`/`]` are shell glob metacharacters); keep it
|
||
# shell-safe and bounded in case a URL source has no clean extractor id.
|
||
vid = _UNSAFE.sub("_", str(meta.video_id))[:40]
|
||
disc = f"_a{asset_id}" if asset_id is not None else ""
|
||
if layout == "flat":
|
||
return f"{title}_{vid}{disc}.{ext}"
|
||
channel = sanitize(meta.uploader or "Unknown_Channel", 80)
|
||
year = meta.upload_date.year if meta.upload_date else 0
|
||
d = meta.upload_date.isoformat() if meta.upload_date else "0000-00-00"
|
||
epname = f"{channel}_-_{d}_-_{title}_{vid}{disc}.{ext}"
|
||
return f"{channel}/Season_{year}/{epname}"
|
||
|
||
|
||
def edit_rel_path(user_id: int, asset_id: int, ext: str) -> str:
|
||
"""On-disk path (relative to DOWNLOAD_ROOT) for a phase-2 editor derivative.
|
||
|
||
Kept in a per-user `.edits/` tree — device-downloadable via the file endpoint, but NOT part of
|
||
the Plex library (these are user clips, not canonical episodes; there are no .nfo/poster
|
||
sidecars). The name is system-decided and bound to the derivative asset id."""
|
||
return f".edits/{user_id}/clip_{asset_id}.{ext}"
|
||
|
||
|
||
def download_filename(display_name: str, container: str | None, path: Path) -> str:
|
||
"""The Content-Disposition filename for a served download: the cleaned display name carrying a single
|
||
correct extension — the asset container, else the file's own suffix — without doubling it when the
|
||
name already ends in that extension. Shared by the private download and the public watch endpoints."""
|
||
ext = container or path.suffix.lstrip(".")
|
||
base = display_filename(display_name)
|
||
if base.lower().endswith(f".{ext.lower()}"):
|
||
base = base[: -(len(ext) + 1)]
|
||
return f"{base}.{ext}"
|
||
|
||
|
||
def abs_path(download_root: str, rel: str) -> Path:
|
||
return Path(download_root) / rel
|
||
|
||
|
||
def safe_abs_path(download_root: str, rel: str) -> Path | None:
|
||
"""The resolved absolute path for `rel`, but ONLY if it stays inside DOWNLOAD_ROOT and exists on
|
||
disk — otherwise None. Centralizes the path-traversal + existence guard every file-serving endpoint
|
||
shares, so the containment check lives in exactly one place."""
|
||
root = Path(download_root).resolve()
|
||
path = abs_path(download_root, rel).resolve()
|
||
if root not in path.parents or not path.exists():
|
||
return None
|
||
return path
|
||
|
||
|
||
def place_file(staging_file: Path, download_root: str, rel: str) -> None:
|
||
"""Move a completed staging file into its final location under DOWNLOAD_ROOT."""
|
||
dest = abs_path(download_root, rel)
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.move(str(staging_file), str(dest))
|
||
|
||
|
||
def _nfo_xml(meta: MediaMeta) -> str:
|
||
aired = meta.upload_date.isoformat() if meta.upload_date else ""
|
||
year = meta.upload_date.year if meta.upload_date else ""
|
||
parts = [
|
||
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>',
|
||
"<episodedetails>",
|
||
f" <title>{escape(meta.title or '')}</title>",
|
||
f" <showtitle>{escape(meta.uploader or '')}</showtitle>",
|
||
f" <studio>{escape(meta.uploader or '')}</studio>",
|
||
f" <aired>{aired}</aired>",
|
||
f" <premiered>{aired}</premiered>",
|
||
f" <year>{year}</year>",
|
||
f" <plot>{escape(meta.description or '')}</plot>",
|
||
f' <uniqueid type="{escape(meta.uniqueid_type)}" default="true">{escape(meta.video_id)}</uniqueid>',
|
||
]
|
||
if meta.duration_s:
|
||
parts.append(f" <runtime>{round(meta.duration_s / 60)}</runtime>")
|
||
parts.append("</episodedetails>")
|
||
return "\n".join(parts)
|
||
|
||
|
||
def _sidecar(media: Path, ext: str) -> Path:
|
||
"""The sibling sidecar path (`.nfo` / `.jpg`) for a media file. Swaps the extension on the FULL
|
||
media path — NOT via `media.with_suffix("")` first: stripping the real extension leaves a stem
|
||
whose OWN last dot may be inside the title (e.g. "Sonka Házilag."), which `with_suffix` would
|
||
then read as the extension and clobber, orphaning the sidecar under a truncated name."""
|
||
return media.with_suffix(ext)
|
||
|
||
|
||
def write_sidecars(download_root: str, rel: str, meta: MediaMeta, thumb_src: Path | None) -> bool:
|
||
"""Write `<basename>.nfo` + `<basename>.jpg` next to the media file. Returns True if written."""
|
||
media = abs_path(download_root, rel)
|
||
try:
|
||
_sidecar(media, ".nfo").write_text(_nfo_xml(meta), encoding="utf-8")
|
||
if thumb_src and thumb_src.exists():
|
||
shutil.copyfile(str(thumb_src), str(_sidecar(media, ".jpg")))
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def delete_asset_files(download_root: str, rel: str) -> None:
|
||
"""Remove the media file and its sidecars; prune now-empty parent dirs (best effort)."""
|
||
media = abs_path(download_root, rel)
|
||
for p in (media, _sidecar(media, ".nfo"), _sidecar(media, ".jpg")):
|
||
try:
|
||
p.unlink()
|
||
except OSError:
|
||
pass
|
||
# Prune empty Season/Channel dirs up to (but not including) the root.
|
||
root = Path(download_root).resolve()
|
||
parent = media.parent
|
||
while parent != root and root in parent.parents:
|
||
try:
|
||
parent.rmdir() # only succeeds if empty
|
||
except OSError:
|
||
break
|
||
parent = parent.parent
|