horror/backend/app/main.py

import asyncio
import logging
import time

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import Body, Cookie, FastAPI, HTTPException, Request, Response
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.staticfiles import StaticFiles

from . import db, events, search, streamproxy, suggestions, sync, watched
from .config import IMAGE_BASE, STATIC_DIR

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
# Ein einzelner Zuschauer erzeugt gut zwanzig Stream-Anfragen je Sekunde.
# Jede davon zu protokollieren füllt bei fünfzehn Leuten die Platte und
# sagt über den Betrieb nichts aus. Fehler kommen weiterhin durch.
logging.getLogger("httpx").setLevel(logging.WARNING)


class OhneStreamZugriffe(logging.Filter):
    def filter(self, eintrag):
        return "/stream/" not in eintrag.getMessage()


logging.getLogger("uvicorn.access").addFilter(OhneStreamZugriffe())
log = logging.getLogger("app")

app = FastAPI(title="Netflix Horror DE")
scheduler = AsyncIOScheduler()
sync_lock = asyncio.Lock()

# Der Ereignisstrom darf nicht durch die Kompression, und Video auch nicht:
# GZipMiddleware sammelt die Antwort, bevor sie ausgeht. Bei Server-Sent
# Events kommt dadurch gar nichts an, solange der Puffer nicht voll ist —
# und der wird bei kurzen Ereignissen nie voll. Videosegmente sind bereits
# komprimiert; sie noch einmal durch gzip zu schicken kostet nur Rechenzeit.
OHNE_KOMPRESSION = ("/api/events", "/stream")


class GzipMitAusnahmen(GZipMiddleware):
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http" and scope["path"].startswith(OHNE_KOMPRESSION):
            await self.app(scope, receive, send)
            return
        await super().__call__(scope, receive, send)


# Der Katalog ist reines JSON und schrumpft dabei auf etwa ein Fünftel.
app.add_middleware(GzipMitAusnahmen, minimum_size=1024)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_methods=["*"],
    allow_headers=["*"],
)


async def guarded_sync():
    if sync_lock.locked():
        log.info("Sync läuft bereits")
        return None
    async with sync_lock:
        return await sync.run()


@app.on_event("startup")
async def startup():
    db.init()
    suggestions.init()
    watched.init()
    scheduler.add_job(guarded_sync, "cron", hour=4, minute=30, id="daily-sync")
    scheduler.start()
    if not db.get_meta("last_sync"):
        log.info("Leere Datenbank, starte ersten Sync im Hintergrund")
        asyncio.create_task(guarded_sync())


@app.on_event("shutdown")
async def shutdown():
    await streamproxy.schliessen()


@app.get("/api/health")
def health():
    return {"ok": True}


@app.get("/api/status")
def status():
    return {
        "last_sync": db.get_meta("last_sync"),
        "movie_count": int(db.get_meta("movie_count") or 0),
        "canon_count": int(db.get_meta("canon_count") or 0),
        "syncing": sync_lock.locked(),
        "image_base": IMAGE_BASE,
    }


def katalog_etag():
    """Der Katalog ändert sich nur beim Sync — der Zeitstempel reicht."""
    return f'W/"{db.get_meta("last_sync")}-{db.get_meta("movie_count")}"'


@app.get("/api/movies")
def movies(request: Request, response: Response):
    etag = katalog_etag()
    # Unverändert? Dann spart der Browser sich fünf Megabyte.
    if request.headers.get("if-none-match") == etag:
        return Response(status_code=304, headers={"ETag": etag})
    response.headers["ETag"] = etag
    response.headers["Cache-Control"] = "no-cache"
    return db.all_movies()


@app.get("/api/movies/{movie_id}")
def movie(movie_id: int):
    found = db.get_movie(movie_id)
    if not found:
        raise HTTPException(status_code=404, detail="Film nicht gefunden")
    return found


@app.get("/api/search")
async def search_movies(q: str):
    """Freie TMDB-Suche für Filme außerhalb des Katalogs."""
    if len(q.strip()) < 2:
        return []
    return await search.search(q.strip())


@app.get("/api/search/{movie_id}")
async def search_detail(movie_id: int):
    """Volle Daten zu einem Suchtreffer. Erst der Katalog, dann TMDB."""
    found = db.get_movie(movie_id)
    if found:
        return found
    try:
        return await search.detail(movie_id)
    except Exception:
        raise HTTPException(status_code=404, detail="Film nicht gefunden")


def sitzung(response: Response, sid):
    """Vorhandene Sitzung weiterverwenden, sonst eine neue setzen."""
    if sid:
        return sid
    neu = suggestions.neue_session()
    response.set_cookie(
        suggestions.COOKIE,
        neu,
        max_age=60 * 60 * 24 * 365,
        httponly=True,
        samesite="lax",
    )
    return neu


# Der letzte Dreh, damit eine Seite mitten in der Animation nachziehen kann
letzter_dreh = {"ergebnis": None, "start": 0.0}


