"""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 `.nfo` (episodedetails) and `.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 from app.downloads import mediakind # 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 — 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 _clean_upload_ext(filename: str) -> str: """The upload's extension reduced to a shell-safe, lowercased, bounded token (``""`` when the upload has no extension). The SINGLE definition feeding both the on-disk suffix and the stored `container` — they must agree (the served Content-Disposition suffix has to match the container), so the sanitize/truncation rule lives in one place, not two drifting copies.""" return _UNSAFE.sub("", Path(filename).suffix.lstrip(".")).lower()[:16] def upload_rel_path(user_id: int, asset_id: int, filename: str) -> str: """On-disk path (relative to DOWNLOAD_ROOT) for a locally-uploaded file. Kept in a per-user `.uploads/` tree — device-downloadable via the file endpoint, but NOT part of the Plex library (an upload is arbitrary user content, not a canonical episode; no .nfo/poster episode sidecars). The name is system-decided and bound to the asset id (`_a{id}` keeps two same-named uploads apart); the original filename is display metadata only. The extension is preserved (shell-safe, bounded) so the served file + its Content-Disposition carry a correct suffix; an extensionless upload stays extensionless.""" stem = sanitize(Path(filename).stem, 80) ext = _clean_upload_ext(filename) suffix = f".{ext}" if ext else "" return f".uploads/{user_id}/{stem}_a{asset_id}{suffix}" def upload_ext(filename: str) -> str | None: """The stored `container` for an uploaded file: its shell-safe lowercased extension, or None when the upload has no extension (a bare name). Shares `_clean_upload_ext` with `upload_rel_path` so the stored container and the on-disk suffix can never diverge.""" return _clean_upload_ext(filename) or None # The Content-Type used to serve an inline-playable media asset, keyed by its stored `container`. # DELIBERATELY a strict allow-list of known audio/video containers: the served type is chosen from # the asset's ffprobe-verified media_kind + this table, NEVER guessed from the raw (user-controlled) # upload extension. Without this, an uploaded media/HTML polyglot named "evil.html" — a real MP3/MP4 # that ffprobe classifies as media, so it dodges the generic-file force-download — would be served # with a Content-Type Starlette guesses from ".html" (text/html) and execute as stored XSS on the # serving origin. A media asset whose container isn't in this table falls back to a neutral # force-download attachment (see `served_media_type`), so nothing script-executable is ever inline. _MEDIA_MIME: dict[str, str] = { # video containers "mp4": "video/mp4", "m4v": "video/mp4", "mov": "video/quicktime", "webm": "video/webm", "mkv": "video/x-matroska", "ogv": "video/ogg", "avi": "video/x-msvideo", # audio containers "m4a": "audio/mp4", "mp3": "audio/mpeg", "aac": "audio/aac", "ogg": "audio/ogg", "oga": "audio/ogg", "opus": "audio/ogg", "weba": "audio/webm", "wav": "audio/wav", "flac": "audio/flac", } def served_media_type(media_kind: str | None, container: str | None) -> str | None: """The Content-Type a library file MUST be served with — a real audio/video MIME for a recognised media container, else ``None`` meaning "serve as a neutral force-download attachment". Returns None for a generic ``file`` kind AND for any media whose container we don't positively recognise as safe. Callers serve None as ``application/octet-stream`` + ``attachment`` so the served Content-Type is NEVER derived from the (attacker-controlled) file extension — the guard that stops uploaded HTML/SVG or a media/HTML polyglot from executing inline on the serving origin. A ``None`` media_kind (a legacy asset before the discriminator was backfilled) is treated as media so an old download keeps its inline player.""" if media_kind == mediakind.FILE: return None return _MEDIA_MIME.get((container or "").lower()) 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 = [ '', "", f" {escape(meta.title or '')}", f" {escape(meta.uploader or '')}", f" {escape(meta.uploader or '')}", f" {aired}", f" {aired}", f" {year}", f" {escape(meta.description or '')}", f' {escape(meta.video_id)}', ] if meta.duration_s: parts.append(f" {round(meta.duration_s / 60)}") parts.append("") 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 `.nfo` + `.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