horror/backend/app/clips.py

"""Clips aus der Aufnahme des Abends.

Eine Marke ist erst nur ein Zeitpunkt: Taste C, fertig, nichts lenkt ab.
Wann immer man will, zieht man ein Fenster darum, sieht es sich an und
schneidet. Dann holt die App den Ausschnitt vom Playback-Server von
MediaMTX und legt ihn als MP4 ab. Jeder sieht nur seine eigenen Clips,
der Host alle — er räumt auf.
"""

from datetime import datetime, timedelta, timezone

import httpx

from .config import CLIPS_DIR, MEDIAMTX_API_URL, MEDIAMTX_PLAYBACK_URL, MTX_SENDEPASSWORT
from .db import connect
from .users import WEM, besitzer, gehoert_mir

SCHEMA = """
CREATE TABLE IF NOT EXISTS clips (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL,
    user_id    INTEGER,
    movie_id   INTEGER,
    marke_at   TEXT NOT NULL,
    vorher     REAL NOT NULL,
    dauer      REAL NOT NULL,
    titel      TEXT,
    datei      TEXT,
    created_at TEXT NOT NULL
);

-- Der Film einer Aufnahme, vom Host gesetzt. Ohne Eintrag gilt der Abend
-- mit dem gleichen Datum — oder keiner.
CREATE TABLE IF NOT EXISTS aufnahme_film (
    start    TEXT PRIMARY KEY,
    movie_id INTEGER NOT NULL
);
"""

STREAM_PFAD = "filmabend"
# Der Player hängt so weit hinter der Live-Kante — die Marke zeigt auf
# das, was die Person gerade gesehen hat, nicht auf das, was gerade ankommt
PLAYER_LATENZ = 4
# Das Fenster um die Marke: Standard und Grenzen
VORHER = 30
DAUER = 35
# Die Marke ist nur ein Anker — das Fenster darf überall auf dem
# Zeitstrahl liegen, auch weit nach der Marke (vorher negativ)
MAX_VORHER = 4 * 3600
MIN_DAUER = 3
MAX_DAUER = 180
MAX_TITEL = 80


def init():
    with connect() as conn:
        conn.executescript(SCHEMA)


def _jetzt():
    return datetime.now(timezone.utc)


def _iso(zeit):
    return zeit.isoformat(timespec="seconds")


def markieren(session_id, movie_id, zeit=None):
    """Eine Marke setzen — vier Sekunden vor jetzt, wegen der Latenz.
    Oder zu einem gegebenen Zeitpunkt, für Clips aus alten Aufnahmen."""
    with connect() as conn:
        clip_id = conn.execute(
            "INSERT INTO clips (session_id, user_id, movie_id, marke_at, vorher, dauer, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                session_id,
                besitzer(session_id),
                movie_id,
                _iso(zeit or _jetzt() - timedelta(seconds=PLAYER_LATENZ)),
                VORHER,
                DAUER,
                _iso(_jetzt()),
            ),
        ).lastrowid
    # Erst nach dem Commit lesen — sonst sieht die neue Verbindung nichts
    return holen(clip_id, session_id)


def _oeffentlich(row, session_id, ich):
    clip = {k: v for k, v in dict(row).items() if k not in ("session_id", "user_id", "freigegeben")}
    clip["geschnitten"] = clip.pop("datei") is not None
    clip["mine"] = gehoert_mir(row, session_id, ich)
    return clip


def holen(clip_id, session_id):
    with connect() as conn:
        row = conn.execute(
            "SELECT c.*, u.name, m.title, m.poster_path FROM clips c "
            "LEFT JOIN users u ON u.id = c.user_id "
            "LEFT JOIN movies m ON m.id = c.movie_id WHERE c.id = ?",
            (clip_id,),
        ).fetchone()
    return _oeffentlich(row, session_id, besitzer(session_id)) if row else None


def alle(session_id, host=False):
    """Die eigenen Clips; der Host sieht alle — er räumt auf."""
    ich = besitzer(session_id)
    with connect() as conn:
        rows = conn.execute(
            "SELECT c.*, u.name, m.title, m.poster_path FROM clips c "
            "LEFT JOIN users u ON u.id = c.user_id "
            "LEFT JOIN movies m ON m.id = c.movie_id "
            "ORDER BY c.marke_at DESC"
        ).fetchall()
    return [
        _oeffentlich(r, session_id, ich)
        for r in rows
        if host or gehoert_mir(r, session_id, ich)
    ]


def _zeile(conn, clip_id):
    return conn.execute("SELECT * FROM clips WHERE id = ?", (clip_id,)).fetchone()


def _darf(row, session_id, host):
    return row is not None and (host or gehoert_mir(row, session_id, besitzer(session_id)))


