Navi.FM/app/navidrome.py

import os
import hashlib
import random
import string
import requests
from typing import List, Dict

class NavidromeClient:
    def __init__(self):
        self.version = "1.16.1"
        self.client_name = "navifm"

    @staticmethod
    def verify_credentials(url: str, username: str, password: str) -> bool:
        salt = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(6))
        token = hashlib.md5((password + salt).encode('utf-8')).hexdigest()
        params = {
            "u": username,
            "t": token,
            "s": salt,
            "v": "1.16.1",
            "c": "navifm",
            "f": "json"
        }
        try:
            url_clean = url.rstrip('/')
            response = requests.get(f"{url_clean}/rest/ping", params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            if "subsonic-response" in data and data["subsonic-response"]["status"] == "ok":
                return True
            return False
        except Exception as e:
            print(f"Credentials verification failed: {e}")
            return False

    @property
    def url(self):
        from database import SessionLocal, get_setting
        db = SessionLocal()
        try:
            return get_setting(db, "navidrome_url", "")
        finally:
            db.close()

    @property
    def user(self):
        from database import SessionLocal, get_setting
        db = SessionLocal()
        try:
            return get_setting(db, "navidrome_user", "")
        finally:
            db.close()

    @property
    def password(self):
        from database import SessionLocal, get_setting
        import crypto
        db = SessionLocal()
        try:
            enc_pwd = get_setting(db, "navidrome_password", "")
            return crypto.decrypt_password(enc_pwd)
        finally:
            db.close()

    def _generate_salt(self, length=6):
        return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(length))

    def _get_auth_params(self):
        salt = self._generate_salt()
        token = hashlib.md5((self.password + salt).encode('utf-8')).hexdigest()
        return {
            "u": self.user,
            "t": token,
            "s": salt,
            "v": self.version,
            "c": self.client_name,
            "f": "json"
        }

    def _request(self, endpoint: str, params: Dict = None):
        if not self.url or not self.user or not self.password:
            print("Navidrome credentials not configured.")
            return None

        url = f"{self.url}/rest/{endpoint}"
        payload = self._get_auth_params()
        if params:
            payload.update(params)
        
        try:
            response = requests.get(url, params=payload, timeout=10)
            response.raise_for_status()
            data = response.json()
            if "subsonic-response" in data and data["subsonic-response"]["status"] == "ok":
                return data["subsonic-response"]
            else:
                print(f"Subsonic Error: {data}")
                return None
        except Exception as e:
            print(f"Request failed: {e}")
            return None

    def get_random_songs(self, count=100) -> List[Dict]:
        """Fetch random songs to discover tracks for the library."""
        res = self._request("getRandomSongs", {"size": count})
        if res and "randomSongs" in res and "song" in res["randomSongs"]:
            return res["randomSongs"]["song"]
        return []

    def get_total_songs(self) -> int:
        """Attempts to get the total number of songs from Navidrome."""
        if not self.url or not self.user or not self.password:
            return 0
        try:
            url_clean = self.url.rstrip('/')
            # Try native Navidrome API (requires real password, not subsonic token)
            login_res = requests.post(f"{url_clean}/auth/login", json={"username": self.user, "password": self.password}, timeout=5)
            if login_res.ok:
                token = login_res.json().get("token")
                if token:
                    headers = {"x-access-token": token}
                    song_res = requests.get(f"{url_clean}/api/song?_end=1", headers=headers, timeout=5)
                    if song_res.ok and "x-total-count" in song_res.headers:
                        return int(song_res.headers["x-total-count"])
        except Exception as e:
            print(f"Failed native API count: {e}")
            
        return 0

    def get_stream_url(self, song_id: str) -> str:
        """Returns the full authenticated URL to stream a song."""
        params = self._get_auth_params()
        params["id"] = song_id
        # We also might want to specify format, but usually Navidrome handles it
        query_string = "&".join(f"{k}={v}" for k, v in params.items())
        return f"{self.url}/rest/stream?{query_string}"
    
    def download_song(self, song_id: str, dest_path: str):
        """Downloads a song to disk for analysis."""
        stream_url = self.get_stream_url(song_id)
        try:
            with requests.get(stream_url, stream=True, timeout=30) as r:
                r.raise_for_status()
                with open(dest_path, 'wb') as f:
                    for chunk in r.iter_content(chunk_size=8192):
                        f.write(chunk)
            return True
        except Exception as e:
            print(f"Failed to download song {song_id}: {e}")
            return False

navidrome_client = NavidromeClient()