rippy/docker/api/main.py

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 fastapi.security import OAuth2PasswordBearer

from config import settings
from config_validation import validate_config, ConfigValidationError
from cache import init_cache, set as cache_set
from auth import (
    create_access_token,
    create_refresh_token,
    decode_token,
    is_blacklisted,
    add_to_blacklist,
)
from ratelimit import (
    check_rate_limit,
    get_rate_limit_remaining,
    validate_api_key,
    api_keys,
    create_api_key as ratelimit_create_api_key,
    delete_api_key as ratelimit_delete_api_key,
)
from prescan import PreScan
from nfo_generator import NFOGenerator
from image_downloader import ImageDownloader

app = FastAPI(
    title="Rippy API",
    description="API für das automatische Ripping-System",
    version="1.0.0"
)

# OAuth2 Scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


@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] = {}


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()
        db.add_log(
            "info", "watcher",
            f"Disc erkannt: {ergebnis.title}"
            + (f" ({ergebnis.year})" if ergebnis.year else "")
            + f" [{ergebnis.disc_type}, Confidence {ergebnis.confidence:.0%}] auf {pfad}",
        )
        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

    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
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """Rate-Limiting Middleware."""
    client_ip = request.client.host
    api_key = request.headers.get("X-API-Key")
    
    # Prüfe API Key
    if api_key:
        key_info = validate_api_key(api_key)
        if not key_info:
            raise HTTPException(status_code=401, detail="Ungültiger API Key")
    
    # 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)


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.
    titel = request.title
    meta_json = None
    disc = DISC_CACHE.get(device_path)
    if disc and not disc.get("_laeuft"):
        if not titel:
            titel = disc.get("title")
        meta_json = json.dumps({
            "year": disc.get("year"),
            "confidence": disc.get("confidence"),
            **(disc.get("metadata") or {}),
        })

    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("/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/<job_id>
    (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/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("/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")


# Metadaten-Lookup Endpoints
class MetadataLookupRequest(BaseModel):
    title: str
    year: Optional[int] = None
    disc_type: str = "dvd"


@app.post("/metadata/lookup")
async def lookup_metadata(request: MetadataLookupRequest):
    """Suche Metadaten für Disc."""
    prescan = PreScan()
    
    # Dummy device für Pre-Scan
    device = "/dev/dvd" if request.disc_type in ["dvd", "bluray"] else "/dev/cdrom"
    
    result = prescan.scan(device)
    
    return {
        "title": result.title,
        "year": result.year,
        "confidence": result.confidence,
        "metadata": result.metadata,
        "tracks": result.tracks
    }


@app.post("/metadata/confirm")
async def confirm_metadata(title: str, year: Optional[int] = None, metadata: Dict = None):
    """Bestätige Metadaten."""
    from cache.keys import generate_confirmed_key
    # In Cache speichern
    cache_key = generate_confirmed_key(title, year)
    cache_set(cache_key, {"title": title, "year": year, "metadata": metadata or {}})
    
    return {"status": "confirmed", "key": cache_key}


@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
class LoginRequest(BaseModel):
    username: str
    password: str


@app.post("/token")
async def login(request: LoginRequest):
    """Login und Token generieren."""
    # Einfache Auth für MVP (in Produktion mit Datenbank); Zugangsdaten aus .env
    if request.username == settings.admin_username and request.password == settings.admin_password:
        access_token = create_access_token(
            data={"sub": request.username, "scopes": ["admin"]}
        )
        refresh_token = create_refresh_token(
            data={"sub": request.username}
        )
        return {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "token_type": "bearer"
        }
    raise HTTPException(status_code=401, detail="Ungültige Anmeldedaten")


@app.post("/token/refresh")
async def refresh_token(refresh_token: str):
    """Refresh Access Token."""
    payload = decode_token(refresh_token)
    if not payload or payload.get("type") != "refresh":
        raise HTTPException(status_code=401, detail="Ungültiges Refresh Token")
    
    access_token = create_access_token(
        data={"sub": payload.get("sub"), "scopes": payload.get("scopes", [])}
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.post("/token/invalidate")
async def invalidate_token(token: str):
    """Invalidate Token (Logout)."""
    if is_blacklisted(token):
        raise HTTPException(status_code=400, detail="Token bereits invalidiert")

    # Review-Fix 22.07.: vorher wurde hier NICHTS geblacklistet (Placebo-Logout)
    add_to_blacklist(token)
    if not is_blacklisted(token):
        raise HTTPException(status_code=400, detail="Ungültiger Token")
    return {"status": "invalidated"}


# API Key Endpoints
class APIKeyCreateRequest(BaseModel):
    name: str


# Review-Fix 22.07.: diese Endpoints nutzten `secrets` und `api_keys`, die in
# diesem Modul NIE existierten (Crash bei jedem Aufruf) — der echte Key-Store
# lebt in ratelimit.py und wird jetzt benutzt.
@app.post("/api-keys")
async def create_api_key(request: APIKeyCreateRequest):
    """Erstelle API Key."""
    # In Produktion mit Auth prüfen
    return ratelimit_create_api_key(request.name)


@app.get("/api-keys")
async def list_api_keys():
    """Liste API Keys."""
    return list(api_keys.values())


@app.delete("/api-keys/{key}")
async def delete_api_key(key: str):
    """Lösche API Key."""
    # In Produktion mit Auth prüfen
    if ratelimit_delete_api_key(key):
        return {"status": "deleted"}
    raise HTTPException(status_code=404, detail="API Key nicht gefunden")