siftlode/backend/app/downloads/storage.py

"""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

from app import logs
from app.downloads import mediakind

log = logs.get_logger(__name__)

# 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 _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 = [
        '<?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 as exc:
        # The download itself still completes, so the job reads as done while the library it was
        # downloaded FOR has no .nfo to read and no poster to show.
        log.warning(
            "the sidecars beside a download could not be written",
            extra=logs.event("download.sidecars.write_failed", reason=type(exc).__name__),
        )
        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 FileNotFoundError:
            pass  # silent-ok: already gone is the outcome this function wanted
        except OSError as exc:
            # The row goes and the bytes stay. Nothing else ever revisits this path, so the disk
            # quietly keeps a file no account has any way to see, let alone delete.
            log.info(
                "a deleted download left a file behind",
                extra=logs.event("download.asset.file_unlink_failed", reason=type(exc).__name__),
            )
    # 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
        # silent-ok: this `rmdir` failing is the loop's stop condition — the parent still
        # holds another download's files, so the prune has gone as far up as it may.
        except OSError:
            break
        parent = parent.parent