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

# Reserve room for the " [id].ext" suffix + parent dirs within the 255-byte component limit.
_MAX_TITLE = 120
# User-facing (Content-Disposition) names can be longer than an on-disk component.
_MAX_DISPLAY = 150
# Filesystem-illegal on Windows/Plex + control chars.
_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
# 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
    duration_s: int | None = None
    description: str | None = None
    thumbnail_url: str | None = None


def sanitize(name: str, limit: int = _MAX_TITLE) -> str:
    """Turn an arbitrary (emoji-laden, clickbait) title into a clean, space-free ASCII path token.

    Steps: NFKC-normalize (fold fancy width/compat forms) → drop emoji/symbols/control chars →
    ASCII-fold accents (á→a, ő→o, ß→ss) → strip filesystem-illegal chars → collapse whitespace and
    runs of the same punctuation → join words with underscores → trim junk punctuation from the ends
    → length-cap. The result is pure ASCII with no spaces or path separators (digits survive)."""
    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 = _ILLEGAL.sub("", text)
    # Collapse repeated punctuation ("!!!", "???", "-----") to a single mark.
    text = re.sub(r"([^\w\s])\1{1,}", r"\1", text)
    # Whitespace → single underscore (no spaces in generated names).
    text = re.sub(r"\s+", "_", text.strip())
    # Tidy separator pile-ups and trim junk from the ends.
    text = re.sub(r"_{2,}", "_", text).strip("_.-·—–|,;:!?ّ ")
    # Cap by BYTES, not characters: a path component's hard limit is 255 BYTES, and rel_path packs
    # the (byte-capped) channel + date + id + ext into that same component — so an accented/multi-byte
    # title capped only by char count could still overflow to ENAMETOOLONG. Truncate on a UTF-8
    # boundary (decode(..., "ignore") drops a trailing partial char) so we never split a character.
    if len(text.encode("utf-8")) > limit:
        text = text.encode("utf-8")[:limit].decode("utf-8", "ignore").rstrip("_.-")
    return text or "untitled"


def display_filename(name: str) -> str:
    """User-facing download filename (Content-Disposition) base: the same ASCII-slug policy as the
    on-disk name — emoji/symbols/control + illegal chars dropped, accents folded, whitespace joined
    with underscores — so a saved file is portable (no spaces, no accents). Only the length cap
    differs (a served name may be longer than an on-disk path component)."""
    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 = _ILLEGAL.sub("", text)
    text = re.sub(r"([^\w\s])\1{1,}", r"\1", text)
    text = re.sub(r"\s+", "_", text.strip())
    text = re.sub(r"_{2,}", "_", text).strip("_.-·—–|,;:!?ّ ")
    if len(text) > _MAX_DISPLAY:
        text = text[:_MAX_DISPLAY].rstrip("_.-")
    return text or "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)
    vid = meta.video_id
    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=\"youtube\" 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 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)
    base = media.with_suffix("")
    try:
        base.with_suffix(".nfo").write_text(_nfo_xml(meta), encoding="utf-8")
        if thumb_src and thumb_src.exists():
            shutil.copyfile(str(thumb_src), str(base.with_suffix(".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)
    base = media.with_suffix("")
    for p in (media, base.with_suffix(".nfo"), base.with_suffix(".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