217 lines
8.1 KiB
Python
217 lines
8.1 KiB
Python
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) |