110 lines
3.4 KiB
Python
110 lines
3.4 KiB
Python
import sqlite3
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Optional
|
|
|
|
@dataclass
|
|
class Track:
|
|
id: int
|
|
title: str
|
|
path: str
|
|
favorite: bool
|
|
rating: Optional[int]
|
|
play_count: int
|
|
last_played: Optional[str]
|
|
playlists: List[str] = field(default_factory=list)
|
|
|
|
@dataclass
|
|
class Album:
|
|
id: int
|
|
title: str
|
|
tracks: List[Track] = field(default_factory=list)
|
|
|
|
@dataclass
|
|
class Artist:
|
|
id: str
|
|
name: str
|
|
albums: List[Album] = field(default_factory=list)
|
|
|
|
class NavidromeDB:
|
|
def __init__(self, db_path: str):
|
|
if not os.path.isfile(db_path):
|
|
raise FileNotFoundError(f"Navidrome DB not found: {db_path}")
|
|
self.conn = sqlite3.connect(db_path)
|
|
self.conn.row_factory = sqlite3.Row
|
|
|
|
def close(self):
|
|
self.conn.close()
|
|
|
|
def get_artists(self) -> List[Artist]:
|
|
artists = []
|
|
c = self.conn.cursor()
|
|
c.execute("SELECT id, name FROM artist ORDER BY name")
|
|
for row in c.fetchall():
|
|
artist = Artist(id=row["id"], name=row["name"])
|
|
artist.albums = self.get_albums(artist.id)
|
|
artists.append(artist)
|
|
return artists
|
|
|
|
def get_albums(self, artist_id: str) -> List[Album]:
|
|
albums = []
|
|
c = self.conn.cursor()
|
|
c.execute(
|
|
"SELECT id, name FROM album WHERE album_artist_id=? ORDER BY name",
|
|
(artist_id,)
|
|
)
|
|
for row in c.fetchall():
|
|
album = Album(id=row["id"], title=row["name"])
|
|
album.tracks = self.get_tracks(album.id)
|
|
albums.append(album)
|
|
return albums
|
|
|
|
def get_tracks(self, album_id: int) -> List[Track]:
|
|
tracks = []
|
|
c = self.conn.cursor()
|
|
c.execute(
|
|
"SELECT t.id, t.name AS Title, t.path, ut.favorite, ut.rating, ut.play_count AS PlayCount, ut.last_played AS LastPlayed "
|
|
"FROM track t "
|
|
"LEFT JOIN usertrack ut ON ut.track_id = t.id "
|
|
"WHERE t.album_id=? ORDER BY t.track_number",
|
|
(album_id,)
|
|
)
|
|
for row in c.fetchall():
|
|
track = Track(
|
|
id=row["id"],
|
|
title=row["Title"],
|
|
path=row["path"],
|
|
favorite=bool(row["favorite"]),
|
|
rating=row["rating"],
|
|
play_count=row["PlayCount"] or 0,
|
|
last_played=row["LastPlayed"],
|
|
playlists=self.get_playlists_for_track(row["id"])
|
|
)
|
|
tracks.append(track)
|
|
return tracks
|
|
|
|
def get_playlists_for_track(self, track_id: int) -> List[str]:
|
|
c = self.conn.cursor()
|
|
c.execute(
|
|
"SELECT p.name FROM playlist p "
|
|
"JOIN playlisttrack pt ON pt.playlist_id=p.id "
|
|
"WHERE pt.track_id=?",
|
|
(track_id,)
|
|
)
|
|
return [row["name"] for row in c.fetchall()]
|
|
|
|
# Example usage
|
|
if __name__ == "__main__":
|
|
db_path = "/mnt/main/data/music/navidrome.db"
|
|
navidb = NavidromeDB(db_path)
|
|
|
|
try:
|
|
artists = navidb.get_artists()
|
|
for artist in artists:
|
|
print(f"Artist: {artist.name}")
|
|
for album in artist.albums:
|
|
print(f" Album: {album.title}")
|
|
for track in album.tracks:
|
|
print(f" Track: {track.title} | Path: {track.path} | Fav: {track.favorite} | Plays: {track.play_count}")
|
|
finally:
|
|
navidb.close()
|