def aendern(clip_id, session_id, host, titel=None, vorher=None, dauer=None):
    """Titel oder Fenster. Ein neues Fenster macht den Schnitt ungültig —
    die Datei fliegt, bis neu geschnitten wird."""
    with connect() as conn:
        row = _zeile(conn, clip_id)
        if not _darf(row, session_id, host):
            return None
        felder = {}
        if titel is not None:
            felder["titel"] = titel.strip()[:MAX_TITEL] or None
        if vorher is not None:
            felder["vorher"] = max(-MAX_VORHER, min(MAX_VORHER, float(vorher)))
        if dauer is not None:
            felder["dauer"] = max(MIN_DAUER, min(MAX_DAUER, float(dauer)))
        fenster_neu = any(
            k in felder and felder[k] != row[k] for k in ("vorher", "dauer")
        )
        if fenster_neu and row["datei"]:
            _datei_weg(row["datei"])
            felder["datei"] = None
        if felder:
            setzt = ", ".join(f"{k} = ?" for k in felder)
            conn.execute(f"UPDATE clips SET {setzt} WHERE id = ?", (*felder.values(), clip_id))
    return holen(clip_id, session_id)


def loeschen(clip_id, session_id, host):
    with connect() as conn:
        row = _zeile(conn, clip_id)
        if not _darf(row, session_id, host):
            return False
        if row["datei"]:
            _datei_weg(row["datei"])
        conn.execute("DELETE FROM clips WHERE id = ?", (clip_id,))
    return True


def _datei_weg(name):
    pfad = CLIPS_DIR / name
    if pfad.exists():
        pfad.unlink()


def fenster(row, vorher=None, dauer=None):
    """Start (RFC3339) und Dauer des Ausschnitts um die Marke."""
    marke = datetime.fromisoformat(row["marke_at"])
    v = row["vorher"] if vorher is None else max(-MAX_VORHER, min(MAX_VORHER, float(vorher)))
    d = row["dauer"] if dauer is None else max(MIN_DAUER, min(MAX_DAUER, float(dauer)))
    return _iso(marke - timedelta(seconds=v)), d


def playback_url(row, vorher=None, dauer=None):
    start, d = fenster(row, vorher, dauer)
    return (
        f"{MEDIAMTX_PLAYBACK_URL}/get",
        {"path": STREAM_PFAD, "start": start, "duration": d, "format": "mp4"},
    )


def zeile(clip_id, session_id, host):
    """Die rohe Zeile, wenn man sie sehen darf: Besitzer oder Host."""
    with connect() as conn:
        row = _zeile(conn, clip_id)
    return row if _darf(row, session_id, host) else None


async def schneiden(clip_id, session_id, host):
    """Den Ausschnitt vom Playback-Server holen und als Datei ablegen.
    Gibt None zurück, wenn der Clip nicht gehört; wirft httpx-Fehler,
    wenn MediaMTX nicht liefert."""
    with connect() as conn:
        row = _zeile(conn, clip_id)
    if not _darf(row, session_id, host):
        return None
    url, params = playback_url(row)
    CLIPS_DIR.mkdir(parents=True, exist_ok=True)
    name = f"{clip_id}.mp4"
    ziel = CLIPS_DIR / name
    async with httpx.AsyncClient(timeout=120.0) as http:
        async with http.stream("GET", url, params=params) as resp:
            resp.raise_for_status()
            with open(ziel, "wb") as f:
                async for stueck in resp.aiter_bytes():
                    f.write(stueck)
    with connect() as conn:
        conn.execute("UPDATE clips SET datei = ? WHERE id = ?", (name, clip_id))
    return holen(clip_id, session_id)


# Ein Abend dauert ein paar Stunden — so weit reicht der Zeitstrahl
SPANNE = 4 * 3600


async def _segmente(von, bis):
    async with httpx.AsyncClient(timeout=15.0) as http:
        resp = await http.get(
            f"{MEDIAMTX_PLAYBACK_URL}/list",
            params={"path": STREAM_PFAD, "start": _iso(von), "end": _iso(bis)},
        )
        resp.raise_for_status()
        return resp.json()


# ── Aufnahmen: die Stücke aus MediaMTX, jedes mit seinem Film ──────────
# Eine Aufnahme hängt an keinem Abend. Der Host weist ihr einen Film zu;
# ohne Zuweisung gilt der Abend mit dem gleichen Datum, sonst keiner.


def _stueck(s):
    return {
        "start": _iso(datetime.fromisoformat(s["start"].replace("Z", "+00:00"))),
        "dauer": round(s["duration"], 1),
    }


def _abendtag(iso):
    """Der Abend reicht bis zum nächsten Mittag — die Nacht gehört dazu."""
    zeit = datetime.fromisoformat(iso.replace("Z", "+00:00"))
    return (zeit - timedelta(hours=12)).strftime("%Y-%m-%d")


