hms-mediaengine/hms_app/media.py

"""Medienbibliothek: Upload/Import, Thumbnails, Metadaten, Löschen."""

from __future__ import annotations

import json
import re
import shutil
import time
import urllib.parse
from pathlib import Path

import gi
gi.require_version("Gst", "1.0")
from gi.repository import Gst

Gst.init(None)

VIDEO_EXT = {".mp4", ".m4v", ".avi", ".mkv", ".mov", ".webm"}
IMAGE_EXT = {".jpg", ".jpeg", ".png", ".bmp"}


def kind_of(name: str) -> str | None:
    ext = Path(name).suffix.lower()
    if ext in VIDEO_EXT:
        return "video"
    if ext in IMAGE_EXT:
        return "image"
    return None


def sanitize(name: str) -> str:
    """Dateiname portabel und sicher machen (keine Pfade/Sonderzeichen)."""
    base = Path(name).name
    base = re.sub(r"[^A-Za-z0-9._-]", "_", base).strip("._")
    return base or "file"


class MediaLibrary:
    """Dateibasierte Bibliothek inkl. Thumbnail- und Metadaten-Cache."""

    def __init__(self, settings: dict, root: Path):
        self.settings = settings
        self.root = root
        self.dir = (root / settings["media"]["dir"]).resolve()
        self.thumbs = (root / settings["media"]["thumbs_dir"]).resolve()

    # ---------- Pfade ----------

    def ensure_dirs(self) -> None:
        self.dir.mkdir(parents=True, exist_ok=True)
        self.thumbs.mkdir(parents=True, exist_ok=True)

    def kind_of(self, name: str) -> str | None:
        """Medientyp einer Datei (video/image) – Instanz-Zugriff für Engine."""
        return kind_of(name)

    def file_path(self, name: str) -> Path:
        return self.dir / sanitize(name)

    def thumb_path(self, name: str) -> Path:
        return self.thumbs / (Path(name).stem + ".jpg")

    def _safe(self, base: Path, raw: str) -> Path | None:
        name = urllib.parse.unquote(raw or "")
        if not name or "/" in name or "\\" in name or name in (".", ".."):
            return None
        p = base / name
        try:
            if p.resolve().parent != base.resolve():
                return None
        except OSError:
            return None
        return p

    def safe_file(self, raw: str) -> Path | None:
        return self._safe(self.dir, raw)

    def safe_thumb(self, raw: str) -> Path | None:
        return self._safe(self.thumbs, raw)

    def count(self) -> int:
        try:
            return sum(1 for f in self.dir.iterdir()
                       if f.is_file() and kind_of(f.name))
        except OSError:
            return 0

    # ---------- Metadaten-Cache ----------

    def _meta_path(self) -> Path:
        return self.thumbs / "meta.json"

    def _load_meta(self) -> dict:
        try:
            return json.loads(self._meta_path().read_text("utf-8"))
        except (OSError, ValueError):
            return {}

    def _save_meta(self, meta: dict) -> None:
        try:
            self._meta_path().write_text(
                json.dumps(meta, ensure_ascii=False, indent=1), "utf-8")
        except OSError as e:
            print(f"[Media] Meta-Cache schreiben fehlgeschlagen: {e}")

    def _duration(self, f: Path) -> float | None:
        try:
            p = Gst.parse_launch(
                f"uridecodebin uri={f.resolve().as_uri()} ! fakesink")
            p.set_state(Gst.State.PAUSED)
            p.get_state(3 * Gst.SECOND)
            ok, dur = p.query_duration(Gst.Format.TIME)
            p.set_state(Gst.State.NULL)
            if ok and dur > 0:
                return round(dur / Gst.SECOND, 2)
        except Exception:  # noqa: BLE001
            pass
        return None

    def make_thumb(self, name: str) -> Path | None:
        src = self.file_path(name)
        if not src.exists():
            return None
        out = self.thumb_path(name)
        if out.exists():
            return out
        got: dict = {}
        try:
            pipeline = Gst.parse_launch(
                f"uridecodebin uri={src.resolve().as_uri()} ! "
                "videoconvert ! videoscale ! "
                "video/x-raw,width=192,height=108 ! "
                "jpegenc quality=75 ! "
                "appsink name=thumb emit-signals=true "
                "max-buffers=1 drop=true")
            sink = pipeline.get_by_name("thumb")

            def on_sample(s):
                sample = s.emit("pull-sample")
                if sample is not None:
                    buf = sample.get_buffer()
                    got["data"] = buf.extract_dup(0, buf.get_size())
                return Gst.FlowReturn.OK

            sink.connect("new-sample", on_sample)
            bus = pipeline.get_bus()
            pipeline.set_state(Gst.State.PLAYING)
            deadline = time.monotonic() + 6.0
            while "data" not in got and time.monotonic() < deadline:
                msg = bus.timed_pop_filtered(
                    100 * Gst.MSECOND,
                    Gst.MessageType.EOS | Gst.MessageType.ERROR)
                if msg is not None:
                    break
                time.sleep(0.02)
            pipeline.set_state(Gst.State.NULL)
        except Exception as e:  # noqa: BLE001
            print(f"[Media] Thumbnail-Fehler ({name}): {e}")
            return None
        if "data" not in got:
            return None
        try:
            out.write_bytes(got["data"])
        except OSError as e:
            print(f"[Media] Thumbnail schreiben fehlgeschlagen: {e}")
            return None
        return out

    # ---------- Öffentliche Operationen ----------

    def scan(self) -> list[dict]:
        self.ensure_dirs()
        meta = self._load_meta()
        items: list[dict] = []
        dirty = False
        for f in sorted(self.dir.iterdir()):
            if not f.is_file() or f.name.endswith(".part"):
                continue
            kind = kind_of(f.name)
            if kind is None:
                continue
            st = f.stat()
            ent = meta.get(f.name)
            if (not isinstance(ent, dict)
                    or ent.get("size") != st.st_size
                    or ent.get("mtime") != int(st.st_mtime)):
                dur = self._duration(f) if kind == "video" else None
                self.make_thumb(f.name)
                ent = {"size": st.st_size, "mtime": int(st.st_mtime),
                       "duration_s": dur}
                meta[f.name] = ent
                dirty = True
            items.append({
                "name": f.name, "size": st.st_size, "type": kind,
                "mtime": ent["mtime"], "duration_s": ent.get("duration_s"),
            })
        if dirty:
            self._save_meta(meta)
        return items

    def _unique(self, base: str) -> str:
        final = base
        n = 1
        while (self.dir / final).exists():
            final = f"{Path(base).stem}-{n}{Path(base).suffix}"
            n += 1
        return final

    def save_stream(self, name: str, src, length: int
                    ) -> tuple[str | None, str | None]:
        """Speichert einen Upload-Stream chunkweise (final_name, error)."""
        base = sanitize(name)
        if kind_of(base) is None:
            return None, "Dateityp nicht unterstützt"
        if length <= 0:
            return None, "Content-Length fehlt"
        limit_mb = int(self.settings["media"].get("max_upload_mb", 2048))
        limit = limit_mb * 1024 * 1024
        if length > limit:
            return None, f"Datei zu groß (max {limit_mb} MB)"
        final = self._unique(base)
        tmp = self.dir / (final + ".part")
        written = 0
        try:
            with open(tmp, "wb") as out:
                while written < length:
                    chunk = src.read(min(1024 * 1024, length - written))
                    if not chunk:
                        break
                    out.write(chunk)
                    written += len(chunk)
        except OSError as e:
            tmp.unlink(missing_ok=True)
            return None, f"IO-Fehler: {e}"
        if written != length:
            tmp.unlink(missing_ok=True)
            return None, f"Upload unvollständig ({written}/{length} Bytes)"
        tmp.rename(self.dir / final)
        try:
            self.make_thumb(final)
        except Exception:  # noqa: BLE001
            pass
        return final, None

    def import_copy(self, src: Path) -> str | None:
        """Importiert eine lokale Datei (CLI-Argument) in die Bibliothek."""
        if not src.exists() or kind_of(src.name) is None:
            return None
        if src.resolve().parent == self.dir:
            return src.name
        final = self._unique(sanitize(src.name))
        try:
            shutil.copy2(src, self.dir / final)
        except OSError as e:
            print(f"[Media] Import fehlgeschlagen ({src}): {e}")
            return None
        self.make_thumb(final)
        return final

    def delete(self, name: str) -> bool:
        p = self.safe_file(name)
        if p is None or not p.exists():
            return False
        try:
            p.unlink()
        except OSError:
            return False
        self.thumb_path(p.name).unlink(missing_ok=True)
        meta = self._load_meta()
        if p.name in meta:
            meta.pop(p.name)
            self._save_meta(meta)
        return True