import os import time from fastapi import FastAPI, Request, HTTPException, Depends from fastapi.responses import PlainTextResponse from fastapi.security import HTTPBasic, HTTPBasicCredentials import secrets import asyncio from pydantic import BaseModel import httpx from fastapi.responses import StreamingResponse from database import init_db, SessionLocal, Track, get_setting, set_setting from worker import start_workers from queue_manager import get_active_profile, set_active_profile, get_next_track, AVAILABLE_PROFILES from navidrome import navidrome_client import bcrypt import crypto # Initialize DB and start workers on startup init_db() db = SessionLocal() try: if get_setting(db, "theme_color") is None: set_setting(db, "theme_color", "purple") if get_setting(db, "app_title") is None: set_setting(db, "app_title", "NAVI.FM") if get_setting(db, "app_tagline") is None: set_setting(db, "app_tagline", "The least intelligent radio software you’ll ever love.") if get_setting(db, "info_modal_text") is None: set_setting(db, "info_modal_text", "Navi.FM is an independent radio project that completely ditches traditional, static playlists. Instead, it relies on a custom, lightweight engine to intelligently shuffle and balance the unique mood, texture, and energy of every single track inside a privately curated, hand-picked music library.\n\nHow it Works\nInstead of just playing songs at random, Navi.FM uses smart audio analysis to understand how different tracks relate to each other and maps out a seamless, organic journey from one song to the next. It constantly schedules the upcoming tracks in real-time, steering the soundscape to keep the stream flowing perfectly.\n\nBehind the Scenes\nThere are no big media companies or human radio hosts pulling the strings. The entire station is a custom-built passion project run entirely out of a home setup.") if get_setting(db, "artist_separation_enabled") is None: set_setting(db, "artist_separation_enabled", "false") if get_setting(db, "artist_separation_count") is None: set_setting(db, "artist_separation_count", "5") # Migration: Encrypt legacy plaintext Navidrome passwords existing_nav_pwd = get_setting(db, "navidrome_password") if existing_nav_pwd: # Fernet tokens start with gAAAAA if not existing_nav_pwd.startswith("gAAAAA"): print("Migrating legacy plaintext Navidrome password to Fernet encryption...") set_setting(db, "navidrome_password", crypto.encrypt_password(existing_nav_pwd)) finally: db.close() start_workers() app = FastAPI(title="Navi.FM Backend") security = HTTPBasic(auto_error=False) FAILED_LOGIN_ATTEMPTS = {} MAX_LOGIN_ATTEMPTS = 5 LOCKOUT_DURATION = 300 # 5 minutes def verify_admin(request: Request, credentials: HTTPBasicCredentials = Depends(security)): client_ip = request.client.host now = time.time() # Check rate limits if client_ip in FAILED_LOGIN_ATTEMPTS: FAILED_LOGIN_ATTEMPTS[client_ip] = [t for t in FAILED_LOGIN_ATTEMPTS[client_ip] if now - t < LOCKOUT_DURATION] if len(FAILED_LOGIN_ATTEMPTS[client_ip]) >= MAX_LOGIN_ATTEMPTS: raise HTTPException(status_code=429, detail="Too many failed login attempts. Please try again later.") if not credentials: raise HTTPException(status_code=401, detail="Not authenticated") db = SessionLocal() try: admin_user = get_setting(db, "admin_user") admin_pass_hash = get_setting(db, "admin_password_hash") finally: db.close() if not admin_user or not admin_pass_hash: raise HTTPException(status_code=401, detail="Admin not setup") correct_username = secrets.compare_digest(credentials.username, admin_user) # Use bcrypt to verify password try: correct_password = bcrypt.checkpw( credentials.password.encode('utf-8'), admin_pass_hash.encode('utf-8') ) except Exception: correct_password = False if not (correct_username and correct_password): FAILED_LOGIN_ATTEMPTS.setdefault(client_ip, []).append(now) raise HTTPException( status_code=401, detail="Incorrect username or password", ) # Clear attempts on success if client_ip in FAILED_LOGIN_ATTEMPTS: del FAILED_LOGIN_ATTEMPTS[client_ip] return credentials.username class AdminSetup(BaseModel): username: str password: str class NavidromeSetup(BaseModel): url: str username: str password: str @app.get("/api/setup/status") async def get_setup_status(): db = SessionLocal() try: admin_setup = get_setting(db, "admin_user") is not None nav_setup = get_setting(db, "navidrome_url") is not None analyzed_tracks = db.query(Track).filter(Track.analyzed == True).count() return { "is_setup": admin_setup and nav_setup and analyzed_tracks > 10, "admin_setup": admin_setup, "navidrome_setup": nav_setup, "analyzed_tracks": analyzed_tracks } finally: db.close() @app.post("/api/setup/admin") async def setup_admin(data: AdminSetup): db = SessionLocal() try: if get_setting(db, "admin_user"): raise HTTPException(status_code=400, detail="Admin already setup") hashed_password = bcrypt.hashpw(data.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') set_setting(db, "admin_user", data.username) set_setting(db, "admin_password_hash", hashed_password) return {"status": "success"} finally: db.close() @app.post("/api/setup/navidrome") async def setup_navidrome(data: NavidromeSetup): db = SessionLocal() try: if get_setting(db, "navidrome_url"): raise HTTPException(status_code=400, detail="Navidrome already setup") from navidrome import NavidromeClient is_valid = NavidromeClient.verify_credentials(data.url, data.username, data.password) if not is_valid: raise HTTPException(status_code=400, detail="Could not connect to Navidrome server. Please check your URL and credentials.") set_setting(db, "navidrome_url", data.url.rstrip("/")) set_setting(db, "navidrome_user", data.username) set_setting(db, "navidrome_password", crypto.encrypt_password(data.password)) return {"status": "success"} finally: db.close() class ProfileUpdate(BaseModel): profile: str @app.get("/api/profiles") async def get_profiles(_: str = Depends(verify_admin)): db = SessionLocal() try: active_profile = get_active_profile(db) track_count = db.query(Track).filter(Track.analyzed == True).count() total_track_count = navidrome_client.get_total_songs() finally: db.close() return { "profiles": AVAILABLE_PROFILES, "active_profile": active_profile, "track_count": track_count, "total_track_count": total_track_count } @app.post("/api/profile") async def update_profile(data: ProfileUpdate, _: str = Depends(verify_admin)): valid_ids = [p["id"] for p in AVAILABLE_PROFILES] if data.profile not in valid_ids: raise HTTPException(status_code=400, detail="Invalid profile") set_active_profile(data.profile) return {"status": "success", "profile": data.profile} class SettingsUpdate(BaseModel): theme_color: str app_title: str app_tagline: str info_modal_text: str favicon_svg_overlay: str = "" favicon_svg_color: str = "white" artist_separation_enabled: bool = False artist_separation_count: int = 5 auto_pilot_schedule: str = "[]" @app.get("/api/settings") async def read_settings(): db = SessionLocal() try: return { "theme_color": get_setting(db, "theme_color", "purple"), "app_title": get_setting(db, "app_title", "COLIE.FM"), "app_tagline": get_setting(db, "app_tagline", ""), "info_modal_text": get_setting(db, "info_modal_text", ""), "favicon_svg_overlay": get_setting(db, "favicon_svg_overlay", ""), "favicon_svg_color": get_setting(db, "favicon_svg_color", "white"), "artist_separation_enabled": get_setting(db, "artist_separation_enabled", "false") == "true", "artist_separation_count": int(get_setting(db, "artist_separation_count", "5")), "auto_pilot_schedule": get_setting(db, "auto_pilot_schedule", "[]") } finally: db.close() @app.post("/api/settings") async def update_settings(data: SettingsUpdate, _: str = Depends(verify_admin)): db = SessionLocal() try: set_setting(db, "theme_color", data.theme_color) set_setting(db, "app_title", data.app_title) set_setting(db, "app_tagline", data.app_tagline) set_setting(db, "info_modal_text", data.info_modal_text) set_setting(db, "favicon_svg_overlay", data.favicon_svg_overlay) set_setting(db, "favicon_svg_color", data.favicon_svg_color) set_setting(db, "artist_separation_enabled", "true" if data.artist_separation_enabled else "false") set_setting(db, "artist_separation_count", str(data.artist_separation_count)) set_setting(db, "auto_pilot_schedule", data.auto_pilot_schedule) finally: db.close() return {"status": "success"} @app.get("/api/history") async def get_history(_: str = Depends(verify_admin)): """ Returns the last 10 tracks played for the admin/debug panel. """ now_data = await now_playing() return {"history": now_data.get("history", [])} import socket @app.post("/api/skip") async def skip_track(_: str = Depends(verify_admin)): """ Sends a telnet command to Liquidsoap to skip the current track. """ try: # Connect to liquidsoap telnet server with socket.create_connection(("127.0.0.1", 1234), timeout=2) as sock: # Send the skip command for the navifm_queue output node sock.sendall(b"navifm_queue.skip\r\nquit\r\n") # Read the 'Done' response to ensure it was processed sock.recv(4096) return {"status": "success", "message": "Track skipped"} except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to skip track: {str(e)}") @app.get("/next", response_class=PlainTextResponse) async def next_track(request: Request): """ Called by Liquidsoap to get the URL of the next track to play. Restricted to localhost for security. """ if request.client.host != "127.0.0.1": raise HTTPException(status_code=403, detail="Forbidden: Internal endpoint") track_data = get_next_track() if track_data: # Generate the stream URL from Navidrome url = navidrome_client.get_stream_url(track_data["id"]) lufs = track_data.get("lufs") if lufs is not None: # Calculate ratio to hit target of -14 LUFS (common streaming standard) target_lufs = -14.0 delta_db = target_lufs - lufs # Cap the adjustment between -15dB and +10dB to avoid extreme boosting delta_db = min(max(delta_db, -15.0), 10.0) ratio = 10 ** (delta_db / 20.0) # Use Liquidsoap's annotate protocol to pass the custom amplify metadata return f'annotate:liq_amplify="{ratio:.4f}":{url}' return url else: # Return empty or some fallback if queue is entirely empty # Sleep for 5 seconds to prevent liquidsoap from looping instantly and spamming logs await asyncio.sleep(5) return "" @app.get("/health") async def health_check(): return {"status": "healthy"} @app.get("/stream") async def proxy_stream(request: Request): """ Proxies the Icecast audio stream so the frontend doesn't need to connect to port 8000 directly. """ client = httpx.AsyncClient(timeout=None) # Pre-flight request to get headers req = client.build_request("GET", "http://127.0.0.1:8000/stream") response = await client.send(req, stream=True) headers = {} for k, v in response.headers.items(): if k.lower().startswith('icy-') or k.lower() == 'content-type': headers[k] = v async def stream_generator(): try: async for chunk in response.aiter_bytes(): if await request.is_disconnected(): break yield chunk except Exception: pass finally: await response.aclose() await client.aclose() return StreamingResponse(stream_generator(), headers=headers) import time current_playing_state = { "rid": None, "start_time": int(time.time()), "track_dict": None, "history": None } @app.get("/api/internal/track_changed") async def track_changed(request: Request, title: str = None, artist: str = None): """ Webhook called by Liquidsoap exactly when a new track starts playing. Restricted to localhost for security. """ if request.client.host != "127.0.0.1": raise HTTPException(status_code=403, detail="Forbidden: Internal endpoint") global current_playing_state # 1. Pop previous track into history if current_playing_state["track_dict"] and current_playing_state["track_dict"].get("title") != "Unknown": old_track = current_playing_state["track_dict"].copy() old_track["last_played"] = int(time.time()) if current_playing_state["history"] is None: current_playing_state["history"] = [] current_playing_state["history"].insert(0, old_track) if len(current_playing_state["history"]) > 10: current_playing_state["history"].pop() # 2. Fetch new track metadata from DB db = SessionLocal() try: if title and artist: track = db.query(Track).filter(Track.title == title, Track.artist == artist).first() if track: current_playing_state["track_dict"] = { "id": track.id, "title": track.title, "artist": track.artist, "bpm": track.bpm, "energy": track.energy, "lufs": track.lufs } else: current_playing_state["track_dict"] = {"id": "", "title": title, "artist": artist, "bpm": 120, "energy": 0.5, "lufs": -14.0} else: current_playing_state["track_dict"] = {"id": "", "title": "Unknown", "artist": "Unknown", "bpm": 120, "energy": 0.5, "lufs": -14.0} current_playing_state["start_time"] = int(time.time()) finally: db.close() return {"status": "ok"} @app.get("/api/now-playing") async def now_playing(): """ Returns the metadata of the currently playing track for the frontend UI. """ global current_playing_state # If UI requests before first track_changed webhook hits, return a fallback if not current_playing_state["track_dict"]: db = SessionLocal() try: from queue_manager import get_current_track current = get_current_track() if current: current_playing_state["track_dict"] = current else: current_playing_state["track_dict"] = {"id": "", "title": "Unknown", "artist": "Unknown", "bpm": 120, "energy": 0.5, "lufs": -14.0} finally: db.close() # Seed history from DB if empty on first boot if current_playing_state["history"] is None: db = SessionLocal() try: recent_tracks = db.query(Track).filter(Track.last_played > 0).order_by(Track.last_played.desc()).limit(15).all() h_list = [] v_time = int(time.time()) # Very rough approximation just so UI isn't empty on boot for t in recent_tracks[:10]: if t.title != current_playing_state["track_dict"].get("title"): h_list.append({ "id": t.id, "title": t.title, "artist": t.artist, "last_played": v_time }) v_time -= (t.duration if t.duration else 180) current_playing_state["history"] = h_list finally: db.close() # Return state res = current_playing_state["track_dict"].copy() res["history"] = current_playing_state["history"] res["server_time"] = int(time.time()) return res from fastapi.staticfiles import StaticFiles import os import re import time from fastapi.responses import HTMLResponse, Response frontend_dist = "/app/frontend/dist" @app.get("/favicon.svg") async def serve_favicon(): db = SessionLocal() try: theme_color = get_setting(db, "theme_color", "purple") overlay = get_setting(db, "favicon_svg_overlay", "") svg_color = get_setting(db, "favicon_svg_color", "white") finally: db.close() palette = { "purple": {"glow": "#8b5cf6"}, "blue": {"glow": "#3b82f6"}, "cyan": {"glow": "#06b6d4"}, "green": {"glow": "#10b981"}, "yellow": {"glow": "#eab308"}, "orange": {"glow": "#f97316"}, "red": {"glow": "#ef4444"}, "pink": {"glow": "#ec4899"} } glow = palette.get(theme_color, palette["purple"])["glow"] overlay_content = "" if overlay: # Remove hardcoded width/height processed = re.sub(r'\b(width|height)=("[^"]*"|\'[^\']*\')', '', overlay, flags=re.IGNORECASE) processed = re.sub(r'{processed}' svg_string = f''' {overlay_content} ''' return Response(content=svg_string, media_type="image/svg+xml", headers={"Cache-Control": "no-cache, no-store, must-revalidate"}) @app.get("/") async def serve_index(): index_path = os.path.join(frontend_dist, "index.html") if not os.path.exists(index_path): return HTMLResponse("Frontend not built") db = SessionLocal() try: app_title = get_setting(db, "app_title", "NAVI.FM") finally: db.close() with open(index_path, "r") as f: html = f.read() # Replace title html = re.sub(r".*?", f"{app_title}", html) # Inject favicon link cache_bust = int(time.time()) favicon_link = f'' html = html.replace("", f" {favicon_link}\n") return HTMLResponse(content=html) # Serve the Vue frontend built files if they exist if os.path.exists(frontend_dist): # Mount frontend dist at root but do not serve index.html directly for / app.mount("/", StaticFiles(directory=frontend_dist, html=False), name="static")