import logging import os import platform import sys import time from typing import Optional import dbus import requests from PIL import Image import io from dbus import Interface from dbus.bus import BusConnection class ClementineScrobbler: dbus_section = 'org.mpris.MediaPlayer2.Player' def __init__(self, log: logging.Logger, property_interface_instance: Interface): self.log = log self.property_interface_instance = property_interface_instance self.last_album = None self.last_song_and_artist = None self.last_time_start = None self.status_cleared = False def update_status(self, run_once: bool = False) -> None: while True: self._update_status() if run_once: break time.sleep(1) def _update_status(self) -> None: self.log.debug("Reading data from Clementine.") metadata = self.property_interface_instance.Get(self.dbus_section, 'Metadata') position_s = self.property_interface_instance.Get(self.dbus_section, 'Position') / 1000000 playback_status = self.property_interface_instance.Get(self.dbus_section, 'PlaybackStatus') if playback_status in ('Stopped', 'Paused'): self._clear_status() elif playback_status == 'Playing': self._set_status(metadata, position_s) def _clear_status(self) -> None: if not self.status_cleared: self.log.info("Cleared status") self.last_album = None self.last_song_and_artist = None self.last_time_start = None self.status_cleared = True def _set_status(self, metadata: dict, position_s: float) -> None: if metadata.get('xesam:album') == 'Youtube': song_and_artist = f"{metadata['xesam:title']}" album = f"{metadata['xesam:artist'][0]}" else: song_and_artist = f"{metadata['xesam:title']} by {metadata['xesam:artist'][0]}" album = metadata.get('xesam:album') time_start = round(time.time() - position_s) if self._status_changed(album, song_and_artist, time_start): self._log_changes(album, song_and_artist, time_start) # Pass author and album_title to _upload_image image_url = self._upload_image(metadata['mpris:artUrl'], metadata['xesam:title'], metadata['xesam:artist'][0], album) self._send_scrobble(metadata, album, image_url) self._update_status_state(album, song_and_artist, time_start) self.log.info(f"Updated status to: {album} - {song_and_artist}") def _status_changed(self, album: str, song_and_artist: str, time_start: int) -> bool: return (album != self.last_album or song_and_artist != self.last_song_and_artist or time_start != self.last_time_start) def _log_changes(self, album: str, song_and_artist: str, time_start: int) -> None: if album != self.last_album: self.log.debug(f"Album changed from {self.last_album} to {album}") if song_and_artist != self.last_song_and_artist: self.log.debug(f"Song and artist changed from {self.last_song_and_artist} to {song_and_artist}") if time_start != self.last_time_start: self.log.debug(f"Time start changed from {self.last_time_start} to {time_start}") @staticmethod def _upload_image(image_file: str, song_title: str, author: str, album_title: str) -> Optional[str]: image_server = os.getenv('IMAGE_SERVER') image_server_auth = os.getenv('IMAGE_SERVER_AUTH') if image_server and image_server_auth: with Image.open(image_file.replace("file://", "")) as img: img.thumbnail((600, 600)) with io.BytesIO() as output: img.save(output, format=img.format) image_bytes = output.getvalue() headers = {'Authorization': f"Bearer {image_server_auth}"} image_server_url = f"{image_server}/create" params = { 'author': author, 'song_title': song_title, 'album_title': album_title } response = requests.post(image_server_url, headers=headers, data=image_bytes, params=params) if response.status_code != 200: logging.error(f"Failed to upload image to image server ({image_server_url}): {response.text}") return None return response.json()['url'] return None def _send_scrobble(self, metadata: dict, album: str, image_url: str) -> None: image_server = os.getenv('IMAGE_SERVER') auth_key = os.getenv('IMAGE_SERVER_AUTH') if not image_server or not auth_key: self.log.error("IMAGE_SERVER or IMAGE_SERVER_AUTH is not set in the environment variables.") return scrobble_endpoint = f"{image_server}/scrobble" data = { 'song_name': metadata['xesam:title'], 'album_name': album, 'cover_url': image_url, 'author': metadata['xesam:artist'][0], 'length_seconds': metadata['mpris:length'] // 1000000, # microseconds to seconds 'year': metadata.get('year'), 'rating': metadata.get('rating'), } headers = { 'Content-Type': 'application/json', 'Authorization': f"Bearer {auth_key}" } response = requests.post(scrobble_endpoint, json=data, headers=headers) if response.status_code == 201: self.log.info("Successfully sent scrobble data to the server.") else: self.log.error(f"Failed to send scrobble data {scrobble_endpoint}: {response.status_code} - {response.text}") def _update_status_state(self, album: str, song_and_artist: str, time_start: int) -> None: self.last_album = album self.last_song_and_artist = song_and_artist self.last_time_start = time_start self.status_cleared = False def get_log(): """ Initializes and returns a logger instance. Returns: A logger instance. """ logging.basicConfig(stream=sys.stdout, level=logging.INFO) log = logging.getLogger(__name__) log.info("Initializing.") return log def get_clementine(log: logging.Logger) -> [BusConnection, Interface]: """ Connects to the D-Bus interface of the Clementine music player. Args: log: A logger instance. Returns: A tuple containing the player object and the property interface object. """ bus = dbus.SessionBus() player_instance = bus.get_object('org.mpris.MediaPlayer2.clementine', '/org/mpris/MediaPlayer2') prop_iface_instance = dbus.Interface(player_instance, dbus_interface='org.freedesktop.DBus.Properties') log.info("Connected to Clementine.") return player_instance, prop_iface_instance def start_loop(log: logging.Logger, property_interface_instance: Interface, run_once: bool = False): """ Continuously updates the status via HTTP based on the currently playing song in Clementine. Args: log: A logger instance. property_interface_instance: A DBus property interface instance for the Clementine music player. run_once: A flag that only makes the loop run once (used in the tests) """ status_updater = StatusUpdater(log, property_interface_instance) status_updater.update_status(run_once=run_once) if __name__ == '__main__': if platform.system() != 'Linux': print("This program can only be run on Linux.") sys.exit(1) logger = get_log() while True: try: player, property_interface = get_clementine(logger) start_loop(logger, property_interface) except dbus.exceptions.DBusException as e: logger.error("Error communicating with Clementine. Reconnecting in 15s.") logger.error(e) time.sleep(15) except Exception as e: logger.error("An error occurred. Reconnecting in 15s.") logger.error(e) time.sleep(15)