navidrome-heart-monitor/main.py

import os, logging
DEBUG = os.environ.get('DEBUG')
logging.basicConfig(
        format='[%(asctime)s] %(levelname)s %(module)s/%(funcName)s - %(message)s',
        level=logging.DEBUG if DEBUG else logging.INFO)

import asyncio
import re
from datetime import datetime, timedelta, timezone
import aiodocker
import aiohttp
import random
import string
import hashlib

import settings

STAR_UNSTAR_WINDOW = timedelta(seconds=4)

# A dictionary to store the timestamp of when a song was starred.
# {song_id: timestamp}
starred_songs = {}

LOG_PATTERN = re.compile(
    r'time="(?P<ts>[^"]+)".*msg="Changing starred" ids="\[`(?P<id>[^`]+)`\]".*starred=(?P<starred>true|false)'
)


def parse_log_line(line):
    """
    Parses a log line to find song star/unstar events.
    Returns a tuple of (timestamp, song_id, is_starred) or None if not a match.
    """
    match = LOG_PATTERN.search(line)
    if not match:
        return None

    data = match.groupdict()
    ts_str = data['ts']
    song_id = data['id']
    is_starred = data['starred'] == 'true'

    # fromisoformat doesn't like Z, so we replace it with +00:00 timezone info.
    timestamp = datetime.fromisoformat(ts_str.replace('Z', '+00:00'))

    return timestamp, song_id, is_starred


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.")
        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

    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}")
        return

    details = await navidrome_get_song_details(song_id)
    if not details:
        logging.info(f"Successfully set rating for song {song_id} to 1, but couldn't get song details.")
        return

    title = details.get('title', 'Unknown Title')
    artist = details.get('artist', 'Unknown Artist')
    album = details.get('album', 'Unknown Album')
    logging.info(f'Set song "{title}" - {artist} ({album}) rating to 1.')


async def main():
    """
    Monitors Navidrome container logs for rapid star/unstar events.
    """
    docker = None
    try:
        docker = aiodocker.Docker()
        container = await docker.containers.get(settings.NAVIDROME_CONTAINER)

        logging.info(f"Monitoring logs for container '{settings.NAVIDROME_CONTAINER}'...")
        logs = container.log(
            stdout=True,
            stderr=True,
            follow=True,
            since=datetime.now(timezone.utc).timestamp()
        )

        async for line in logs:
            parsed = parse_log_line(line)
            if not parsed:
                continue

            timestamp, song_id, is_starred = parsed

            if is_starred:
                starred_songs[song_id] = timestamp
            elif song_id in starred_songs:  # is_starred is False (unstarred)
                time_diff = timestamp - starred_songs[song_id]
                if time_diff <= STAR_UNSTAR_WINDOW:
                    logging.info(
                        f"Song {song_id} was starred and then unstarred within {STAR_UNSTAR_WINDOW.seconds} seconds."
                    )
                    await navidrome_set_rating(song_id)
                # Remove song from tracking after it has been unstarred
                del starred_songs[song_id]

    except aiodocker.exceptions.DockerError as e:
        if e.status == 404:
            logging.error(f"Container '{settings.NAVIDROME_CONTAINER}' not found.")
        else:
            logging.error(f"Error connecting to Docker or getting container: {e}")
    except Exception as e:
        logging.error(f"An unexpected error occurred: {e}")
    finally:
        if docker:
            await docker.close()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("Exiting.")