From e954c28aa5e21dba0fa1a6d858d78d3e41578f7e Mon Sep 17 00:00:00 2001 From: anon Date: Wed, 9 Oct 2024 23:36:48 +0000 Subject: [PATCH] initial commit --- clementine_scrobbler.py | 217 ++++++++++++++++++++++++++++++++++++++++ readme.md | 0 requirements.txt | 4 + 3 files changed, 221 insertions(+) create mode 100644 clementine_scrobbler.py create mode 100644 readme.md create mode 100644 requirements.txt diff --git a/clementine_scrobbler.py b/clementine_scrobbler.py new file mode 100644 index 0000000..70c710e --- /dev/null +++ b/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) \ No newline at end of file diff --git a/readme.md b/readme.md new file mode 100644 index 0000000..e69de29 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..abc13f5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +dbus-python==1.3.2 +pypresence==4.3.0 +requests==2.28.2 +pillow==9.5.0 \ No newline at end of file