274 lines
9.0 KiB
Python
274 lines
9.0 KiB
Python
|
"""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
|