app.py

"""
app.py
FastAPI‑Server für das Ausliefern von Musik‑ und Bilddateien inkl. Metadaten.
"""

from __future__ import annotations

import base64
import json
import logging
import os
import subprocess
import urllib.parse
from pathlib import Path
from typing import List

import uvicorn
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from mutagen import File as MutagenFile
from dominant_color import get_dominant_color

# --------------------------------------------------------------------------- #
# --------------------------- Logging-Konfiguration ------------------------ #
# --------------------------------------------------------------------------- #


# Farben für die Konsole (nur für Development, bei Produktion meist deaktivieren)
class AnsiColor:
    GREY = "\033[90m"
    BLUE = "\033[94m"
    YELLOW = "\033[93m"
    RED_BG = "\033[101m"
    RED = "\033[31m"
    RESET = "\033[0m"


logging.basicConfig(
    level=logging.INFO,
    format="%(levelname)s %(message)s",
    handlers=[logging.StreamHandler()],
)

# Schönere Level‑Bezeichnungen (optional)
logging.addLevelName(logging.DEBUG, f"{AnsiColor.GREY}DEBUG{AnsiColor.RESET}")
logging.addLevelName(logging.INFO, f"{AnsiColor.BLUE}INFO{AnsiColor.RESET}")
logging.addLevelName(logging.WARNING, f"{AnsiColor.YELLOW}WARNING{AnsiColor.RESET}")
logging.addLevelName(logging.ERROR, f"{AnsiColor.RED_BG}{AnsiColor.RED}ERROR{AnsiColor.RESET}")

log = logging.getLogger(__name__)

# --------------------------------------------------------------------------- #
# ---------------------------- Konfiguration -------------------------------- #
# --------------------------------------------------------------------------- #

SERVER_HOST = "0.0.0.0"
SERVER_PORT = 7860

BASE_URL = os.getenv("BASE_URL", "https://sebastiankay-my-ai-songs.hf.space/")
GET_FILE_URL = f"{BASE_URL}getfile/"

# Pfade
STATIC_DIR = Path("static")
TEMPLATE_DIR = Path("templates")
MUSIC_DIR = Path("files/music")
CACHE_FILE = Path("music_files.json")

# --------------------------------------------------------------------------- #
# ---------------------------- FastAPI‑App ---------------------------------- #
# --------------------------------------------------------------------------- #

