Update app.py

app.py

@@ -1,284 +1,355 @@
-# MARK: app.py
+"""
+app.py
+FastAPI‑Server für das Ausliefern von Musik‑ und Bilddateien inkl. Metadaten.
+"""
+
+from __future__ import annotations
 
-from fastapi import FastAPI, Request, HTTPException
-from fastapi.responses import HTMLResponse, FileResponse, StreamingResponse
-from fastapi.staticfiles import StaticFiles
-from fastapi.templating import Jinja2Templates
-from fastapi.middleware.cors import CORSMiddleware
-from pathlib import Path
-from dominant_color import get_luminance, complementary, get_dominant_color, _is_grayscale, _rgb_to_hsv, hsv_to_rgb, rgb_to_hsv
-import subprocess
-import urllib.parse
-import os
-import json
-from mutagen import File
-from io import BytesIO
 import base64
-from PIL import Image
-from colorsys import rgb_to_hls, hls_to_rgb
+import json
 import logging
-import wave
-import struct
-import math
-import random
-import shutil
-
-# command = "cls"
-# if os.name in ("nt", "dos"):  # If Machine is running on Win OS, use cls
-#     subprocess.call(command, shell=True)
-# else:  # If Machine is running on OSX/Linux
-#     subprocess.call(["printf", "\033c"])
-
-
-no_style = "\033[0m"
-bold = "\033[91m"
-black = "\033[30m"
-grey = "\033[90m"
-white = "\033[97m"
-white_bg = "\033[107m"
-blue = "\033[94m"
-blue_bg = "\033[104m"
-green = "\033[92m"
-green_bg = "\033[102m"
-yellow = "\033[93m"
-red = "\033[31m"
-red_light = "\033[91m"
-red_light_bg = "\033[101m"
-
-global logger
-logger = logging.getLogger()
-handler = logging.StreamHandler()
-logging.addLevelName(logging.DEBUG, grey + "DEBUG:   " + no_style)
-logging.addLevelName(logging.INFO, blue + "ℹ️")
-logging.addLevelName(logging.WARNING, no_style + yellow + "WARNING:")
-logging.addLevelName(logging.ERROR, no_style + red_light_bg + black + "ERROR:   " + no_style)
-logging.addLevelName(logging.CRITICAL, f"{no_style + red}CRITICAL:{no_style}")
-handler.setLevel(logging.INFO)  # DEBUG INFO WARNING ERROR CRITICAL
-# formatter = logging.Formatter(fmt=" %(levelname)-8s %(message)s", style="%")
-# formatter = logging.Formatter(fmt=" %(levelname)-8s %(message)s")
-# # formatter = AnsiColorFormatter("{levelname:<8s} {message}", style="{")
-formatter = logging.Formatter("{levelname:s} {message}", style="{")
-handler.setFormatter(formatter)
-logger.addHandler(handler)
-logger.setLevel(logging.INFO)  # DEBUG INFO WARNING ERROR CRITICAL
-
-
-# MARK: FastAPI
-app = FastAPI()
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-SERVER_HOST = "0.0.0.0"
-SERVER_PORT = 7860
-BASE_URL = "https://sebastiankay-my-ai-songs.hf.space/"
-GET_FILE_URL = BASE_URL + "getfile/"
-
+import os
+import subprocess
+import urllib.parse
+from pathlib import Path
+from typing import List
 
-# Serve static files from the "static" directory
-app.mount("/static", StaticFiles(directory="static"), name="static")
+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
 
-# Set up Jinja2 templates
-templates = Jinja2Templates(directory="templates")
+# --------------------------------------------------------------------------- #
+# --------------------------- Logging-Konfiguration ------------------------ #
+# --------------------------------------------------------------------------- #
 
