The idea is that the client, like Substreamer, connects to this script (running at localhost:8000, username and password currently set to admin / admin) and the script exposes the MM library as if it were a Subsonic library - artists, albums, genres, playlists are, I think, working. File browsing not so much at the moment, but that's what you get for a few hours of work.
It also supports transcoding of incompatible or of all tracks via ffmpeg (files are converted before being served rather than during playback so that seeking is supported), and should update playcount, lastplayed and rating if you give the track a rating in the client (these last few functions are a mixed bag, maybe there's a better way to do this than what I've implemented - COM for MM4 rating and playcount incrementing works fine, but last played is a SQL edit, and MM5 doesn't work with the COM functions I guess).
Code: Select all
import os
import queue
import threading
import asyncio
import hashlib
import logging
import re
import subprocess
import time
from typing import Optional
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.responses import JSONResponse, Response, FileResponse
import pythoncom
import win32com.client
import mutagen
import sys
import json
from cachetools import LRUCache
import sqlite3
# --- CONFIGURATION ---
DEFAULT_CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
def load_config():
cfg = {
"mediamonkey_db_path": None,
"subsonic_user": os.getenv("SUBSONIC_USER", "admin"),
"subsonic_pass": os.getenv("SUBSONIC_PASS", "admin"),
"transcode_mode": "incompatible_only", # "off" | "incompatible_only" | "always"
"transcode_format": "mp3",
"transcode_bitrate": 192,
"transcode_sample_rate": 44100,
"native_formats": ["mp3", "flac", "ogg", "m4a", "wav"],
"transcode_cache_max_mb": 5000,
}
if os.path.exists(DEFAULT_CONFIG_PATH):
with open(DEFAULT_CONFIG_PATH, "r") as f:
cfg.update(json.load(f))
return cfg
CONFIG = load_config()
MM_VERSION = 5
# If mediamonkey_db_path is explicitly set in config, use it directly (skip MM5/MM4 autodetect)
if CONFIG["mediamonkey_db_path"]:
MM_DB_PATH = CONFIG["mediamonkey_db_path"]
if not os.path.exists(MM_DB_PATH):
raise RuntimeError(f"Configured mediamonkey_db_path does not exist: {MM_DB_PATH}")
# Infer version from the provided filename
if os.path.basename(MM_DB_PATH).lower() == "mm.db":
MM_VERSION = 4
else:
MM_VERSION = 5
else:
MM5_DB = os.path.expandvars(r"%APPDATA%\MediaMonkey5\MM5.DB")
MM4_DB = os.path.expandvars(r"%APPDATA%\MediaMonkey\MM.DB")
if os.path.exists(MM5_DB):
MM_DB_PATH = MM5_DB
MM_VERSION = 5
elif os.path.exists(MM4_DB):
MM_DB_PATH = MM4_DB
MM_VERSION = 4
else:
raise RuntimeError("Could not locate a MediaMonkey database.")
API_VERSION = "1.16.1"
if sys.platform == "win32":
from asyncio import WindowsSelectorEventLoopPolicy
asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())
VALID_USER = CONFIG["subsonic_user"]
VALID_PASS = CONFIG["subsonic_pass"]
MEDIA_MAP = {}
art_cache = LRUCache(maxsize=100 * 1024 * 1024, getsizeof=lambda v: len(v[0]))
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("subsonic-mm")
# --- TRANSCODE CACHE ---
TRANSCODE_CACHE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "transcode_cache")
os.makedirs(TRANSCODE_CACHE_DIR, exist_ok=True)
MAX_CACHE_BYTES = CONFIG["transcode_cache_max_mb"] * 1024 * 1024
_transcode_locks = {}
_transcode_locks_guard = threading.Lock()
def _get_transcode_lock(key: str) -> threading.Lock:
with _transcode_locks_guard:
if key not in _transcode_locks:
_transcode_locks[key] = threading.Lock()
return _transcode_locks[key]
def needs_transcode(ext: str) -> bool:
mode = CONFIG["transcode_mode"]
if mode == "off":
return False
if mode == "always":
return True
return ext.lower() not in CONFIG["native_formats"]
def _cache_key(track_id: str, source_path: str) -> str:
mtime = os.path.getmtime(source_path)
fmt = CONFIG["transcode_format"]
bitrate = CONFIG["transcode_bitrate"]
sr = CONFIG["transcode_sample_rate"]
raw = f"{track_id}:{mtime}:{fmt}:{bitrate}:{sr}"
return hashlib.sha1(raw.encode()).hexdigest()
def _cache_ext() -> str:
return {"mp3": "mp3", "aac": "m4a", "ogg": "ogg"}.get(CONFIG["transcode_format"], "mp3")
def _cache_path(key: str) -> str:
return os.path.join(TRANSCODE_CACHE_DIR, f"{key}.{_cache_ext()}")
def _evict_if_needed():
files = [os.path.join(TRANSCODE_CACHE_DIR, f) for f in os.listdir(TRANSCODE_CACHE_DIR)]
files = [f for f in files if os.path.isfile(f)]
total = sum(os.path.getsize(f) for f in files)
if total <= MAX_CACHE_BYTES:
return
files.sort(key=lambda f: os.path.getatime(f)) # oldest-accessed first
for f in files:
if total <= MAX_CACHE_BYTES:
break
try:
total -= os.path.getsize(f)
os.remove(f)
except OSError:
pass
def _transcode_to_file(source_path: str, dest_path: str):
fmt = CONFIG["transcode_format"]
bitrate = CONFIG["transcode_bitrate"]
sample_rate = CONFIG["transcode_sample_rate"]
codec = {"mp3": "libmp3lame", "aac": "aac", "ogg": "libvorbis"}.get(fmt, "libmp3lame")
container = {"mp3": "mp3", "aac": "adts", "ogg": "ogg"}.get(fmt, "mp3")
tmp_path = dest_path + ".tmp"
cmd = [
"ffmpeg", "-y", "-i", source_path,
"-vn", "-ar", str(sample_rate), "-ac", "2",
"-b:a", f"{bitrate}k", "-c:a", codec,
"-map_metadata", "0", # carry over all readable source tags/metadata
"-id3v2_version", "3", # ensures MP3 output writes ID3v2 tags at all (off by default for some builds)
"-write_id3v1", "1", # extra compatibility for older/simpler players
"-f", container,
tmp_path,
]
result = subprocess.run(cmd, capture_output=True)
if result.returncode != 0 or not os.path.exists(tmp_path):
stderr_text = result.stderr.decode(errors="ignore")
tail = stderr_text[-1500:]
raise RuntimeError(f"ffmpeg failed (exit {result.returncode}): ...{tail}")
os.replace(tmp_path, dest_path)
def get_or_create_transcode(track_id: str, source_path: str) -> str:
"""Blocking. Run via asyncio.to_thread. Returns path to a playable, seekable cached file."""
key = _cache_key(track_id, source_path)
cached_path = _cache_path(key)
if os.path.exists(cached_path):
return cached_path
lock = _get_transcode_lock(key)
with lock:
if os.path.exists(cached_path): # re-check after acquiring, in case a concurrent request finished it
return cached_path
_transcode_to_file(source_path, cached_path)
_evict_if_needed()
return cached_path
# --- COM WORKER THREAD ---
class COMWorker(threading.Thread):
def __init__(self):
super().__init__(daemon=True)
self.request_queue = queue.Queue()
self.sdb = None
def _is_mm_running(self):
"""Checks Windows tasklist to ensure MediaMonkey is already open."""
try:
# Check for standard MediaMonkey
output = subprocess.check_output('tasklist /FI "IMAGENAME eq MediaMonkey.exe"', shell=True).decode()
if "MediaMonkey.exe" in output:
return True
# Check for non-skinned MediaMonkey (some MM4 users use this version)
output_ns = subprocess.check_output('tasklist /FI "IMAGENAME eq MediaMonkey (non-skinned).exe"', shell=True).decode()
if "MediaMonkey (non-skinned).exe" in output_ns:
return True
return False
except Exception:
return False
def _connect_com(self):
# Determine priority based on the global MM_VERSION
if MM_VERSION == 4:
prog_ids = [("SongsDB.SDBApplication", 4), ("SongsDB5.SDBApplication", 5)]
else:
prog_ids = [("SongsDB5.SDBApplication", 5), ("SongsDB.SDBApplication", 4)]
# 1. Try to cleanly grab the running MM instance first (GetActiveObject)
for prog_id, version in prog_ids:
try:
self.sdb = win32com.client.GetActiveObject(prog_id)
self.mm_version = version
print(f"[COM] Connected via {prog_id} (GetActiveObject)")
return
except pythoncom.com_error:
continue
# 2. If UAC blocks GetActiveObject, use Dispatch to force the hook into MM
for prog_id, version in prog_ids:
try:
self.sdb = win32com.client.Dispatch(prog_id)
self.mm_version = version
print(f"[COM] Connected via {prog_id} (Dispatch)")
return
except pythoncom.com_error:
continue
print("[COM] MediaMonkey connection failed. Is it running?")
self.sdb = None
def run(self):
pythoncom.CoInitialize()
self._connect_com()
while True:
task = self.request_queue.get()
if task is None: break
action, payload, future, loop = task
if not self.sdb:
self._connect_com()
if not self.sdb:
loop.call_soon_threadsafe(_safe_set_exception,future, RuntimeError("MediaMonkey is not running."))
self.request_queue.task_done()
continue
try:
if action == "evaluate_playlist":
with db_lock:
playlist = self.sdb.PlaylistByID(int(payload["id"]))
tracks = playlist.Tracks
result_tracks = []
for i in range(tracks.Count):
t = tracks.Item(i)
result_tracks.append({
"id": f"TRACK_{t.ID}",
"title": t.Title,
"artist": t.ArtistName,
"album": t.AlbumName,
"duration": int(t.SongLength / 1000) if t.SongLength else 0,
"track": safe_track_num(t.TrackOrder),
"path": t.Path,
"coverArt": f"TRACK_{t.ID}"
})
loop.call_soon_threadsafe(_safe_set_result, future, result_tracks)
elif action == "rate_track":
track_id = int(payload["id"].replace("TRACK_", ""))
new_rating = payload["rating"] * 20 if payload["rating"] > 0 else -1
# Strictly use the globally defined MM_VERSION
if MM_VERSION == 5:
script = f"""
(function() {{
app.db.executeQueryAsync('UPDATE Songs SET Rating={new_rating} WHERE ID={track_id}')
.then(function() {{ runJSCode_callback('OK'); }})
.catch(function(err) {{ runJSCode_callback('ERROR: ' + err); }});
}})();
"""
result = self.sdb.runJSCode(script, True)
print(f"[rate_track] MM5 result: {result}")
else:
print(f"[rate_track] Looking up track ID {track_id} in MM4")
iterator = self.sdb.Database.QuerySongs(f"AND Songs.ID = {track_id}")
if not iterator.EOF:
track = iterator.Item
print(f"[rate_track] Found track '{track.Title}', current rating={track.Rating}, setting to {new_rating}")
track.Rating = new_rating
track.UpdateDB()
print(f"[rate_track] UpdateDB() called, rating now reads back as {track.Rating}")
else:
print(f"[rate_track] QuerySongs.EOF True — no track found for ID {track_id}")
loop.call_soon_threadsafe(_safe_set_result, future, True)
elif action == "scrobble_track":
track_id = int(payload["id"].replace("TRACK_", ""))
if MM_VERSION == 5:
script = f"""
(function() {{
app.db.executeQueryAsync('UPDATE Songs SET PlayCounter = PlayCounter + 1, LastTimePlayed = (julianday(\\'now\\') - julianday(\\'1899-12-30\\')) WHERE ID={track_id}')
.then(function() {{ runJSCode_callback('OK'); }})
.catch(function(err) {{ runJSCode_callback('ERROR: ' + err); }});
}})();
"""
result = self.sdb.runJSCode(script, True)
print(f"[scrobble_track] MM5 result: {result}")
else:
iterator = self.sdb.Database.QuerySongs(f"AND Songs.ID = {track_id}")
if not iterator.EOF:
track = iterator.Item
# 1. Update PlayCounter via COM so the MM4 UI immediately reflects it
track.PlayCounter += 1
track.UpdateDB()
# 2. LastTimePlayed is read-only via MM4 COM, so we patch it directly in the DB
patch_sql = f"UPDATE Songs SET LastTimePlayed = (julianday('now') - julianday('1899-12-30')) WHERE ID={track_id}"
self.sdb.Database.ExecSQL(patch_sql)
print(f"[scrobble_track] MM4 PlayCounter incremented and LastTimePlayed patched via SQL for track ID {track_id}")
loop.call_soon_threadsafe(_safe_set_result, future, True)
except (pythoncom.com_error, AttributeError) as e:
# Catch zombie objects (AttributeError) and drop the connection so it self-heals next request
logger.error(f"COM Error or Zombie Object: {e}")
self.sdb = None
loop.call_soon_threadsafe(_safe_set_exception,future, e)
except Exception as e:
logger.exception("Unexpected error in COM worker")
loop.call_soon_threadsafe(_safe_set_exception,future, e)
self.request_queue.task_done()
def submit(self, action, payload):
loop = asyncio.get_running_loop()
future = loop.create_future()
self.request_queue.put((action, payload, future, loop))
return future
com_worker = COMWorker()
# --- DB LAYER ---
db_lock = threading.Lock()
def _safe_set_result(future, value):
if not future.done():
future.set_result(value)
def _safe_set_exception(future, exc):
if not future.done():
future.set_exception(exc)
def _iunicode_collation(a, b):
a, b = (a or "").lower(), (b or "").lower()
return (a > b) - (a < b)
def _sync_query(query, params, fetchall):
with db_lock:
db_uri = f"file:{MM_DB_PATH}?mode=ro"
conn = sqlite3.connect(db_uri, uri=True, timeout=15.0)
conn.row_factory = sqlite3.Row
conn.create_collation("IUNICODE", _iunicode_collation)
try:
cur = conn.execute(query, params)
return cur.fetchall() if fetchall else cur.fetchone()
finally:
conn.close()
async def query_db(query: str, params: tuple = (), fetchall: bool = True):
return await asyncio.to_thread(_sync_query, query, params, fetchall)
def normalize_year(val):
if not val or val <= 0:
return None
s = str(val)
if len(s) >= 4:
try:
year = int(s[:4])
if 1000 <= year <= 9999:
return year
except ValueError:
pass
return val if val < 9999 else None
@asynccontextmanager
async def lifespan(app: FastAPI):
if not os.path.exists(MM_DB_PATH):
raise RuntimeError(f"Database not found at {MM_DB_PATH}")
com_worker.start()
rows = await query_db("SELECT IDMedia, DriveLetter FROM Medias WHERE DriveLetter IS NOT NULL")
for r in rows:
drive = r["DriveLetter"]
if isinstance(drive, bytes):
drive = drive.decode()
elif isinstance(drive, int):
drive = chr(ord("A") + drive)
drive = str(drive).strip().upper().rstrip(":\\")
if len(drive) == 1 and "A" <= drive <= "Z":
MEDIA_MAP[r["IDMedia"]] = f"{drive}:\\"
logger.info(f"Loaded Drive Mappings: {MEDIA_MAP}")
logger.info(f"Transcode mode: {CONFIG['transcode_mode']} -> {CONFIG['transcode_format']} @ {CONFIG['transcode_bitrate']}kbps")
yield
com_worker.request_queue.put(None)
com_worker.join()
app = FastAPI(title="MediaMonkey Subsonic Gateway", lifespan=lifespan)
@app.middleware("http")
async def strip_view_suffix(request: Request, call_next):
if request.url.path.startswith("/rest/") and request.url.path.endswith(".view"):
request.scope["path"] = request.url.path[:-5]
return await call_next(request)
# --- UTILS ---
def resolve_path(mm_path: str, id_media: int) -> str:
if id_media in MEDIA_MAP:
drive = MEDIA_MAP[id_media]
match = re.match(r"^.?:\\", mm_path)
if match:
return drive + mm_path[match.end():]
return mm_path
def safe_track_num(val):
if not val: return 1
try: return int(str(val).split("/")[0])
except (ValueError, TypeError): return 1
def format_subsonic_track(row):
path = row["SongPath"] or ""
ext = os.path.splitext(path)[1].lstrip(".").lower() or "mp3"
transcoding = needs_transcode(ext)
if transcoding:
out_ext = CONFIG["transcode_format"]
content_type_map = {"mp3": "audio/mpeg", "aac": "audio/mp4", "ogg": "audio/ogg"}
suffix = out_ext
content_type = content_type_map.get(out_ext, "audio/mpeg")
size = 0
bitrate = CONFIG["transcode_bitrate"]
else:
content_type_map = {
"mp3": "audio/mpeg", "flac": "audio/flac", "ogg": "audio/ogg",
"m4a": "audio/mp4", "wav": "audio/wav", "wma": "audio/x-ms-wma",
}
suffix = ext
content_type = content_type_map.get(ext, "audio/mpeg")
size = row["FileLength"] if "FileLength" in row.keys() and row["FileLength"] else 0
bitrate = int(row["Bitrate"] / 1000) if "Bitrate" in row.keys() and row["Bitrate"] else 0
return {
"id": f"TRACK_{row['ID']}",
"parent": f"ALBUM_{row['IDAlbum']}",
"isDir": False,
"title": row["SongTitle"] or "Unknown Title",
"album": row["Album"] or "Unknown Album",
"artist": row["Artist"] or "Unknown Artist",
"track": safe_track_num(row["TrackNumber"]),
"year": normalize_year(row["Year"]) if row["Year"] and row["Year"] > 0 else None,
"genre": row["Genre"] or "Unknown",
"duration": int(row["SongLength"] / 1000) if row["SongLength"] else 0,
"coverArt": f"TRACK_{row['ID']}",
"type": "music",
"suffix": suffix,
"contentType": content_type,
"size": size,
"bitRate": bitrate,
"transcodedSuffix": suffix if transcoding else None,
"transcodedContentType": content_type if transcoding else None,
}
# --- AUTHENTICATION ---
class SubsonicAuthError(Exception):
pass
@app.exception_handler(SubsonicAuthError)
async def subsonic_auth_exception_handler(request: Request, exc: SubsonicAuthError):
return JSONResponse(
status_code=200,
content={"subsonic-response": {"status": "failed", "version": API_VERSION,
"error": {"code": 40, "message": "Wrong username or password."}}}
)
def verify_auth(u: str, p: Optional[str] = None, t: Optional[str] = None, s: Optional[str] = None):
if t and s:
expected_token = hashlib.md5((VALID_PASS + s).encode()).hexdigest()
if u == VALID_USER and t == expected_token:
return True
elif u == VALID_USER and p:
if p.startswith("enc:"):
try:
p = bytes.fromhex(p[4:]).decode()
except Exception:
raise SubsonicAuthError()
if p == VALID_PASS:
return True
raise SubsonicAuthError()
# --- STUBS & BASE PROTOCOL ---
@app.get("/rest/ping")
@app.get("/rest/getLicense")
async def ping(auth = Depends(verify_auth)):
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION, "license": {"valid": True, "email": "local@user"}}})
@app.get("/rest/getUser")
async def get_user(username: str, auth = Depends(verify_auth)):
return JSONResponse({
"subsonic-response": {
"status": "ok", "version": API_VERSION,
"user": {
"username": username, "email": "local@user", "scrobblingEnabled": True,
"adminRole": True, "settingsRole": True, "downloadRole": True, "uploadRole": True,
"playlistRole": True, "coverArtRole": True, "commentRole": True, "podcastRole": True,
"streamRole": True, "jukeboxRole": True, "sharedRole": True
}
}
})
@app.get("/rest/getPodcasts")
@app.get("/rest/getInternetRadioStations")
@app.get("/rest/getBookmarks")
@app.get("/rest/getStarred2")
@app.get("/rest/search3")
async def stubs(auth = Depends(verify_auth)):
return JSONResponse({
"subsonic-response": {
"status": "ok", "version": API_VERSION,
"searchResult3": {"artist": [], "album": [], "song": []},
"genres": {"genre": []},
"starred2": {"song": [], "album": [], "artist": []}
}
})
# --- ID3 BROWSING ---
@app.get("/rest/getIndexes")
async def get_indexes(auth = Depends(verify_auth)):
rows = await query_db("SELECT ID, Artist FROM Artists WHERE Artist IS NOT NULL AND Artist != '' ORDER BY Artist")
# Added albumCount: 1 and lastModified to prevent offline clients from assuming an empty state
artists = [{"id": f"ARTIST_{r['ID']}", "name": r["Artist"], "albumCount": 1} for r in rows]
return JSONResponse({
"subsonic-response": {
"status": "ok", "version": API_VERSION,
"indexes": {
"lastModified": 1717200000000,
"index": [{"name": "Library", "artist": artists}]
}
}
})
@app.get("/rest/getArtists")
async def get_artists(auth = Depends(verify_auth)):
rows = await query_db("SELECT ID, Artist FROM Artists WHERE Artist IS NOT NULL AND Artist != '' ORDER BY Artist")
artists = [{"id": f"ARTIST_{r['ID']}", "name": r["Artist"], "albumCount": 1} for r in rows]
return JSONResponse({
"subsonic-response": {
"status": "ok", "version": API_VERSION,
"artists": {"index": [{"name": "Library", "artist": artists}]}
}
})
@app.get("/rest/getArtist")
async def get_artist(id: str, auth = Depends(verify_auth)):
artist_id = id.replace("ARTIST_", "")
# Optimization: Dropped the heavy Songs join, using MediaMonkey's pre-calculated Tracks column
query = """
SELECT al.ID, al.Album, al.Artist, al.Year, al.Tracks as songCount
FROM Albums al
JOIN ArtistsAlbums aa ON aa.IDAlbum = al.ID
WHERE aa.IDArtist = ?
ORDER BY al.Album
"""
rows = await query_db(query, (artist_id,))
albums = [{
"id": f"ALBUM_{r['ID']}",
"artistId": id,
"name": r["Album"],
"title": r["Album"],
"artist": r["Artist"],
"coverArt": f"ALBUM_{r['ID']}",
"songCount": r["songCount"] or 0,
"year": normalize_year(r["Year"]) if r["Year"] and r["Year"] > 0 else None,
"isDir": True
} for r in rows]
return JSONResponse({
"subsonic-response": {
"status": "ok",
"version": API_VERSION,
"artist": {
"id": id,
"name": albums[0]["artist"] if albums else "Unknown",
"albumCount": len(albums),
"album": albums
}
}
})
@app.get("/rest/getGenres")
async def get_genres(auth = Depends(verify_auth)):
rows = await query_db("SELECT GenreName, UsageCount FROM Genres WHERE GenreName IS NOT NULL AND GenreName != '' ORDER BY GenreName")
genres = [{"value": r["GenreName"], "songCount": r["UsageCount"] or 0, "albumCount": 0} for r in rows]
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION, "genres": {"genre": genres}}})
@app.get("/rest/getMusicFolders")
async def get_music_folders(auth = Depends(verify_auth)):
return JSONResponse({
"subsonic-response": {
"status": "ok", "version": API_VERSION,
"musicFolders": {"musicFolder": [{"id": 1, "name": "MediaMonkey Library"}]}
}
})
@app.get("/rest/getMusicDirectory")
async def get_music_directory(id: str, auth = Depends(verify_auth)):
if id == "1":
rows = await query_db("""
SELECT DISTINCT a.ID, a.Artist
FROM Artists a
JOIN ArtistsAlbums aa ON aa.IDArtist = a.ID
WHERE a.Artist IS NOT NULL AND a.Artist != ''
ORDER BY a.Artist
""")
child_nodes = [{
"id": f"ARTIST_{r['ID']}",
"parent": "1",
"isDir": True,
"title": r["Artist"],
} for r in rows]
elif id.startswith("ARTIST_"):
artist_id = id.replace("ARTIST_", "")
query = """
SELECT al.ID, al.Album, al.Artist
FROM Albums al
JOIN ArtistsAlbums aa ON aa.IDAlbum = al.ID
WHERE aa.IDArtist = ?
ORDER BY al.Album
"""
rows = await query_db(query, (artist_id,))
child_nodes = [{
"id": f"ALBUM_{r['ID']}",
"parent": id,
"isDir": True,
"title": r["Album"],
"artist": r["Artist"],
"coverArt": f"ALBUM_{r['ID']}"
} for r in rows]
elif id.startswith("ALBUM_"):
album_id = id.replace("ALBUM_", "")
query = "SELECT ID, SongTitle, TrackNumber, SongLength, Year, IDAlbum, Album, Artist, Genre, SongPath, FileLength, Bitrate FROM Songs WHERE IDAlbum = ?"
rows = await query_db(query, (album_id,))
child_nodes = sorted([format_subsonic_track(r) for r in rows], key=lambda t: t["track"])
for c in child_nodes: c["parent"] = id
else:
child_nodes = []
return JSONResponse({
"subsonic-response": {
"status": "ok", "version": API_VERSION,
"directory": {"id": id, "name": "Directory", "child": child_nodes}
}
})
@app.get("/rest/getSongsByGenre")
async def get_songs_by_genre(genre: str, count: int = 10, offset: int = 0, auth = Depends(verify_auth)):
query = """
SELECT DISTINCT s.ID, s.SongTitle, s.TrackNumber, s.SongLength, s.Year, s.IDAlbum, s.Album, s.Artist, s.Genre, s.SongPath, s.FileLength, s.Bitrate
FROM Songs s
JOIN GenresSongs gs ON gs.IDSong = s.ID
JOIN Genres g ON g.IDGenre = gs.IDGenre
WHERE g.GenreName = ?
ORDER BY s.Artist, s.Album, s.TrackNumber
LIMIT ? OFFSET ?
"""
rows = await query_db(query, (genre, count, offset))
songs = [format_subsonic_track(r) for r in rows]
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION, "songsByGenre": {"song": songs}}})
@app.get("/rest/getSong")
async def get_song(id: str, auth = Depends(verify_auth)):
clean_id = id.replace("TRACK_", "")
query = "SELECT ID, SongTitle, TrackNumber, SongLength, Year, IDAlbum, Album, Artist, Genre, SongPath, FileLength, Bitrate FROM Songs WHERE ID = ?"
row = await query_db(query, (clean_id,), fetchall=False)
if not row:
raise HTTPException(status_code=404, detail="Track not found")
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION, "song": format_subsonic_track(row)}})
@app.get("/rest/getAlbum")
async def get_album(id: str, auth = Depends(verify_auth)):
album_id = id.replace("ALBUM_", "")
query = "SELECT ID, SongTitle, TrackNumber, SongLength, Year, IDAlbum, Album, Artist, Genre, SongPath, FileLength, Bitrate FROM Songs WHERE IDAlbum = ?"
rows = await query_db(query, (album_id,))
tracks = sorted([format_subsonic_track(r) for r in rows], key=lambda t: t["track"])
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION, "album": {"id": id, "name": tracks[0]["album"] if tracks else "Unknown", "song": tracks}}})
@app.get("/rest/getAlbumList2")
async def get_album_list2(type: str = "newest", size: int = 500, offset: int = 0,
fromYear: Optional[int] = None, toYear: Optional[int] = None,
auth = Depends(verify_auth)):
order_map = {
"newest": "al.ID DESC", "recent": "al.ID DESC", "frequent": "al.ID DESC",
"random": "RANDOM()",
"alphabeticalByName": "al.Album ASC",
"alphabeticalByArtist": "al.Artist ASC, al.Album ASC",
"byYear": "al.Year ASC" if not (fromYear and toYear and fromYear > toYear) else "al.Year DESC",
}
order_clause = order_map.get(type, "al.ID DESC")
where_clause = ""
params = []
if type == "byYear" and fromYear is not None and toYear is not None:
lo, hi = sorted([fromYear, toYear])
where_clause = "WHERE al.Year BETWEEN ? AND ?"
params += [lo, hi]
query = f"""
SELECT DISTINCT al.ID, al.Album, al.Artist, al.Year, al.Tracks as songCount
FROM Albums al
{where_clause}
ORDER BY {order_clause}
LIMIT ? OFFSET ?
"""
params += [size, offset]
rows = await query_db(query, tuple(params))
albums = [{
"id": f"ALBUM_{r['ID']}",
"name": r["Album"] or "Unknown",
"title": r["Album"] or "Unknown",
"artist": r["Artist"] or "Unknown",
"coverArt": f"ALBUM_{r['ID']}",
"songCount": r["songCount"] or 0,
"year": normalize_year(r["Year"]) if r["Year"] and r["Year"] > 0 else None,
"isDir": True
} for r in rows]
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION, "albumList2": {"album": albums}}})
# --- PLAYLIST MANAGEMENT ---
@app.get("/rest/getPlaylists")
async def get_playlists(auth = Depends(verify_auth)):
rows = await query_db("SELECT IDPlaylist, PlaylistName, IsAutoPlaylist FROM Playlists WHERE PlaylistName IS NOT NULL")
playlists = [{"id": f"PLAYLIST_{r['IDPlaylist']}", "name": r["PlaylistName"], "comment": "Auto" if r['IsAutoPlaylist'] == 1 else "Static"} for r in rows]
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION, "playlists": {"playlist": playlists}}})
@app.get("/rest/getPlaylist")
async def get_playlist(id: str, auth = Depends(verify_auth)):
clean_id = id.replace("PLAYLIST_", "")
meta = await query_db("SELECT IsAutoPlaylist, PlaylistName FROM Playlists WHERE IDPlaylist = ?", (clean_id,), fetchall=False)
if not meta: raise HTTPException(status_code=404, detail="Playlist missing")
entries = []
if meta['IsAutoPlaylist'] == 1:
try:
com_tracks = await asyncio.wait_for(com_worker.submit("evaluate_playlist", {"id": clean_id}), timeout=300.0)
track_ids = [t["id"].replace("TRACK_", "") for t in com_tracks]
if track_ids:
placeholders = ",".join("?" * len(track_ids))
query = f"""
SELECT ID, SongTitle, TrackNumber, SongLength, Year, IDAlbum, Album, Artist, Genre, SongPath, FileLength, Bitrate
FROM Songs WHERE ID IN ({placeholders})
"""
rows = await query_db(query, tuple(track_ids))
rows_by_id = {str(r["ID"]): r for r in rows}
entries = [format_subsonic_track(rows_by_id[tid]) for tid in track_ids if tid in rows_by_id]
except asyncio.TimeoutError:
raise HTTPException(status_code=504, detail="MediaMonkey COM operation timed out")
else:
query = """
SELECT s.ID, s.SongTitle, s.TrackNumber, s.SongLength, s.Year, s.IDAlbum, s.Album, s.Artist, s.Genre, s.SongPath, s.FileLength, s.Bitrate
FROM PlaylistSongs ps
JOIN Songs s ON ps.IDSong = s.ID
WHERE ps.IDPlaylist = ? ORDER BY ps.SongOrder
"""
rows = await query_db(query, (clean_id,))
entries = [format_subsonic_track(r) for r in rows]
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION, "playlist": {"id": id, "name": meta["PlaylistName"], "entry": entries}}})
_playlist_cache = {}
_playlist_cache_ttl = 300 # seconds
async def get_playlist_cached(clean_id):
now = time.time()
if clean_id in _playlist_cache:
cached_at, tracks = _playlist_cache[clean_id]
if now - cached_at < _playlist_cache_ttl:
return tracks
tracks = await asyncio.wait_for(com_worker.submit("evaluate_playlist", {"id": clean_id}), timeout=180.0)
_playlist_cache[clean_id] = (now, tracks)
return tracks
# --- HYBRID COVER ART ENGINE ---
@app.get("/rest/getCoverArt")
async def get_cover_art(id: str, auth = Depends(verify_auth)):
if id in art_cache:
return Response(content=art_cache[id][0], media_type=art_cache[id][1])
track = None
if id.startswith("TRACK_"):
track = await query_db("SELECT ID, SongPath, IDMedia FROM Songs WHERE ID = ?", (id.replace("TRACK_", ""),), fetchall=False)
elif id.startswith("ALBUM_"):
track = await query_db("SELECT ID, SongPath, IDMedia FROM Songs WHERE IDAlbum = ? LIMIT 1", (id.replace("ALBUM_", ""),), fetchall=False)
elif id.startswith("ARTIST_"):
track = await query_db("""
SELECT s.ID, s.SongPath, s.IDMedia FROM Songs s
JOIN ArtistsSongs asng ON asng.IDSong = s.ID
WHERE asng.IDArtist = ? LIMIT 1
""", (id.replace("ARTIST_", ""),), fetchall=False)
if track:
real_path = resolve_path(track["SongPath"], track["IDMedia"])
cover_row = await query_db("SELECT CoverPath FROM Covers WHERE IDSong = ? AND CoverPath IS NOT NULL AND CoverPath != '' ORDER BY CoverOrder LIMIT 1", (track["ID"],), fetchall=False)
if cover_row:
c_path = cover_row["CoverPath"]
if not os.path.isabs(c_path):
c_path = os.path.join(os.path.dirname(real_path), c_path)
if os.path.exists(c_path):
with open(c_path, "rb") as f:
art_data = f.read()
mime = "image/png" if c_path.lower().endswith(".png") else "image/jpeg"
art_cache[id] = (art_data, mime)
return Response(content=art_data, media_type=mime)
if os.path.exists(real_path):
try:
audio = mutagen.File(real_path)
art_data, mime = None, None
if audio and audio.tags:
for tag in audio.tags.values():
if tag.__class__.__name__ == 'APIC':
art_data, mime = tag.data, tag.mime
break
if not art_data and hasattr(audio, "pictures") and audio.pictures:
art_data, mime = audio.pictures[0].data, audio.pictures[0].mime
if art_data:
art_cache[id] = (art_data, mime)
return Response(content=art_data, media_type=mime)
except Exception:
logger.exception(f"Mutagen failed to extract art for {real_path}")
raise HTTPException(status_code=404, detail="Artwork not found")
# --- STREAMING ENGINE ---
@app.get("/rest/stream")
async def stream_file(id: str, auth = Depends(verify_auth)):
clean_id = id.replace("TRACK_", "")
track = await query_db("SELECT SongPath, IDMedia FROM Songs WHERE ID = ?", (clean_id,), fetchall=False)
if not track: raise HTTPException(status_code=404, detail="Track missing from DB")
real_path = resolve_path(track["SongPath"], track["IDMedia"])
if not os.path.exists(real_path):
raise HTTPException(status_code=404, detail="File missing from disk")
ext = os.path.splitext(real_path)[1].lstrip(".").lower()
if needs_transcode(ext):
try:
cached_path = await asyncio.to_thread(get_or_create_transcode, clean_id, real_path)
except Exception as e:
logger.exception(f"Transcode failed for track {clean_id}")
raise HTTPException(status_code=500, detail=f"Transcode failed: {e}")
mime_map = {"mp3": "audio/mpeg", "aac": "audio/mp4", "ogg": "audio/ogg"}
media_type = mime_map.get(CONFIG["transcode_format"], "audio/mpeg")
return FileResponse(cached_path, media_type=media_type)
media_type = "audio/flac" if ext == "flac" else "audio/mpeg"
return FileResponse(real_path, media_type=media_type)
# --- METADATA WRITE-BACKS ---
@app.get("/rest/setRating")
async def set_rating(id: str, rating: int, auth = Depends(verify_auth)):
clamped_rating = max(0, min(rating, 5))
try:
await asyncio.wait_for(com_worker.submit("rate_track", {"id": id, "rating": clamped_rating}), timeout=10.0)
except asyncio.TimeoutError:
logger.warning("setRating timed out waiting for MediaMonkey COM")
except RuntimeError as e:
logger.warning(f"setRating skipped: {e}")
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION}})
@app.get("/rest/scrobble")
async def scrobble(id: str, auth = Depends(verify_auth)):
try:
await asyncio.wait_for(com_worker.submit("scrobble_track", {"id": id}), timeout=10.0)
except asyncio.TimeoutError:
logger.warning("Scrobble timed out waiting for MediaMonkey COM")
except RuntimeError as e:
logger.warning(f"Scrobble skipped: {e}")
return JSONResponse({"subsonic-response": {"status": "ok", "version": API_VERSION}})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)