from fastapi import FastAPI, HTTPException, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, StreamingResponse from pydantic import BaseModel from typing import List, Optional, Dict from pathlib import Path import asyncio import json import os import shutil import uuid import db import devices as device_discovery import mounts as mount_verwaltung import notify from celery_client import celery_client, start_rip from detection import CDS_DISC_OK, CDS_NO_DISC, CDS_TRAY_OPEN, drive_status from config_validation import validate_config, ConfigValidationError from cache import init_cache from ratelimit import check_rate_limit, get_rate_limit_remaining from prescan import PreScan from nfo_generator import NFOGenerator from image_downloader import ImageDownloader # Auth (JWT/Login/API-Keys) KOMPLETT entfernt — Commander-Entscheid 24.07.2026: # Rippy läuft ausschließlich im Heimnetz, die Endpoints schützten ohnehin # nichts (kein Login-Flow im UI) und waren damit Placebo-Oberfläche. # Dokumentiert in KONZEPT.md Abschnitt 10. app = FastAPI( title="Rippy API", description="API für das automatische Ripping-System", version="1.0.0" ) @app.on_event("startup") async def startup_event(): """Initialisiere Cache + Datenbank, validiere Konfiguration, starte Disc-Wache.""" init_cache() db.init_db() try: validate_config() except ConfigValidationError as e: print(f"⚠️ Konfigurations-Warnung: {e}") # Gespeicherte Netzwerk-Speicherziele wiederherstellen def remount(): for meldung in mount_verwaltung.alle_remounten(): db.add_log("info", "mounts", meldung) await asyncio.to_thread(remount) asyncio.create_task(disc_watcher()) # Auto-Pre-Scan-Ergebnisse je Laufwerk: das Dashboard zeigt damit sofort, # WAS im Laufwerk liegt (Titel/Jahr/Poster), ohne dass jemand klicken muss. DISC_CACHE: Dict[str, Dict] = {} def _duplikat_suchen(fingerprint: str): """Wurde eine Disc mit diesem Fingerabdruck schon erfolgreich gerippt?""" if not fingerprint: return None for job in db.list_jobs(200): if job.get("status") != "completed": continue try: meta = json.loads(job.get("meta") or "{}") except ValueError: continue if meta.get("fingerprint") == fingerprint: return { "job_id": job["id"], "title": job.get("title"), "finished_at": job["finished_at"].isoformat() if job.get("finished_at") else None, } return None async def _auto_prescan(pfad: str): """Identifiziert die eingelegte Disc im Hintergrund und cached das Ergebnis.""" if DISC_CACHE.get(pfad, {}).get("_laeuft"): return DISC_CACHE[pfad] = {"_laeuft": True, "title": "Wird erkannt…"} try: prescan = PreScan() ergebnis = await asyncio.to_thread(prescan.scan, pfad) DISC_CACHE[pfad] = ergebnis.to_dict() jahr_text = f" ({ergebnis.year})" if ( ergebnis.year and not ergebnis.title.endswith(f"({ergebnis.year})") ) else "" db.add_log( "info", "watcher", f"Disc erkannt: {ergebnis.title}{jahr_text}" + f" [{ergebnis.disc_type}, Confidence {ergebnis.confidence:.0%}] auf {pfad}", ) # Duplikat-Warnung: dieselbe Disc (Fingerabdruck) schon fertig gerippt? dup = await asyncio.to_thread(_duplikat_suchen, ergebnis.fingerprint) if dup: DISC_CACHE[pfad]["bereits_gerippt"] = dup db.add_log( "warning", "watcher", f"Diese Disc wurde bereits gerippt ({dup.get('title')}, " f"Job {dup['job_id'][:8]}…) — Dashboard zeigt den Hinweis.", ) await _auto_rip_wenn_aktiviert(pfad) except Exception as e: DISC_CACHE.pop(pfad, None) print(f"Auto-Pre-Scan {pfad}: {e}") async def _auto_rip_wenn_aktiviert(pfad: str): """Vollautomatik (Setting autoRipStart): Disc erkannt → Rip startet sofort. Commander-Wunsch 24.07.: wahlweise Popup ODER Automatik. Ziel-Ordner kommt aus den Schnellwahl-Einstellungen (Serie → seriesDir, sonst movieDir; CD → musicDir) — genau wie ein Klick im Dialog. """ einstellungen = await asyncio.to_thread(db.get_settings) if not einstellungen.get("autoRipStart"): return if await asyncio.to_thread(db.has_active_job, pfad): return disc = DISC_CACHE.get(pfad) or {} if disc.get("_laeuft"): return if disc.get("bereits_gerippt"): await asyncio.to_thread( db.add_log, "info", "api", "Automatik übersprungen: Disc wurde bereits gerippt — " 'manuell per „Rippen starten" trotzdem möglich.', ) return basis = einstellungen.get("outputDir") or MEDIA_ROOT meta = disc.get("metadata") or {} if disc.get("disc_type") == "CD": unterordner = einstellungen.get("musicDir") or "music" elif meta.get("type") == "tv": unterordner = einstellungen.get("seriesDir") or "series" else: unterordner = einstellungen.get("movieDir") or "movies" ziel = os.path.normpath(os.path.join(basis, unterordner)) if not ziel.startswith(MEDIA_ROOT): ziel = None job_id = str(uuid.uuid4()) meta_json = json.dumps({ "year": disc.get("year"), "confidence": disc.get("confidence"), **meta, }) await asyncio.to_thread( db.insert_job, job_id, pfad, None, disc.get("title"), ziel, meta_json ) await asyncio.to_thread( db.add_log, "info", "api", f'Automatik: Rip für „{disc.get("title")}" gestartet ({pfad} → {ziel})', ) start_rip(pfad, job_id, ziel) async def disc_watcher(): """Disc-Wache: pollt die Laufwerke, protokolliert Einwurf/Auswurf und stößt beim Einlegen automatisch den Pre-Scan an (Dashboard-Disc-Karte). Ersetzt den nie gebauten udev-Daemon aus dem KONZEPT: udev funktioniert im Container nicht sinnvoll (kein udevd) — ein 3-Sekunden-Poll per ioctl ist für den Heim-Use-Case gleichwertig und läuft überall. """ bekannt: Dict[str, int] = {} while True: try: for pfad in device_discovery.list_optical_devices(): try: status = await asyncio.to_thread(drive_status, pfad) except OSError: continue vorher = bekannt.get(pfad) if vorher is None: # Erststart: liegt schon eine Disc drin, direkt erkennen if status == CDS_DISC_OK: asyncio.create_task(_auto_prescan(pfad)) elif status != vorher: if status == CDS_DISC_OK: db.add_log("info", "watcher", f"Disc eingelegt: {pfad}") asyncio.create_task(_auto_prescan(pfad)) elif status in (CDS_NO_DISC, CDS_TRAY_OPEN) and vorher == CDS_DISC_OK: db.add_log("info", "watcher", f"Disc entfernt: {pfad}") DISC_CACHE.pop(pfad, None) bekannt[pfad] = status except Exception as e: print(f"Disc-Wache: {e}") await asyncio.sleep(3) # Middleware für Rate-Limiting (pro Client-IP — schützt vor Amok-Skripten, # nicht vor Angreifern; Rippy ist Heimnetz-only) @app.middleware("http") async def rate_limit_middleware(request: Request, call_next): """Rate-Limiting Middleware.""" client_ip = request.client.host # Rate Limit prüfen if not check_rate_limit(client_ip): return Response( content=json.dumps({"error": "Rate limit exceeded"}), status_code=429, media_type="application/json" ) response = await call_next(request) # Füge Rate-Limit Header hinzu remaining = get_rate_limit_remaining(client_ip) response.headers["X-RateLimit-Remaining"] = str(remaining) return response # CORS hinzufügen app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class Job(BaseModel): id: str type: str status: str device: str startTime: str endTime: Optional[str] = None progress: int = 0 title: Optional[str] = None error: Optional[str] = None can_retry: bool = False # Rohdaten vorhanden → „Neu komprimieren" sinnvoll class Device(BaseModel): id: str name: str type: str path: str status: str model: Optional[str] = None serial: Optional[str] = None disc: Optional[Dict] = None # Auto-Pre-Scan-Ergebnis (Titel/Jahr/Poster) def _job_row_to_model(zeile: dict) -> Job: """DB-Zeile → UI-Form (Worker-Status 'running' heißt im UI 'processing').""" status_map = {"running": "processing"} return Job( id=zeile["id"], type=zeile.get("disc_type") or "unknown", status=status_map.get(zeile["status"], zeile["status"]), device=zeile.get("device") or "", startTime=zeile["created_at"].isoformat() if zeile.get("created_at") else "", endTime=zeile["finished_at"].isoformat() if zeile.get("finished_at") else None, progress=zeile.get("progress") or 0, title=zeile.get("title"), error=zeile.get("error"), ) @app.get("/health") async def health_check(): return {"status": "ok", "service": "api"} @app.get("/") async def root(): return { "name": "Rippy", "version": "1.0.0", "description": "Automatisches Ripping-System für CD, DVD und Blu-ray" } def _kann_neu_komprimieren(job: dict, work_dir: str) -> bool: """Nur wenn Rohdaten wirklich noch daliegen — der „Neu komprimieren"-Knopf an einem Job, der nie gerippt hat, war Unsinn (Befund 24.07.).""" if job.get("status") != "failed": return False if os.path.isdir(os.path.join("/app/temp/raw", job["id"])): return True return work_dir.startswith(MEDIA_ROOT) and os.path.isdir(os.path.join(work_dir, job["id"])) @app.get("/jobs", response_model=List[Job]) async def get_jobs(): """Holt alle Jobs aus der Datenbank (neueste zuerst).""" def sammle(): work_dir = os.path.normpath((db.get_settings().get("workDir") or "").strip() or "/") modelle = [] for z in db.list_jobs(): modell = _job_row_to_model(z) modell.can_retry = _kann_neu_komprimieren(z, work_dir) modelle.append(modell) return modelle return await asyncio.to_thread(sammle) @app.delete("/jobs/{job_id}") async def delete_job(job_id: str): """Entfernt einen erledigten Job aus der Liste (Dateien bleiben liegen).""" job = await asyncio.to_thread(db.get_job, job_id) if not job: raise HTTPException(status_code=404, detail="Job nicht gefunden") if job["status"] not in ("completed", "failed"): raise HTTPException(status_code=409, detail="Job läuft noch — erst abbrechen") await asyncio.to_thread(db.delete_job, job_id) await asyncio.to_thread(db.add_log, "info", "api", f"Job {job_id} aus der Liste entfernt") return {"status": "deleted"} @app.delete("/jobs") async def delete_finished_jobs(): """Räumt ALLE erledigten Jobs (fertig + fehlgeschlagen) aus der Liste.""" anzahl = await asyncio.to_thread(db.delete_finished_jobs) await asyncio.to_thread( db.add_log, "info", "api", f"Job-Liste aufgeräumt ({anzahl} erledigte Einträge entfernt)" ) return {"deleted": anzahl} class JobCreateRequest(BaseModel): device_path: Optional[str] = None device: Optional[str] = None # Alias, so schickt es das UI title: Optional[str] = None target_dir: Optional[str] = None # Ablageziel unter /app/media (frei wählbar) series: Optional[str] = None # Serien-Flow: Ablage /Season NN season: Optional[int] = None main_feature_only: Optional[bool] = None # pro Rip; None = Setting gilt titles: Optional[List[int]] = None # exakte Titel-Auswahl (Track-Tabelle) MEDIA_ROOT = "/app/media" def _validiere_ziel(target_dir: Optional[str]) -> Optional[str]: """Ziel muss unter /app/media liegen — Pfad-Ausbrüche (..) fliegen raus.""" if not target_dir: return None normalisiert = os.path.normpath(target_dir) if not normalisiert.startswith(MEDIA_ROOT): raise HTTPException( status_code=422, detail=f"Ziel muss unter {MEDIA_ROOT} liegen (Shares dort einhängen)", ) return normalisiert @app.post("/jobs", status_code=201) async def create_job(request: JobCreateRequest): """Legt einen Rip-Job an und schickt ihn an den Worker. Das war DIE fehlende Stelle: bis 23.07. gab es keinerlei Code-Pfad, der je einen Rip ausgelöst hat. """ device_path = request.device_path or request.device if not device_path: raise HTTPException(status_code=422, detail="device_path fehlt") if device_path not in device_discovery.list_optical_devices(): raise HTTPException(status_code=404, detail=f"Laufwerk {device_path} nicht gefunden") ziel = _validiere_ziel(request.target_dir) # Titel + Metadaten aus der Disc-Erkennung übernehmen — der Worker nutzt # sie für den Ordnernamen und die Media-Server-Aufbereitung (NFO/Poster), # das UI fürs Job-Detail-Popup. Serien-Flow und Hauptfilm-Wahl wandern # ebenfalls in die Job-Metadaten. titel = request.title meta_dict = {} disc = DISC_CACHE.get(device_path) if disc and not disc.get("_laeuft"): if not titel: titel = disc.get("title") meta_dict = { "year": disc.get("year"), "confidence": disc.get("confidence"), "fingerprint": disc.get("fingerprint"), **(disc.get("metadata") or {}), } if request.series and request.series.strip(): meta_dict["series"] = request.series.strip() meta_dict["season"] = max(1, int(request.season or 1)) if request.main_feature_only is not None: meta_dict["main_feature_only"] = request.main_feature_only if request.titles: titel_liste = sorted({int(t) for t in request.titles if int(t) >= 0})[:200] if titel_liste: meta_dict["titles"] = titel_liste meta_json = json.dumps(meta_dict) if meta_dict else None job_id = str(uuid.uuid4()) await asyncio.to_thread(db.insert_job, job_id, device_path, None, titel, ziel, meta_json) await asyncio.to_thread( db.add_log, "info", "api", f"Job {job_id} angelegt für {device_path}" + (f" → {ziel}" if ziel else ""), ) start_rip(device_path, job_id, ziel) return {"id": job_id, "status": "pending", "device": device_path, "target_dir": ziel} @app.get("/jobs/{job_id}/detail") async def get_job_detail(job_id: str): """Alles zu EINEM Job — fürs Klick-Popup auf den Titel in „Neueste Jobs": Metadaten (Poster/Jahr/Beschreibung), Ziel, Ausgabepfad, Fehler.""" job = await asyncio.to_thread(db.get_job, job_id) if not job: raise HTTPException(status_code=404, detail="Job nicht gefunden") detail = _job_row_to_model(job).dict() detail["target_dir"] = job.get("target_dir") detail["output_path"] = job.get("output_path") try: detail["meta"] = json.loads(job["meta"]) if job.get("meta") else None except ValueError: detail["meta"] = None return detail def _sicherer_dateiname(name: str) -> bool: """Pure Funktion (testbar): nur nackte Dateinamen, keine Pfad-Tricks.""" return bool(name) and "/" not in name and "\\" not in name and not name.startswith(".") def _job_ausgabeordner(job: dict) -> str: """Validierter Ausgabeordner eines Jobs — strikt unter /app/media.""" ausgabe = os.path.normpath(job.get("output_path") or "") if not ausgabe.startswith(MEDIA_ROOT): raise HTTPException(status_code=404, detail="Job hat keinen Ausgabeordner unter /app/media") return ausgabe @app.get("/jobs/{job_id}/files") async def list_job_files(job_id: str): """Dateien eines fertigen Jobs — fürs Download-Menü im Dashboard. Vorher kam man an fertige MKVs nur per scp auf die VM. """ job = await asyncio.to_thread(db.get_job, job_id) if not job: raise HTTPException(status_code=404, detail="Job nicht gefunden") ausgabe = _job_ausgabeordner(job) def liste(): try: eintraege = sorted(os.listdir(ausgabe)) except OSError: return None dateien = [] for name in eintraege: pfad = os.path.join(ausgabe, name) if os.path.isfile(pfad): try: groesse_mb = round(os.path.getsize(pfad) / 1024**2, 1) except OSError: groesse_mb = None dateien.append({"name": name, "size_mb": groesse_mb}) return dateien dateien = await asyncio.to_thread(liste) if dateien is None: raise HTTPException( status_code=404, detail="Ausgabeordner nicht lesbar — Job noch nicht fertig oder Ziel ausgehängt?", ) return {"job_id": job_id, "output_path": ausgabe, "files": dateien} @app.get("/jobs/{job_id}/files/{dateiname}") async def download_job_file(job_id: str, dateiname: str): """Streamt EINE Datei eines Jobs zum Browser (Download-Knopf). Pfad-Validierung strikt: nackter Dateiname, realpath muss unter /app/media bleiben (kein ..-Ausbruch, kein Symlink nach draußen). """ job = await asyncio.to_thread(db.get_job, job_id) if not job: raise HTTPException(status_code=404, detail="Job nicht gefunden") ausgabe = _job_ausgabeordner(job) if not _sicherer_dateiname(dateiname): raise HTTPException(status_code=422, detail="Ungültiger Dateiname") pfad = os.path.join(ausgabe, dateiname) def pruefe(): return os.path.isfile(pfad) and os.path.realpath(pfad).startswith(MEDIA_ROOT) if not await asyncio.to_thread(pruefe): raise HTTPException(status_code=404, detail="Datei nicht gefunden") return FileResponse(pfad, filename=dateiname, media_type="application/octet-stream") @app.get("/storage-targets") async def storage_targets(): """Verfügbare Ablageziele: Verzeichnisse unter /app/media inkl. Mounts. NFS/SMB-Shares, die auf der VM unter /srv/rippy/media eingehängt werden, tauchen hier automatisch auf (rslave-Bind in docker-compose). """ def sammle(): ziele = [] try: eintraege = sorted(os.listdir(MEDIA_ROOT)) except OSError: return ziele for name in eintraege: pfad = os.path.join(MEDIA_ROOT, name) if not os.path.isdir(pfad): continue try: nutzung = shutil.disk_usage(pfad) frei_gb = round(nutzung.free / 1024**3, 1) except OSError: frei_gb = None ziele.append({ "name": name, "path": pfad, "is_mount": os.path.ismount(pfad), "free_gb": frei_gb, }) return ziele return await asyncio.to_thread(sammle) @app.post("/devices/{name}/eject") async def eject_device(name: str): """Wirft die Disc aus. Verweigert, wenn auf dem Gerät gerade ein Job läuft.""" device_path = f"/dev/{name}" if device_path not in device_discovery.list_optical_devices(): raise HTTPException(status_code=404, detail=f"Laufwerk {device_path} nicht gefunden") if await asyncio.to_thread(db.has_active_job, device_path): raise HTTPException( status_code=409, detail="Auf diesem Laufwerk läuft gerade ein Job" ) try: await asyncio.to_thread(device_discovery.eject, device_path) except OSError as e: raise HTTPException(status_code=500, detail=f"Auswurf fehlgeschlagen: {e}") await asyncio.to_thread(db.add_log, "info", "api", f"Disc ausgeworfen: {device_path}") return {"status": "ejected", "device": device_path} @app.post("/devices/{name}/scan-tracks") async def scan_tracks_starten(name: str): """Titel-Scan der eingelegten Disc anstoßen (Track-Auswahl-Tabelle). Läuft als Worker-Task (nur der hat makemkvcon + Laufwerk); das UI pollt GET /devices/{name}/tracks. Dauert je nach Disc 20–120 s. """ device_path = f"/dev/{name}" if device_path not in device_discovery.list_optical_devices(): raise HTTPException(status_code=404, detail=f"Laufwerk {device_path} nicht gefunden") if await asyncio.to_thread(db.has_active_job, device_path): raise HTTPException(status_code=409, detail="Auf diesem Laufwerk läuft gerade ein Job") await asyncio.to_thread(db.save_settings, {"status": "running"}, f"tracks:{device_path}") celery_client.send_task("worker.tasks.scan_tracks", args=[device_path]) return {"status": "scanning"} @app.get("/devices/{name}/tracks") async def scan_tracks_ergebnis(name: str): """Ergebnis des Titel-Scans (Polling-Ziel des UI).""" device_path = f"/dev/{name}" daten = await asyncio.to_thread(db.get_settings, f"tracks:{device_path}") if not daten: return {"status": "none"} return daten @app.post("/jobs/{job_id}/retry-transcode") async def retry_transcode(job_id: str): """Stößt die Kompression eines Jobs neu an — OHNE die Disc neu zu rippen. Voraussetzung: die Rohdateien liegen noch in /app/temp/raw/ (bei Kompressions-Fehlschlägen bleiben sie dort absichtlich erhalten). """ job = await asyncio.to_thread(db.get_job, job_id) if not job: raise HTTPException(status_code=404, detail="Job nicht gefunden") if job["status"] in ("running", "pending"): raise HTTPException(status_code=409, detail="Job rippt noch") # Roh-Verzeichnis: respektiert das konfigurierbare Arbeitsverzeichnis # (Einstellungen → Verarbeitung), sonst Container-Default /app/temp/raw. einstellungen = await asyncio.to_thread(db.get_settings) work_dir = os.path.normpath((einstellungen.get("workDir") or "").strip() or "/") raw_basis = work_dir if work_dir.startswith(MEDIA_ROOT) else "/app/temp/raw" raw_dir = f"{raw_basis}/{job_id}" # Zielordner: der Worker schreibt das geplante Ziel beim Rip-Start nach # output_path (sprechender Name statt UUID) — alter Fallback bleibt. basis = job.get("target_dir") or f"{MEDIA_ROOT}/{job.get('disc_type') or 'bluray'}" final_dir = job.get("output_path") or f"{basis}/{job_id}" celery_client.send_task( "worker.tasks.transcode_files", args=[job_id, raw_dir, final_dir], queue="transcode", ) await asyncio.to_thread(db.update_job, job_id, status="transcoding", progress=0, error=None) await asyncio.to_thread(db.add_log, "info", "api", f"Job {job_id}: Kompression neu eingereiht") return {"id": job_id, "status": "transcoding"} @app.post("/jobs/{job_id}/cancel") async def cancel_job(job_id: str): """Bittet den Worker, den Job abzubrechen (kooperativ über die DB). Der Worker prüft das Flag bei jedem Fortschritts-Update und beendet den Encoder-Prozess sauber — kein Celery-Task-ID-Tracking nötig. """ job = await asyncio.to_thread(db.get_job, job_id) if not job: raise HTTPException(status_code=404, detail="Job nicht gefunden") if job["status"] in ("completed", "failed"): raise HTTPException(status_code=409, detail="Job ist bereits beendet") await asyncio.to_thread(db.update_job, job_id, status="canceling") await asyncio.to_thread(db.add_log, "warning", "api", f"Job {job_id}: Abbruch angefordert") return {"id": job_id, "status": "canceling"} @app.get("/capabilities") async def capabilities(): """Welche Encoder sind auf welchen Workern WIRKLICH verfügbar — inkl. Live-Erreichbarkeit (Celery-Ping + Herzschlag-Alter). Jeder Worker meldet sich selbst (caps.py, minütlich) — auch optionale Remote-GPU-Worker tauchen hier automatisch auf. """ def sammle(): zeilen = db.list_workers() try: antworten = celery_client.control.ping(timeout=1.0) or [] online_namen = { knoten.split("@", 1)[-1] for antwort in antworten for knoten in antwort.keys() } except Exception: online_namen = set() for zeile in zeilen: # Celery-Ping meldet den HOSTNAME des Knotens — bei gesetztem # WORKER_NAME (Anzeigename) steckt der echte Hostname in info. hostname = (zeile.get("info") or {}).get("hostname") or zeile["name"] zeile["online"] = hostname in online_namen return zeilen return {"workers": await asyncio.to_thread(sammle)} @app.delete("/workers/{name}") async def delete_worker(name: str): """Verwaisten Worker-Eintrag entfernen (alte Container-IDs nach Rebuilds). Ein AKTIVER Worker meldet sich binnen einer Minute einfach wieder an — löschen ist also immer gefahrlos.""" await asyncio.to_thread(db.delete_worker, name) await asyncio.to_thread(db.add_log, "info", "api", f"Worker-Eintrag '{name}' entfernt") return {"status": "deleted"} @app.get("/metadata/tv/{tv_id}/season/{season}") async def tv_season_laufzeiten(tv_id: int, season: int): """Episoden-Laufzeiten einer Staffel (TMDB) — Basis fürs Episoden-Matching des Workers (Serien-Flow).""" def hole(): prescan = PreScan() daten = prescan.tmdb.get_tv_season(tv_id, season) if not daten: return None return { "episodes": [ {"episode": e.get("episode_number"), "runtime": e.get("runtime")} for e in daten.get("episodes", []) ] } ergebnis = await asyncio.to_thread(hole) if ergebnis is None: raise HTTPException(status_code=404, detail="Staffel bei TMDB nicht gefunden") return ergebnis class MediaServerRefreshRequest(BaseModel): url: str api_key: str @app.post("/mediaserver/refresh") async def mediaserver_refresh(request: MediaServerRefreshRequest): """Bibliotheks-Scan von Jellyfin/Emby anstoßen — auch als Verbindungs-Test aus den Einstellungen (POST /Library/Refresh, Header X-Emby-Token).""" import requests as _requests url = request.url.strip().rstrip("/") if not url.startswith(("http://", "https://")): raise HTTPException(status_code=422, detail="Server-URL muss mit http(s):// beginnen") def anstossen(): return _requests.post( url + "/Library/Refresh", headers={"X-Emby-Token": request.api_key.strip()}, timeout=15, ) try: antwort = await asyncio.to_thread(anstossen) except Exception as e: raise HTTPException(status_code=400, detail=f"Server nicht erreichbar: {e}") if antwort.status_code >= 300: raise HTTPException( status_code=400, detail=f"Server antwortete mit HTTP {antwort.status_code} — API-Key prüfen " "(Jellyfin: Administration → API-Schlüssel)", ) await asyncio.to_thread(db.add_log, "info", "api", f"Bibliotheks-Refresh angestoßen ({url})") return {"status": "refreshed"} @app.get("/jobs/export") async def export_jobs(): """Job-Historie als CSV (Semikolon + BOM — öffnet sauber in deutschem Excel).""" def baue(): import csv import io puffer = io.StringIO() w = csv.writer(puffer, delimiter=";") w.writerow(["ID", "Titel", "Typ", "Status", "Fortschritt %", "Gerät", "Start", "Ende", "Ablage", "Fehler"]) for j in db.list_jobs(1000): w.writerow([ j["id"], j.get("title") or "", j.get("disc_type") or "", j.get("status") or "", j.get("progress") or 0, j.get("device") or "", j["created_at"].isoformat() if j.get("created_at") else "", j["finished_at"].isoformat() if j.get("finished_at") else "", j.get("output_path") or "", (j.get("error") or "").replace("\n", " "), ]) return puffer.getvalue() inhalt = await asyncio.to_thread(baue) return Response( content="" + inhalt, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="rippy-jobs.csv"'}, ) @app.get("/metadata/status") async def metadata_status(): """Live-Prüfung der Metadaten-Quellen — beantwortet „funktioniert mein Key?" sofort statt durch stilles Wegfallen einer Quelle.""" def pruefe(): from clients.omdb import OMDB_BASE_URL from clients.tmdb import TMDB_BASE_URL status = {} prescan = PreScan() # Bewusst am Cache VORBEI — ein alter Treffer soll keinen kaputten # Key als "ok" tarnen. /configuration ist der kleinste Auth-Aufruf. if not prescan.tmdb.api_key: status["tmdb"] = "kein_key" else: try: antwort = prescan.tmdb.session.get( f"{TMDB_BASE_URL}/configuration", params=prescan.tmdb._key_params, timeout=10, ) status["tmdb"] = "ok" if antwort.status_code == 200 else "fehler" except Exception: status["tmdb"] = "fehler" if not prescan.omdb.api_key: status["omdb"] = "kein_key" else: try: antwort = prescan.omdb.session.get( OMDB_BASE_URL, params={"apikey": prescan.omdb.api_key, "t": "Inception"}, timeout=10, ).json() status["omdb"] = "ok" if antwort.get("Response") == "True" else "fehler" except Exception: status["omdb"] = "fehler" status["jikan"] = "ok" # keyless — fällt nur bei Netzproblemen aus return status return await asyncio.to_thread(pruefe) class MountRequest(BaseModel): name: str type: str # nfs | cifs source: str # host:/export bzw. //host/share options: Optional[str] = None username: Optional[str] = None password: Optional[str] = None @app.get("/storage-mounts") async def get_storage_mounts(): """Konfigurierte Netzwerk-Speicherziele inkl. Live-Mount-Status.""" eintraege = await asyncio.to_thread(db.list_mounts) return [ { "name": e["name"], "type": e["typ"], "source": e["quelle"], "mounted": mount_verwaltung.ist_gemountet(e["name"]), "has_credentials": bool(e.get("username")), } for e in eintraege ] @app.post("/storage-mounts", status_code=201) async def create_storage_mount(request: MountRequest): """Hängt ein NFS/SMB-Ziel ein und speichert es für den nächsten Start.""" if not mount_verwaltung.validiere_name(request.name): raise HTTPException(status_code=422, detail="Name: nur a-z, 0-9, Bindestrich (2-31 Zeichen)") if request.type not in ("nfs", "cifs"): raise HTTPException(status_code=422, detail="Typ muss nfs oder cifs sein") if any(e["name"] == request.name for e in await asyncio.to_thread(db.list_mounts)): raise HTTPException(status_code=409, detail="Name bereits vergeben") try: schreibbar = await asyncio.to_thread( mount_verwaltung.mounten, request.name, request.type, request.source, request.options or "", request.username or "", request.password or "", ) except RuntimeError as e: raise HTTPException(status_code=400, detail=str(e)) await asyncio.to_thread( db.save_mount, request.name, request.type, request.source, request.options or "", request.username or "", request.password or "", ) await asyncio.to_thread( db.add_log, "success" if schreibbar else "warning", "mounts", f"Speicherziel '{request.name}' ({request.type}) eingehängt: {request.source}" + ("" if schreibbar else " — ACHTUNG: NUR LESBAR (Schreibtest fehlgeschlagen)"), ) return {"name": request.name, "mounted": True, "writable": schreibbar} @app.get("/storage-mounts/shares") async def list_shares(host: str, username: str = "", password: str = ""): """SMB-Freigaben eines Rechners auflisten (PC/NAS per Klick wählen).""" try: freigaben = await asyncio.to_thread( mount_verwaltung.liste_smb_freigaben, host, username, password ) except RuntimeError as e: raise HTTPException(status_code=400, detail=str(e)) return {"host": host, "shares": freigaben} @app.delete("/storage-mounts/{name}") async def delete_storage_mount(name: str): """Hängt ein Netzwerk-Speicherziel aus und entfernt es aus der Konfiguration.""" try: await asyncio.to_thread(mount_verwaltung.aushaengen, name) except RuntimeError as e: raise HTTPException(status_code=400, detail=str(e)) await asyncio.to_thread(db.delete_mount, name) await asyncio.to_thread(db.add_log, "info", "mounts", f"Speicherziel '{name}' entfernt") return {"status": "removed"} @app.get("/browse") async def browse(path: str = MEDIA_ROOT): """Server-seitiger Ordner-Browser für die Ziel-Auswahl (nur unter /app/media).""" normalisiert = os.path.normpath(path) if not normalisiert.startswith(MEDIA_ROOT): raise HTTPException(status_code=422, detail=f"Nur Pfade unter {MEDIA_ROOT}") def liste(): try: eintraege = sorted(os.listdir(normalisiert)) except OSError: return None ordner, dateien = [], [] for name in eintraege: voll = os.path.join(normalisiert, name) if os.path.isdir(voll): ordner.append({"name": name, "path": voll}) else: # Dateien MIT anzeigen (Befund 24.07.: der Browser wirkte # „leer", weil er nur Ordner listete — die MKVs im # bluray-Ordner waren unsichtbar). try: groesse_mb = round(os.path.getsize(voll) / 1024**2, 1) except OSError: groesse_mb = None dateien.append({"name": name, "size_mb": groesse_mb}) return ordner, dateien ergebnis = await asyncio.to_thread(liste) if ergebnis is None: raise HTTPException(status_code=404, detail="Ordner nicht lesbar") ordner, dateien = ergebnis eltern = os.path.dirname(normalisiert) if normalisiert != MEDIA_ROOT else None return {"path": normalisiert, "parent": eltern, "dirs": ordner, "files": dateien} class MkdirRequest(BaseModel): path: str name: str @app.post("/browse/mkdir", status_code=201) async def browse_mkdir(request: MkdirRequest): """Neuen Ordner unter /app/media anlegen (Speicherziele-Verwaltung).""" basis = os.path.normpath(request.path) if not basis.startswith(MEDIA_ROOT): raise HTTPException(status_code=422, detail=f"Nur Pfade unter {MEDIA_ROOT}") name = request.name.strip() if not name or "/" in name or "\\" in name or name.startswith("."): raise HTTPException(status_code=422, detail="Ungültiger Ordnername") ziel = os.path.join(basis, name) try: await asyncio.to_thread(os.makedirs, ziel, exist_ok=True) except OSError as e: raise HTTPException(status_code=400, detail=f"Anlegen fehlgeschlagen: {e}") return {"path": ziel} @app.get("/worker-setup/windows") async def worker_setup_windows(): """Windows-Installer-Skript für den nativen Transcode-Worker. Das UI (Einstellungen → Worker → Windows) zeigt den passenden Zwei-Zeilen-Aufruf — alles kommt von dieser Rippy-Instanz selbst. """ pfad = "worker_dist/install.ps1" if not os.path.isfile(pfad): raise HTTPException(status_code=404, detail="Installer nicht im Image — API neu bauen") return FileResponse(pfad, media_type="text/plain", filename="install-rippy-worker.ps1") @app.get("/worker-setup/paket") async def worker_setup_paket(): """Worker-Quellcode als Zip — der Windows-Installer lädt ihn von hier. Kein git, kein Docker auf der Zielmaschine nötig: die Rippy-Instanz versorgt ihre Worker selbst (Dateien liegen via Dockerfile im Image). """ def baue(): import io import zipfile if not os.path.isdir("worker_dist"): return None puffer = io.BytesIO() with zipfile.ZipFile(puffer, "w", zipfile.ZIP_DEFLATED) as z: for name in sorted(os.listdir("worker_dist")): if name.endswith(".py") and not name.startswith(("test_", "conftest")): z.write(os.path.join("worker_dist", name), name) elif name == "requirements.txt": z.write(os.path.join("worker_dist", name), name) return puffer.getvalue() inhalt = await asyncio.to_thread(baue) if not inhalt: raise HTTPException(status_code=404, detail="Worker-Paket nicht im Image — API neu bauen") return Response( content=inhalt, media_type="application/zip", headers={"Content-Disposition": 'attachment; filename="rippy-worker.zip"'}, ) @app.get("/system/info") async def system_info(): """System-Selbstauskunft (Einstellungen → System): Werkzeug-Versionen der Worker, freier Platz auf Media- und Arbeits-Volume, MakeMKV-Key-Status.""" def sammle(): info = {"api_version": app.version, "plaetze": [], "workers": db.list_workers()} for name, pfad in (("Media (/app/media)", MEDIA_ROOT), ("Arbeitsverzeichnis (/app/temp)", "/app/temp")): try: nutzung = shutil.disk_usage(pfad) info["plaetze"].append({ "name": name, "frei_gb": round(nutzung.free / 1024**3, 1), "gesamt_gb": round(nutzung.total / 1024**3, 1), }) except OSError: pass einstellungen = db.get_settings() info["makemkv_key_ui"] = bool((einstellungen.get("makemkvAppKey") or "").strip()) info["webhook_gesetzt"] = bool((einstellungen.get("notificationWebhook") or "").strip()) return info return await asyncio.to_thread(sammle) class NotificationTestRequest(BaseModel): url: str @app.post("/notifications/test") async def notification_test(request: NotificationTestRequest): """Test-Nachricht an den Webhook — beweist die Anbindung SOFORT statt erst beim ersten Job-Ende.""" url = request.url.strip() if not url.startswith(("http://", "https://")): raise HTTPException(status_code=422, detail="Webhook-URL muss mit http(s):// beginnen") try: await asyncio.to_thread( notify.sende, url, "🔔 Rippy: Test-Benachrichtigung", "Wenn du das liest, funktioniert die Anbindung. Rippy meldet sich " "hier, sobald ein Job fertig ist oder fehlschlägt.", "info", ) except RuntimeError as e: raise HTTPException(status_code=400, detail=str(e)) await asyncio.to_thread( db.add_log, "info", "notify", f"Test-Benachrichtigung gesendet ({notify.erkenne_webhook_typ(url)})" ) return {"status": "sent", "typ": notify.erkenne_webhook_typ(url)} @app.get("/setup") async def setup_status(): """First-Run-Erkennung: wurde der Einrichtungs-Assistent abgeschlossen?""" einstellungen = await asyncio.to_thread(db.get_settings, "setup") return {"done": bool(einstellungen.get("done"))} @app.post("/setup/complete") async def setup_complete(): await asyncio.to_thread(db.save_settings, {"done": True}, "setup") await asyncio.to_thread(db.add_log, "success", "setup", "Einrichtungs-Assistent abgeschlossen") return {"done": True} @app.get("/logs") async def get_logs(limit: int = 200): """Echte Ereignisse aus der Datenbank (Watcher, API, Worker).""" zeilen = await asyncio.to_thread(db.list_logs, min(limit, 1000)) return [ { "id": str(z["id"]), "timestamp": z["ts"].isoformat() if z.get("ts") else "", "level": z.get("level") or "info", "source": z.get("source") or "system", "message": z.get("message") or "", } for z in zeilen ] @app.get("/settings") async def get_settings(): """UI-Einstellungen aus der Datenbank (leeres Objekt = Defaults im UI).""" return await asyncio.to_thread(db.get_settings) @app.post("/settings") async def save_settings(werte: Dict): """Speichert die UI-Einstellungen als JSON in der Datenbank.""" await asyncio.to_thread(db.save_settings, werte) return {"status": "saved"} @app.get("/devices", response_model=List[Device]) async def get_devices(): """Alle optischen Laufwerke mit ehrlichem Status (leer/bereit + Disc-Typ). Der alte Weg (udevadm + /dev/disc-Symlinks) lieferte im Container prinzipbedingt nichts: kein udevd, keine udev-Datenbank, kein Daemon, der Symlinks anlegt. Jetzt: /sys fürs Modell, ioctl für den Disc-Status. """ geraete = [] for pfad in device_discovery.list_optical_devices(): info = await asyncio.to_thread(device_discovery.device_info, pfad) disc = DISC_CACHE.get(pfad) if disc and not disc.get("_laeuft"): info["disc"] = disc geraete.append(Device(**info)) return geraete # SSE-Stream für Echtzeit-Updates @app.get("/stream/jobs") async def job_stream(): """SSE-Stream für Job-Updates. Fix 23.07.: Der alte Generator sendete nur, wenn `sse_connections` gefüllt war — aber NICHTS hat diese Liste je befüllt. Der Stream war ein Placebo. """ async def event_generator(): while True: jobs = await get_jobs() yield f"data: {json.dumps([j.dict() for j in jobs])}\n\n" await asyncio.sleep(2) return StreamingResponse(event_generator(), media_type="text/event-stream") # /metadata/lookup + /metadata/confirm entfernt (24.07., mit der # Metadaten-Seite): lookup scannte ein DUMMY-Device (/dev/dvd — existiert # nicht) und confirm schrieb in einen Cache-Key, den nie jemand las. # Die echte Korrektur läuft über /metadata/search + /metadata/override. @app.get("/metadata/search") async def metadata_search(q: str): """Manuelle Korrektur: Titel-Kandidaten aus ALLEN Quellen (TMDB/Jikan/OMDb). KONZEPT Schritt 5: „Commander bestätigt oder korrigiert manuell" — das hier ist der Korrektur-Teil, wenn die Automatik danebenliegt. """ def sammle(): prescan = PreScan() ergebnisse = [] for movie in (prescan.tmdb.search_movie(q) or [])[:4]: ergebnisse.append({ "title": movie.get("title", ""), "year": int(movie["release_date"][:4]) if movie.get("release_date") else None, "poster": f"https://image.tmdb.org/t/p/w342{movie['poster_path']}" if movie.get("poster_path") else "", "overview": (movie.get("overview") or "")[:200], "type": "movie", "source": "tmdb", "id": str(movie.get("id", "")), }) for show in (prescan.tmdb.search_tv(q) or [])[:3]: ergebnisse.append({ "title": show.get("name", ""), "year": int(show["first_air_date"][:4]) if show.get("first_air_date") else None, "poster": f"https://image.tmdb.org/t/p/w342{show['poster_path']}" if show.get("poster_path") else "", "overview": (show.get("overview") or "")[:200], "type": "tv", "source": "tmdb", "id": str(show.get("id", "")), }) ergebnisse += prescan.jikan.suche(q) ergebnisse += prescan.omdb.suche(q) # Mit Poster zuerst, Duplikate (Titel+Jahr) raus gesehen, dedup = set(), [] for e in sorted(ergebnisse, key=lambda x: 0 if x.get("poster") else 1): schluessel = ((e.get("title") or "").lower(), e.get("year")) if schluessel in gesehen: continue gesehen.add(schluessel) dedup.append(e) return dedup[:12] return await asyncio.to_thread(sammle) class MetadataOverride(BaseModel): device_path: str title: str year: Optional[int] = None poster: Optional[str] = None overview: Optional[str] = None type: Optional[str] = "movie" source: Optional[str] = None id: Optional[str] = None @app.post("/metadata/override") async def metadata_override(request: MetadataOverride): """Nutzer-Wahl für DIESE Disc merken: Karte, Jobs und Cache (30 Tage). Der Disc-Fingerabdruck (Label+Größe) macht die Korrektur wiedererkennbar — dieselbe Disc wird beim nächsten Einlegen sofort richtig angezeigt. """ if request.device_path not in device_discovery.list_optical_devices(): raise HTTPException(status_code=404, detail="Laufwerk nicht gefunden") def speichere(): from cache import set as cache_setter from cache.keys import generate_prescan_key from prescan.prescan import disc_fingerprint ergebnis = { "disc_type": (DISC_CACHE.get(request.device_path) or {}).get("disc_type", "Blu-ray"), "title": request.title, "year": request.year, "confidence": 0.99, "metadata": { "type": request.type or "movie", "id": request.id or "", "title": request.title, "year": request.year, "overview": request.overview or "", "poster_path": request.poster or "", "backdrop_path": "", "runtime": 0, "genres": [], "source": request.source or "manuell", }, "tracks": [], } abdruck = disc_fingerprint(request.device_path) cache_setter( generate_prescan_key(request.device_path, False, abdruck), ergebnis, expire=30 * 86400, ) DISC_CACHE[request.device_path] = ergebnis return ergebnis ergebnis = await asyncio.to_thread(speichere) await asyncio.to_thread( db.add_log, "success", "api", f"Metadaten manuell festgelegt: {request.title}" + (f" ({request.year})" if request.year else ""), ) return ergebnis # Pre-Scan Endpoint class PreScanRequest(BaseModel): device_path: str @app.post("/prescan") async def run_prescan(request: PreScanRequest): """Führe Pre-Scan durch.""" try: prescan = PreScan() result = prescan.scan(request.device_path) return result.to_dict() except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Jellyfin-Formatierung Endpoints class JellyfinFormatRequest(BaseModel): title: str year: Optional[int] metadata: Dict disc_type: str output_dir: str @app.post("/jellyfin/format") async def jellyfin_format(request: JellyfinFormatRequest): """Formatiere für Jellyfin (NFO + Images).""" try: nfo_gen = NFOGenerator() img_downloader = ImageDownloader() # Ordnerstruktur erstellen output_path = Path(request.output_dir) if request.disc_type in ["dvd", "bluray"]: # Film-Formatierung title = request.metadata.get("title", request.title) year = request.year or request.metadata.get("year") # movie.nfo movie_nfo = nfo_gen.generate_movie_nfo( title=title, year=year or 2000, overview=request.metadata.get("overview", ""), rating=request.metadata.get("rating", 0), runtime=request.metadata.get("runtime", 0), genres=request.metadata.get("genres", []), director=request.metadata.get("director", ""), actors=request.metadata.get("actors", []) ) nfo_path = output_path / "movie.nfo" nfo_gen.save_nfo(movie_nfo, nfo_path) # Poster und Fanart img_downloader.download_poster(title, output_path, 500) img_downloader.download_fanart(title, output_path, 1920) return { "status": "formatted", "nfo_path": str(nfo_path), "poster_path": str(output_path / "poster.jpg"), "fanart_path": str(output_path / "fanart.jpg") } else: # Audio-Formatierung artist = request.metadata.get("artist", "Unknown Artist") album = request.metadata.get("title", request.title) # Review-Fix 22.07.: `year` war hier undefiniert (existierte nur im Film-Zweig) year = request.year or request.metadata.get("year") # album.nfo album_nfo = nfo_gen.generate_album_nfo( title=album, artist=artist, year=year or 2000, genres=request.metadata.get("genres", []) ) nfo_path = output_path / "album.nfo" nfo_gen.save_nfo(album_nfo, nfo_path) # Album-Cover img_downloader.download_music_images(artist, album, output_path) return { "status": "formatted", "nfo_path": str(nfo_path), "album_cover_path": str(output_path / "album.jpg") } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Auth-Endpoints (/token, /api-keys) entfernt — Commander-Entscheid 24.07.: # Heimnetz-only, kein Login-Flow im UI, die Endpoints waren Placebo.