def film_zuweisen(start, movie_id):
    """Der Host hängt einen Film an eine Aufnahme — oder nimmt ihn weg."""
    with connect() as conn:
        if movie_id is None:
            conn.execute("DELETE FROM aufnahme_film WHERE start = ?", (start,))
        else:
            conn.execute(
                "INSERT OR REPLACE INTO aufnahme_film (start, movie_id) VALUES (?, ?)",
                (start, movie_id),
            )


def _filme(conn):
    """Zuweisungen je Start und Abende je Tag — um Aufnahmen ihren Film zu geben."""
    zugewiesen = {
        r["start"]: {"id": r["movie_id"], "title": r["title"], "poster_path": r["poster_path"]}
        for r in conn.execute(
            "SELECT a.start, a.movie_id, m.title, m.poster_path FROM aufnahme_film a "
            "LEFT JOIN movies m ON m.id = a.movie_id"
        )
    }
    abende = {}
    for w in conn.execute(
        "SELECT movie_id, title, poster_path, watched_at FROM watched ORDER BY id"
    ):
        abende.setdefault(
            _abendtag(w["watched_at"]),
            {"id": w["movie_id"], "title": w["title"], "poster_path": w["poster_path"]},
        )
    return zugewiesen, abende


def _mit_film(stueck, zugewiesen, abende):
    film = zugewiesen.get(stueck["start"])
    return {
        **stueck,
        "film": film or abende.get(_abendtag(stueck["start"])),
        "zugewiesen": film is not None,
    }


async def aufnahmen_alle():
    """Alle Aufnahmen, neueste zuerst, jede mit ihrem Film."""
    segmente = await _segmente(datetime(2000, 1, 1, tzinfo=timezone.utc), _jetzt())
    with connect() as conn:
        zugewiesen, abende = _filme(conn)
    stuecke = [_mit_film(_stueck(s), zugewiesen, abende) for s in segmente]
    return sorted(stuecke, key=lambda s: s["start"], reverse=True)


def film_der_aufnahme(start):
    """Nur die id — für neue Marken."""
    with connect() as conn:
        zugewiesen, abende = _filme(conn)
    film = _mit_film({"start": start}, zugewiesen, abende)["film"]
    return film["id"] if film else None


async def film_live():
    """Der Film der gerade laufenden Aufnahme — für die Taste C. None,
    wenn nichts aufgenommen wird oder MediaMTX nicht antwortet."""
    try:
        segmente = await _segmente(_jetzt() - timedelta(hours=6), _jetzt())
    except httpx.HTTPError:
        return None
    for s in segmente:
        stueck = _stueck(s)
        ende = datetime.fromisoformat(stueck["start"]) + timedelta(seconds=stueck["dauer"])
        if ende >= _jetzt() - timedelta(seconds=90):
            return film_der_aufnahme(stueck["start"])
    return None


def _im_stueck(segmente, start, dauer):
    """Die Segmente (aus der API), die zu einem Stück der Aufnahme gehören:
    ihr Beginn liegt zwischen dessen Start und Ende. Eine Sekunde Luft an
    beiden Enden — Playback rundet die Zeiten anders als die API."""
    von = datetime.fromisoformat(start) - timedelta(seconds=1)
    bis = datetime.fromisoformat(start) + timedelta(seconds=dauer + 1)
    aus = []
    for s in segmente:
        beginn = datetime.fromisoformat(s["start"].replace("Z", "+00:00"))
        if von <= beginn <= bis:
            aus.append(s["start"])
    return aus


async def aufnahme_loeschen(start, dauer):
    """Ein Stück der Aufnahme löschen — MediaMTX räumt die Segmente weg.
    Gibt zurück, wie viele es waren."""
    auth = ("sender", MTX_SENDEPASSWORT)
    async with httpx.AsyncClient(timeout=30.0, auth=auth) as http:
        resp = await http.get(f"{MEDIAMTX_API_URL}/v3/recordings/get/{STREAM_PFAD}")
        resp.raise_for_status()
        starts = _im_stueck(resp.json().get("segments") or [], start, dauer)
        for s in starts:
            resp = await http.delete(
                f"{MEDIAMTX_API_URL}/v3/recordings/deletesegment",
                params={"path": STREAM_PFAD, "start": s},
            )
            resp.raise_for_status()
    return len(starts)


async def aufnahme(um, spanne=SPANNE):
    """Welche Stücke der Aufnahme es um einen Zeitpunkt herum gibt — für
    den Zeitstrahl im Editor. Relativ zur Marke, in Sekunden."""
    marke = datetime.fromisoformat(um)
    segmente = await _segmente(marke - timedelta(seconds=spanne), marke + timedelta(seconds=spanne))
    aus = []
    for s in segmente:
        start = datetime.fromisoformat(s["start"].replace("Z", "+00:00"))
        von = (start - marke).total_seconds()
        aus.append({"von": round(von, 1), "bis": round(von + s["duration"], 1)})
    return aus