app = FastAPI(
    title="Sebastians Music API",
    description="Ausliefern von Musikdateien, Cover‑Art und Metadaten.",
    version="1.0.0",
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Statisches Verzeichnis (HTML/JS/CSS)
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
# Für direkte Downloads aus dem music‑Ordner
app.mount("/files/music", StaticFiles(directory=MUSIC_DIR), name="music")

templates = Jinja2Templates(directory=TEMPLATE_DIR)

# --------------------------------------------------------------------------- #
# --------------------------- Hilfs‑Funktionen ------------------------------- #
# --------------------------------------------------------------------------- #


def _create_peaks_file(stem: str) -> None:
    """
    Erstellt die *.peaks.json‑Datei mit dem externen Tool `audiowaveform`.
    Wir nutzen `subprocess.run` – das gibt uns besseres Fehler‑Handling
    als `subprocess.call`.
    """
    input_path = MUSIC_DIR / f"{stem}.mp3"
    output_path = MUSIC_DIR / f"{stem}.peaks.json"

    if not input_path.is_file():
        log.warning(f"Audio‑Datei nicht gefunden: {input_path}")
        return

    cmd = [
        "audiowaveform",
        "-i",
        str(input_path),
        "-o",
        str(output_path),
        "-b",
        "8",
    ]

    try:
        subprocess.run(cmd, check=True, capture_output=True)
        log.info(f"Peaks‑File erzeugt: {output_path.name}")
    except subprocess.CalledProcessError as exc:
        log.error(f"Fehler beim Erzeugen von {output_path.name}: {exc.stderr.decode()}")


def _extract_metadata(file_path: Path) -> dict | None:
    """
    Liest Metadaten mit Mutagen und gibt ein Dictionary zurück.
    Rückgabe `None` bei komplettem Fehlschlag (z. B. kein unterstütztes Format).
    """
    audio = MutagenFile(file_path)
    if audio is None:
        log.warning(f"Mutagen kann {file_path.name} nicht lesen.")
        return None

    # Track‑Nummer (z. B. "5/12") → int, fallback 0
    track_raw = audio.get("TRCK", ["0"])[0].split("/")[0]
    track_number = int(track_raw) if track_raw.isdigit() else 0

    # Grund‑Metadaten (mit sinnvollen Fallbacks)
    meta = {
        "encoded_title": base64.b64encode(file_path.name.encode()).decode(),
        "title": str(audio.get("TIT2", ["Unknown"])[0]),
        "artist": str(audio.get("TPE1", ["Unknown"])[0]),
        "genre": str(audio.get("TCON", ["Unknown"])[0]),
        "duration": int(audio.info.length) if getattr(audio, "info", None) else 0,
        "track_number": track_number,
        "audiofile": f"{GET_FILE_URL}{file_path.name}",
    }

    # Cover‑Art und Peaks‑File ergänzen (falls vorhanden)
    stem = file_path.stem
    cover_name = f"{stem}.coverart.png"
    peaks_name = f"{stem}.peaks.json"

    meta["coverart"] = f"{GET_FILE_URL}{urllib.parse.quote(cover_name)}"
    if (MUSIC_DIR / peaks_name).is_file():
        meta["peaksfile"] = f"{BASE_URL}api/peaks/{urllib.parse.quote(stem)}"
    else:
        # Wenn das Peaks‑File fehlt, on‑the‑fly erzeugen
        _create_peaks_file(stem)
        if (MUSIC_DIR / peaks_name).is_file():
            meta["peaksfile"] = f"{BASE_URL}api/peaks/{urllib.parse.quote(stem)}"
        else:
            meta["peaksfile"] = ""

    # Farben aus dem Cover‑Art bestimmen (falls vorhanden)
    cover_path = MUSIC_DIR / cover_name
    if cover_path.is_file():
        try:
            # Wir holen vier Farben: 2 dominante + 2 komplementäre
            colors = get_dominant_color(
                cover_path,
                min_brightness=0.4,
                max_brightness=0.8,
                n_clusters=4,
            )
            meta.update(
                {
                    "dominant_color_1": ", ".join(map(str, colors[0])),
                    "complementary_color_1": ", ".join(map(str, colors[1])),
                    "dominant_color_2": ", ".join(map(str, colors[2])),
                    "complementary_color_2": ", ".join(map(str, colors[3])),
                }
            )
        except Exception as exc:
            log.error(f"Fehler beim Auswerten von {cover_name}: {exc}")
    else:
        log.debug(f"Kein Cover‑Art gefunden: {cover_name}")

    return meta


def _load_or_build_cache() -> List[dict]:
    """
    Lädt die JSON‑Cache‑Datei, wenn sie existiert, ansonsten wird sie neu gebaut.
    Der Cache reduziert das Laden von Metadaten bei jedem Request dramatisch.
    """
    if CACHE_FILE.is_file():
        log.info("Cache‑Datei gefunden → lade Metadaten")
        try:
            return json.loads(CACHE_FILE.read_text(encoding="utf-8"))
        except json.JSONDecodeError as exc:
            log.warning(f"Cache ist korrupt: {exc} → neu bauen")
            CACHE_FILE.unlink(missing_ok=True)

    # ---- Neu bauen ---------------------------------------------------------
    log.info("Cache wird neu erzeugt …")
    music_data: List[dict] = []

    for entry in MUSIC_DIR.iterdir():
        if not entry.is_file():
            continue
        # Ignorier‑Regeln (wie du sie hattest)
        if any(
            (
                "_HIDETHIS" in entry.name,
                entry.suffix.lower() in {".json", ".png"},
            )
        ):
            log.debug(f"Ignoriere Datei: {entry.name}")
            continue

        meta = _extract_metadata(entry)
        if meta:
            music_data.append(meta)

    # Sortieren nach Track‑Nummer (absteigend)
    music_data.sort(key=lambda x: x["track_number"], reverse=True)

    # Cache schreiben
    CACHE_FILE.write_text(json.dumps(music_data, indent=4, ensure_ascii=False), encoding="utf-8")
    log.info("Cache geschrieben → %s Einträge", len(music_data))
    return music_data


# --------------------------------------------------------------------------- #
# ----------------------------- API‑Endpoints -------------------------------- #
# --------------------------------------------------------------------------- #


@app.get("/", response_class=FileResponse, include_in_schema=False)
async def serve_index():
    """
    Liefert das `index.html` aus dem static‑Verzeichnis.
    """
    index_path = STATIC_DIR / "index.html"
    if not index_path.is_file():
        raise HTTPException(status_code=404, detail="index.html fehlt")
    return FileResponse(index_path)


@app.get("/{track_name}", response_class=FileResponse, include_in_schema=False)
async def serve_spa(track_name: str):
    """
    Für Single‑Page‑Applications: jede unbekannte Route liefert wieder das `index.html`.
    """
    # Wir ignorieren den Parameter – das Frontend übernimmt das Routing.
    return await serve_index()


@app.get("/getfile/{filename}")
async def download_file(filename: str):
    """
    Stellt jede Datei aus dem Musik‑Ordner zum Download bereit.
    Sicherheits‑Check verhindert Pfad‑Traversal.
    """
    if ".." in filename or filename.startswith("/"):
        raise HTTPException(status_code=400, detail="Ungültiger Dateiname")

    file_path = MUSIC_DIR / filename
    if not file_path.is_file():
        raise HTTPException(status_code=404, detail="Datei nicht gefunden")

    # Media‑Type ermitteln – kann später zu einem Mapping ausgelagert werden.
    mime = {
        ".wav": "audio/wav",
        ".mp3": "audio/mpeg",
        ".ogg": "audio/ogg",
        ".flac": "audio/flac",
        ".png": "image/png",
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".gif": "image/gif",
    }.get(file_path.suffix.lower(), "application/octet-stream")

    return FileResponse(path=file_path, media_type=mime, filename=filename)


@app.get("/api/peaks/{stem}")
async def get_peaks(stem: str):
    """
    Liefert das bereits generierte Peaks‑JSON.
    """
    peaks_path = MUSIC_DIR / f"{stem}.peaks.json"
    if not peaks_path.is_file():
        # Noch nicht vorhanden → versuchen wir, es on‑the‑fly zu erzeugen
        _create_peaks_file(stem)
        if not peaks_path.is_file():
            raise HTTPException(status_code=404, detail="Peaks‑File nicht gefunden")

    try:
        data = json.loads(peaks_path.read_text(encoding="utf-8"))
        return JSONResponse(content=data.get("data", []))
    except json.JSONDecodeError:
        raise HTTPException(status_code=500, detail="Ungültiges Peaks‑JSON")


@app.get("/api/music")
async def list_music():
    """
    Gibt die komplette Metadaten‑Liste zurück.
    """
    return _load_or_build_cache()


# --------------------------------------------------------------------------- #
# ------------------------------- Startup ------------------------------------ #
# --------------------------------------------------------------------------- #


@app.on_event("startup")
async def startup_event():
    """
    Beim Server‑Start den Cache (wenn nötig) bauen.
    """
    # Nur einmal beim Start – danach bleibt das im Speicher.
    _load_or_build_cache()
    log.info("FastAPI‑Server bereit 🎉")


# --------------------------------------------------------------------------- #
# ------------------------------ Main‑Block --------------------------------- #
# --------------------------------------------------------------------------- #

if __name__ == "__main__":
    # Im Entwicklungs‑Modus alte Cache‑Datei entfernen (optional)
    if CACHE_FILE.is_file():
        log.debug("Alte Cache‑Datei wird gelöscht")
        CACHE_FILE.unlink()

    # Server starten
    uvicorn.run(
        app,
        host=SERVER_HOST,
        port=SERVER_PORT,
        log_level="info",
        # reload=True,  # <-- für lokales Entwickeln aktivieren
    )