initial commit

clementine_scrobbler.py

@@ -0,0 +1,217 @@
+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)

requirements.txt

@@ -0,0 +1,4 @@
+dbus-python==1.3.2
+pypresence==4.3.0
+requests==2.28.2
+pillow==9.5.0