-# Configure the base directory for your music files
-app.mount("/files/music", StaticFiles(directory="files/music"), name="music")
-MUSIC_DIR = Path("files/music")
-MUSIC_DIR_2 = Path("files/music_2")
-
-
-def create_peaks_json(file_name):
-    input_file = f"{MUSIC_DIR / file_name}.mp3"
-    output_file = f"{MUSIC_DIR / file_name}.peaks.json"
-    command = f'audiowaveform -i "{input_file}" -o "{output_file}" -b 8'
-    # print("Executing command:", command)
-    subprocess.call(command, shell=True)
-
-    # with open(output_file, "r") as f:
-    #     peaks = json.load(f)
-
-    # # return peaks["data"]
-
-
-# Function to get the list of music files with metadata
-global get_music_files_with_metadata
-def get_music_files_with_metadata():
-    if Path("music_files.json").is_file():
-        print("Loading music_files.json")
-        with open("music_files.json", "r") as f:
-            return json.load(f)
-
-    music_files = []
-    for file in MUSIC_DIR.iterdir():
-        if file.is_file():
-
-            if "_HIDETHIS" in file.name or file.name.endswith(".json") or file.name.endswith(".png"):
-                print(f"{yellow}Ignoring File: {no_style} {file.name}")
-                continue
-
-            filename_without_mp3 = file.name.rsplit(".", 1)[0]
-            filename_coverart = filename_without_mp3 + ".coverart.png"
-            filename_coverart_encoded = urllib.parse.quote(filename_coverart)
-            filename_peaks = filename_without_mp3 + ".peaks.json"
-            filename_peaks_encoded = urllib.parse.quote(filename_without_mp3)
-            if not Path(MUSIC_DIR / filename_peaks).is_file():
-                print(f"{blue}No peaks file found for: {no_style} {file.name}")
-                create_peaks_json(filename_without_mp3)
-
-            try:  # Wrap metadata extraction in try-except
-                audio = File(file)
-                if audio is None:  # Handle cases where mutagen can't read the file
-                    print(f"Warning: Could not read metadata for {file.name}")
-                    continue
-
-                track_number_str = audio.get("TRCK", ["0"])[0].split("/")[0]
-                track_number = int(track_number_str) if track_number_str.isdigit() else 0
-
-                metadata = {
-                    "encoded_title": base64.b64encode(file.name.encode("utf-8")).decode("utf-8"),
-                    "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 hasattr(audio, "info") and hasattr(audio.info, "length") else 0,
-                    "track_number": track_number,
-                    "audiofile": GET_FILE_URL + file.name,
-                    "coverart": GET_FILE_URL + filename_coverart_encoded,
-                    "peaksfile": str(BASE_URL + "api/peaks/" + filename_peaks_encoded) if Path(MUSIC_DIR / filename_peaks).is_file() else "",
-                    "dominant_color_1": "128, 128, 128",
-                    "complementary_color_1": "128, 128, 128",
-                    "dominant_color_2": "128, 128, 128",
-                    "complementary_color_2": "128, 128, 128",
-                    # "peaks": peaks,
-                }
 
-                music_files.append(metadata)
+# 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"
 
-                # --- Cover Art Processing ---
-                cover_art_path = MUSIC_DIR / filename_coverart
-                if cover_art_path.is_file():
-                    try:
-                        colors = get_dominant_color(cover_art_path, min_brightness=0.4, max_brightness=0.8, n_clusters=4)
 
-                        metadata["dominant_color_1"] = ", ".join(map(str, colors[0]))
-                        metadata["complementary_color_1"] = ", ".join(map(str, colors[1]))
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(levelname)s %(message)s",
+    handlers=[logging.StreamHandler()],
+)
 
-                        metadata["dominant_color_2"] = ", ".join(map(str, colors[2]))
-                        metadata["complementary_color_2"] = ", ".join(map(str, colors[3]))
+# 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}")
 
-                    except Exception as e:
-                        print(f"{red_light}Error{no_style} processing cover art {metadata['coverart']}: {e}")
-                else:
-                    print(f"Cover art file not found: {cover_art_path}")
-                # --- End Cover Art Processing ---
+log = logging.getLogger(__name__)
 
-            except Exception as e:
-                print(f"Error processing file {file.name}: {e}")
+# --------------------------------------------------------------------------- #
+# ---------------------------- Konfiguration -------------------------------- #
+# --------------------------------------------------------------------------- #
 
-    # Sort music files by track number in descending order
-    music_files.sort(key=lambda x: x["track_number"], reverse=True)
+SERVER_HOST = "0.0.0.0"
+SERVER_PORT = 7860
 
