hms-mediaengine/hms_app/projects.py

"""Projekt-Speicherung: Layer-/FX-Setup als Datei in projects/.

Ein Projekt enthält den kompletten Showzustand (Layer mit Position,
Alpha, Z-Order, Effekte + Intensitaeten, Master) und kann beim
Programmstart automatisch geladen werden (Autostart).
"""

from __future__ import annotations

import json
import re
import time
from pathlib import Path

FORMAT = "hms-project"
VERSION = 1
SUFFIX = ".hms.json"


def sanitize(name: str) -> str:
    """Projektname sicher: keine Pfade/Sonderzeichen, Leerzeichen erlaubt."""
    base = Path(str(name)).name  # schneidet Pfadanteile ab (Traversal-Schutz)
    base = re.sub(r"[^A-Za-z0-9._ -]", "_", base).strip("._ ")
    base = re.sub(r"\s+", " ", base)  # Mehrfach-Leerzeichen zusammenfassen
    return base or "projekt"


class ProjectStore:
    """Dateibasierter Projektspeicher in <root>/projects/."""

    def __init__(self, root: Path):
        self.dir = Path(root) / "projects"

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

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

    # ---------------- Speichern ----------------

    def save(self, name: str, state: dict) -> dict | None:
        """Speichert den Zustand atomar. Rueckgabe: Info-Dict oder None."""
        self._ensure()
        final = sanitize(name)
        if not final:
            return None
        doc = {
            "format": FORMAT,
            "version": VERSION,
            "name": final,
            "saved_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "saved_at_epoch": time.time(),
            "state": state,
        }
        target = self.dir / (final + SUFFIX)
        tmp = target.with_suffix(SUFFIX + ".tmp")
        try:
            tmp.write_text(
                json.dumps(doc, ensure_ascii=False, indent=2), "utf-8")
            tmp.replace(target)
        except OSError as e:
            print(f"[Projects] Schreiben fehlgeschlagen: {e}")
            tmp.unlink(missing_ok=True)
            return None
        return {
            "saved": final,
            "saved_at": doc["saved_at"],
            "layers": len(state.get("layers", [])),
        }

    # ---------------- Laden ----------------

    def load(self, name: str) -> dict | None:
        """Liest ein Projekt. Rueckgabe: geparstes Dokument oder None."""
        p = self.path(name)
        if not p.is_file():
            return None
        try:
            doc = json.loads(p.read_text("utf-8"))
        except (OSError, ValueError) as e:
            print(f"[Projects] Lesefehler ({p.name}): {e}")
            return None
        if not isinstance(doc, dict) or doc.get("format") != FORMAT:
            return None
        return doc

    def exists(self, name: str) -> bool:
        return self.load(name) is not None

    # ---------------- Liste / Loeschen ----------------

    def list(self) -> list[dict]:
        self._ensure()
        items: list[dict] = []
        for f in sorted(self.dir.glob("*" + SUFFIX)):
            try:
                doc = json.loads(f.read_text("utf-8"))
            except (OSError, ValueError):
                continue
            if not isinstance(doc, dict) or doc.get("format") != FORMAT:
                continue
            items.append({
                "name": str(doc.get("name", f.stem)),
                "saved_at": doc.get("saved_at"),
                "layers": len((doc.get("state") or {}).get("layers", [])),
            })
        return items

    def delete(self, name: str) -> bool:
        p = self.path(name)
        if not p.is_file():
            return False
        try:
            p.unlink()
            return True
        except OSError:
            return False