tanner/navidrome-heart-monitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Abstract Navidrome Subsonic API calls and flatten logic

Browse Source 
Co-authored-by: aider (gemini/gemini-2.5-pro) <aider@aider.chat>
This commit is contained in:
Tanner Collin
2026-02-07 12:10:37 -07:00
parent 12e35de8d9
commit 334da4bd17
1 changed files with 30 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
main.py
View File

@@ -46,103 +46,83 @@ def parse_log_line(line):
return timestamp, song_id, is_starred
async def _call_subsonic_api(endpoint, **kwargs):
"""
A generic helper to call the Navidrome Subsonic API.
"""
navidrome_url = settings.NAVIDROME_URL
username = settings.NAVIDROME_USER
if not all([navidrome_url, username]):
logging.error("NAVIDROME_URL and NAVIDROME_USER must be set in settings.py.")
return None
salt = settings.SUBSONIC_SALT
token = settings.SUBSONIC_TOKEN
if not all([salt, token]):
password = settings.NAVIDROME_PASSWORD
if not password:
logging.error("Either (SUBSONIC_SALT and SUBSONIC_TOKEN) or NAVIDROME_PASSWORD must be set in settings.py.")
return None
# Subsonic API requires a salt and a token (md5(password + salt))
salt = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(10))
token = hashlib.md5((password + salt).encode('utf-8')).hexdigest()
params = {
'u': username,
't': token,
's': salt,
'v': '1.16.1',
'c': 'heart-monitor',
'f': 'json',
}
params.update(kwargs)
api_url = f"{navidrome_url.rstrip('/')}/rest/{endpoint}"
try:
async with aiohttp.ClientSession() as session:
async with session.get(api_url, params=params) as response:
response.raise_for_status()
data = await response.json()
except aiohttp.ClientError as e:
logging.error(f"Error calling Navidrome API endpoint {endpoint} for song {kwargs.get('id')}: {e}")
return None
if data.get('subsonic-response', {}).get('status') != 'ok':
logging.error(f"Failed to call Navidrome API endpoint {endpoint} for song {kwargs.get('id')}. Response: {data}")
return None
return data.get('subsonic-response', {})
async def navidrome_get_song_details(song_id):
"""
Gets song details from Navidrome's Subsonic API.
"""
navidrome_url = settings.NAVIDROME_URL
username = settings.NAVIDROME_USER
if not all([navidrome_url, username]):
logging.error("NAVIDROME_URL and NAVIDROME_USER must be set in settings.py.")
response = await _call_subsonic_api('getSong', id=song_id)
if response:
return response.get('song')
return None
salt = settings.SUBSONIC_SALT
token = settings.SUBSONIC_TOKEN
if not all([salt, token]):
password = settings.NAVIDROME_PASSWORD
if not password:
logging.error("Either (SUBSONIC_SALT and SUBSONIC_TOKEN) or NAVIDROME_PASSWORD must be set in settings.py.")
return None
# Subsonic API requires a salt and a token (md5(password + salt))
salt = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(10))
token = hashlib.md5((password + salt).encode('utf-8')).hexdigest()
params = {
'u': username,
't': token,
's': salt,
'v': '1.16.1',
'c': 'heart-monitor',
'f': 'json',
'id': song_id,
}
api_url = f"{navidrome_url.rstrip('/')}/rest/getSong"
try:
async with aiohttp.ClientSession() as session:
async with session.get(api_url, params=params) as response:
response.raise_for_status()
data = await response.json()
except aiohttp.ClientError as e:
logging.error(f"Error calling Navidrome getSong API for song {song_id}: {e}")
return None
if data.get('subsonic-response', {}).get('status') != 'ok':
logging.error(f"Failed to get song details for song {song_id}. Response: {data}")
return None
return data.get('subsonic-response', {}).get('song')
async def navidrome_set_rating(song_id):
navidrome_url = settings.NAVIDROME_URL
username = settings.NAVIDROME_USER
"""
Sets a 1-star rating for a song using Navidrome's Subsonic API.
"""
response = await _call_subsonic_api('setRating', id=song_id, rating=1)
return response is not None
if not all([navidrome_url, username]):
logging.error("NAVIDROME_URL and NAVIDROME_USER must be set in settings.py.")
return
salt = settings.SUBSONIC_SALT
token = settings.SUBSONIC_TOKEN
if not all([salt, token]):
password = settings.NAVIDROME_PASSWORD
if not password:
logging.error("Either (SUBSONIC_SALT and SUBSONIC_TOKEN) or NAVIDROME_PASSWORD must be set in settings.py.")
return
# Subsonic API requires a salt and a token (md5(password + salt))
salt = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(10))
token = hashlib.md5((password + salt).encode('utf-8')).hexdigest()
params = {
'u': username,
't': token,
's': salt,
'v': '1.16.1',
'c': 'heart-monitor',
'f': 'json',
'id': song_id,
'rating': 1,
}
api_url = f"{navidrome_url.rstrip('/')}/rest/setRating"
try:
async with aiohttp.ClientSession() as session:
async with session.get(api_url, params=params) as response:
response.raise_for_status()
data = await response.json()
except aiohttp.ClientError as e:
logging.error(f"Error calling Navidrome API for song {song_id}: {e}")
return
if data.get('subsonic-response', {}).get('status') != 'ok':
logging.error(f"Failed to set rating for song {song_id}. Response: {data}")
async def handle_star_unstar_event(song_id):
"""
Sets song rating to 1 and logs the song details.
"""
success = await navidrome_set_rating(song_id)
if not success:
# _call_subsonic_api already logs errors
return
details = await navidrome_get_song_details(song_id)
@@ -188,7 +168,7 @@ async def main():
logging.info(
f"Song {song_id} was starred and then unstarred within {STAR_UNSTAR_WINDOW.seconds} seconds."
)
await navidrome_set_rating(song_id)
await handle_star_unstar_event(song_id)
# Remove song from tracking after it has been unstarred
del starred_songs[song_id]