-    with open("music_files.json", "w") as f:
-        json.dump(music_files, f, indent=4)
+BASE_URL = os.getenv("BASE_URL", "https://sebastiankay-my-ai-songs.hf.space/")
+GET_FILE_URL = f"{BASE_URL}getfile/"
 
-    return music_files
+# Pfade
+STATIC_DIR = Path("static")
+TEMPLATE_DIR = Path("templates")
+MUSIC_DIR = Path("files/music")
+CACHE_FILE = Path("music_files.json")
 
+# --------------------------------------------------------------------------- #
+# ---------------------------- FastAPI‑App ---------------------------------- #
+# --------------------------------------------------------------------------- #
 
-@app.get("/", response_class=FileResponse)
-async def read_root():
-    """Serves the main index.html file from the static directory."""
-    static_file_path = Path("static/index.html")
-    if not static_file_path.is_file():
-        raise HTTPException(status_code=404, detail="index.html not found in static directory")
-    return FileResponse(static_file_path)
+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=["*"],
+)
 
-@app.get("/{trackname}", response_class=FileResponse)
-async def read_root():
-    """Serves the main index.html file from the static directory."""
-    static_file_path = Path("static/index.html")
-    if not static_file_path.is_file():
-        raise HTTPException(status_code=404, detail="index.html not found in static directory")
-    return FileResponse(static_file_path)
+# 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 get_file(filename: str):
-    """Endpoint to download files from the music directory."""
-    # Security check: prevent path traversal
+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="Invalid filename")
+        raise HTTPException(status_code=400, detail="Ungültiger Dateiname")
 
-    # Create the full file path
     file_path = MUSIC_DIR / filename
-
-    # Check if file exists and is a file
     if not file_path.is_file():
-        raise HTTPException(status_code=404, detail="File not found")
-
-    # get_peaks(file_path)
-
-    # Determine media type based on file extension
-    media_type = "application/octet-stream"  # default binary type
-    if filename.lower().endswith(".wav"):
-        media_type = "audio/wav"
-    elif filename.lower().endswith(".mp3"):
-        media_type = "audio/mpeg"
-    elif filename.lower().endswith(".ogg"):
-        media_type = "audio/ogg"
-    elif filename.lower().endswith(".flac"):
-        media_type = "audio/flac"
-    elif filename.lower().endswith(".png"):
-        media_type = "image/png"
-    elif filename.lower().endswith(".jpg") or filename.lower().endswith(".jpeg"):
-        media_type = "image/jpeg"
-    elif filename.lower().endswith(".gif"):
-        media_type = "image/gif"
-
-    # def iterfile():  # (1)
-    #     with open(file_path, mode="rb") as file_like:  # (2)
-    #         yield from file_like  # (3)
-
-    # return StreamingResponse(iterfile(), media_type=media_type)
-    return FileResponse(path=file_path, media_type=media_type, filename=filename)
-
-
-@app.get("/api/peaks/{filename}")
-async def get_music_data_peaks(filename: str):
-    with open(f"{MUSIC_DIR / filename}.peaks.json", "r") as f:
-        peaks = json.load(f)
-        peaks_data = peaks["data"]
-    return peaks_data
+        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 get_music_data():
-    music_files = get_music_files_with_metadata()
-    return music_files
+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__":
-    import uvicorn
-
-    if Path("music_files.json").is_file():
-        print("Removing old music_files.json")
-        os.remove("music_files.json")
-
-    get_music_files_with_metadat()
-
-    # print(f"\n\nStarting server on: \nhttp://{SERVER_HOST}:{SERVER_PORT}")
-    print()
-    logger.info("=============================================================")
-    logger.info(" ")
-    logger.info(f"API DOCS: {BASE_URL}docs")
-    logger.info(f"ENDPOINT: {BASE_URL}api/music")
-    logger.info(f"ENDPOINT PEAKS: {BASE_URL}api/peaks")
-    logger.info("")
-    logger.info("=============================================================")
-    print()
-    print()
-    uvicorn.run(app, host=SERVER_HOST, port=SERVER_PORT)
+    # 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
+    )