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/watched/rating/{rating_id}") def wertung_loeschen(rating_id: int): """Der Host räumt eine einzelne Wertung weg.""" if not watched.stimme_loeschen(rating_id): raise HTTPException(status_code=404, detail="Wertung nicht gefunden") events.senden("watched") return {"removed": 1} @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")