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