@app.get("/api/events")
async def ereignisse(request: Request):
    queue = events.anmelden()
    return StreamingResponse(
        events.stream(queue, request),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.post("/api/spin")
def drehen():
    """Einer dreht, alle sehen dasselbe Ergebnis."""
    laeuft = (
        letzter_dreh["ergebnis"]
        and time.monotonic() - letzter_dreh["start"] < suggestions.DREH_DAUER
    )
    if laeuft:
        return {"started": False, "reason": "dreht bereits"}

    ergebnis = suggestions.ziehen()
    if not ergebnis:
        raise HTTPException(status_code=400, detail="Keine Vorschläge")

    letzter_dreh["ergebnis"] = ergebnis
    letzter_dreh["start"] = time.monotonic()
    events.senden("spin", ergebnis)
    return {"started": True, **ergebnis}


@app.get("/api/spin")
def letzter():
    """Für Seiten, die mitten im Dreh geladen werden."""
    ergebnis = letzter_dreh["ergebnis"]
    if not ergebnis:
        return {"running": False, "result": None}
    vergangen = time.monotonic() - letzter_dreh["start"]
    return {
        "running": vergangen < suggestions.DREH_DAUER,
        # Wer später dazukommt, überspringt den bereits gelaufenen Teil
        "elapsed": round(vergangen, 2),
        "result": ergebnis,
    }


@app.get("/api/suggestions")
def liste_vorschlaege(response: Response, filmabend_sid: str | None = Cookie(None)):
    sid = sitzung(response, filmabend_sid)
    eintraege = suggestions.alle()
    return {
        "items": [
            {k: v for k, v in e.items() if k != "session_id"} for e in eintraege
        ],
        # Damit die Oberfläche den eigenen Vorschlag hervorheben kann
        "mine": next(
            (e["id"] for e in eintraege if e["session_id"] == sid),
            None,
        ),
    }


@app.post("/api/suggestions")
async def vorschlagen(
    response: Response,
    movie_id: int = Body(..., embed=True),
    name: str | None = Body(None, embed=True),
    filmabend_sid: str | None = Cookie(None),
):
    sid = sitzung(response, filmabend_sid)
    movie = await film_holen(movie_id)
    # Ein zweiter Vorschlag ersetzt den ersten, statt abgelehnt zu werden
    eintrag = suggestions.ersetzen(sid, movie, name)
    events.senden("suggestions")
    return {k: v for k, v in eintrag.items() if k != "session_id"}


@app.delete("/api/suggestions")
def vorschlag_loeschen(response: Response, filmabend_sid: str | None = Cookie(None)):
    sid = sitzung(response, filmabend_sid)
    entfernt = suggestions.loeschen(sid)
    if entfernt:
        events.senden("suggestions")
    return {"removed": entfernt}


async def film_holen(movie_id: int):
    """Erst der Katalog, dann TMDB — Suchtreffer stehen nicht in der DB."""
    movie = db.get_movie(movie_id)
    if movie:
        return movie
    try:
        return await search.detail(movie_id)
    except Exception:
        raise HTTPException(status_code=404, detail="Film nicht gefunden")


@app.get("/api/watched")
def liste_gesehen(response: Response, filmabend_sid: str | None = Cookie(None)):
    sid = sitzung(response, filmabend_sid)
    return watched.alle(sid)


@app.post("/api/watched")
async def als_gesehen(movie_id: int = Body(..., embed=True)):
    """Der Host trägt ein, was am Abend lief."""
    eintrag = watched.markieren(await film_holen(movie_id))
    events.senden("watched")
    return eintrag


@app.delete("/api/watched/{watched_id}")
def gesehen_entfernen(watched_id: int):
    entfernt = watched.entfernen(watched_id)
    if entfernt:
        events.senden("watched")
    return {"removed": entfernt}


@app.patch("/api/watched/{watched_id}")
def gesehen_aendern(
    watched_id: int,
    watched_at: str | None = Body(None, embed=True),
    hidden: bool | None = Body(None, embed=True),
):
    """Der Host rückt das Datum nach oder blendet einen Film aus."""
    geaendert = 0
    if watched_at:
        geaendert += watched.datum_setzen(watched_id, watched_at)
    if hidden is not None:
        geaendert += watched.sichtbarkeit(watched_id, hidden)
    if not geaendert:
        raise HTTPException(status_code=404, detail="Film steht nicht in der Liste")
    events.senden("watched")
    return {"ok": True}


@app.post("/api/watched/{watched_id}/rating")
def bewerten(
    watched_id: int,
    response: Response,
    stars: int = Body(..., embed=True),
    name: str | None = Body(None, embed=True),
    filmabend_sid: str | None = Cookie(None),
):
    sid = sitzung(response, filmabend_sid)
    stars = max(1, min(watched.MAX_STERNE, stars))
    eintrag = watched.bewerten(watched_id, sid, stars, name)
    if not eintrag:
        raise HTTPException(status_code=404, detail="Film steht nicht in der Liste")
    events.senden("watched")
    return eintrag


@app.delete("/api/suggestions/alle")
def runde_zuruecksetzen():
    """Host leert die Runde: Vorschläge weg, Rad wieder leer."""
    entfernt = suggestions.alles_loeschen()
    letzter_dreh["ergebnis"] = None
    letzter_dreh["start"] = 0.0
    events.senden("reset")
    return {"removed": entfernt}


@app.get("/stream/{pfad:path}")
async def stream(request: Request, pfad: str):
    """HLS von MediaMTX unter eigenem Ursprung ausliefern."""
    return await streamproxy.weiterleiten(request, pfad)


@app.post("/api/sync")
async def trigger_sync():
    if sync_lock.locked():
        return {"started": False, "reason": "läuft bereits"}
    asyncio.create_task(guarded_sync())
    return {"started": True}


if STATIC_DIR.is_dir():
    app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")