import logging
from typing import TYPE_CHECKING
+from music_assistant_models.config_entries import (
+ Config,
+ ConfigEntry,
+ ConfigValueType,
+)
+from music_assistant_models.enums import ConfigEntryType
+
if TYPE_CHECKING:
from music_assistant_models.event import MassEvent
from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
"""Base class to aid scrobbling tracks."""
logger: logging.Logger
+ config: ScrobblerConfig
currently_playing: str | None = None
last_scrobbled: str | None = None
- def __init__(self, logger: logging.Logger) -> None:
+ def __init__(self, logger: logging.Logger, config: ScrobblerConfig | None = None) -> None:
"""Initialize."""
self.logger = logger
+ self.config = config or ScrobblerConfig(suffix_version=False)
def _is_configured(self) -> bool:
"""Override if subclass needs specific configuration."""
return True
+ def get_name(self, report: MediaItemPlaybackProgressReport) -> str:
+ """Get the track name to use for scrobbling, possibly appended with version info."""
+ if self.config.suffix_version and report.version:
+ return f"{report.name} ({report.version})"
+
+ return report.name
+
async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
"""Send a Now Playing update to the scrobbling service."""
# the exact context in which the event was fired
# we can only rely on fully_played for now
return bool(report.fully_played)
+
+
+CONF_VERSION_SUFFIX = "suffix_version"
+
+
+class ScrobblerConfig:
+ """Shared configuration options for scrobblers."""
+
+ def __init__(self, suffix_version: bool) -> None:
+ """Initialize."""
+ self.suffix_version = suffix_version
+
+ @staticmethod
+ def get_shared_config_entries(values: dict[str, ConfigValueType] | None) -> list[ConfigEntry]:
+ """Shared config entries."""
+ return [
+ ConfigEntry(
+ key=CONF_VERSION_SUFFIX,
+ type=ConfigEntryType.BOOLEAN,
+ label="Suffix version to track names",
+ required=True,
+ description="Whether to add the version as suffix to track names,"
+ "e.g. 'Amazing Track (Live)'.",
+ default_value=True,
+ value=values.get(CONF_VERSION_SUFFIX) if values else None,
+ )
+ ]
+
+ @staticmethod
+ def create_from_config(config: Config) -> ScrobblerConfig:
+ """Extract relevant shared config values."""
+ return ScrobblerConfig(bool(config.get_value(CONF_VERSION_SUFFIX, True)))
from music_assistant.constants import MASS_LOGGER_NAME
from music_assistant.helpers.auth import AuthenticationHelper
-from music_assistant.helpers.scrobbler import ScrobblerHelper
+from music_assistant.helpers.scrobbler import ScrobblerConfig, ScrobblerHelper
from music_assistant.mass import MusicAssistant
from music_assistant.models import ProviderInstanceType
from music_assistant.models.plugin import PluginProvider
await super().loaded_in_mass()
# subscribe to media_item_played event
- handler = LastFMEventHandler(self.network, self.logger)
+ handler = LastFMEventHandler(self.network, self.logger, self.config)
self._on_unload.append(
self.mass.subscribe(handler._on_mass_media_item_played, EventType.MEDIA_ITEM_PLAYED)
)
async def unload(self, is_removed: bool = False) -> None:
- """
- Handle unload/close of the provider.
+ """Handle unload/close of the provider.
Called when provider is deregistered (e.g. MA exiting or config reloading).
"""
network: pylast._Network
- def __init__(self, network: pylast._Network, logger: logging.Logger) -> None:
+ def __init__(
+ self, network: pylast._Network, logger: logging.Logger, config: ProviderConfig
+ ) -> None:
"""Initialize."""
- super().__init__(logger)
+ super().__init__(logger, ScrobblerConfig.create_from_config(config))
self.network = network
async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
await asyncio.to_thread(
self.network.update_now_playing,
report.artist,
- report.name,
+ self.get_name(report),
report.album,
duration=report.duration,
mbid=report.mbid,
await asyncio.to_thread(
self.network.scrobble,
report.artist or "unknown artist",
- report.name,
+ self.get_name(report),
int(time.time()),
report.album,
duration=report.duration,
provider = str(values.get(CONF_PROVIDER))
# collect all config entries to show
- entries: list[ConfigEntry] = [
+ entries: list[ConfigEntry] = ScrobblerConfig.get_shared_config_entries(values)
+ entries += [
ConfigEntry(
key=CONF_PROVIDER,
type=ConfigEntryType.STRING,
from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
from music_assistant_models.provider import ProviderManifest
-from music_assistant.helpers.scrobbler import ScrobblerHelper
+from music_assistant.helpers.scrobbler import ScrobblerConfig, ScrobblerHelper
from music_assistant.mass import MusicAssistant
from music_assistant.models import ProviderInstanceType
from music_assistant.models.plugin import PluginProvider
"""Call after the provider has been loaded."""
await super().loaded_in_mass()
- handler = ListenBrainzEventHandler(self._client, self.logger)
+ handler = ListenBrainzEventHandler(self._client, self.logger, self.config)
# subscribe to media_item_played event
self._on_unload.append(
class ListenBrainzEventHandler(ScrobblerHelper):
"""Handles the event handling."""
- def __init__(self, client: ListenBrainz, logger: logging.Logger) -> None:
+ def __init__(
+ self, client: ListenBrainz, logger: logging.Logger, config: ProviderConfig
+ ) -> None:
"""Initialize."""
- super().__init__(logger)
+ super().__init__(logger, ScrobblerConfig.create_from_config(config))
self._client = client
def _make_listen(self, report: MediaItemPlaybackProgressReport) -> Listen:
# https://pylistenbrainz.readthedocs.io/en/latest/api_ref.html#class-listen
return Listen(
- track_name=report.name,
+ track_name=self.get_name(report),
artist_name=report.artist,
artist_mbids=report.artist_mbids,
release_name=report.album,
) -> tuple[ConfigEntry, ...]:
"""Return Config entries to setup this provider."""
return (
+ *ScrobblerConfig.get_shared_config_entries(values),
ConfigEntry(
key=CONF_USER_TOKEN,
type=ConfigEntryType.SECURE_STRING,
from music_assistant_models.event import MassEvent
from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
-from music_assistant.helpers.scrobbler import ScrobblerHelper
+from music_assistant.helpers.scrobbler import ScrobblerConfig, ScrobblerHelper
class DummyHandler(ScrobblerHelper):
_tracked = 0
_now_playing = 0
- def __init__(self, logger: logging.Logger) -> None:
+ def __init__(self, logger: logging.Logger, config: ScrobblerConfig | None = None) -> None:
"""Initialize."""
- super().__init__(logger)
+ super().__init__(logger, config)
def _is_configured(self) -> bool:
return True
assert handler._now_playing == 0
+async def test_it_suffixes_the_version_if_enabled_and_available() -> None:
+ """Test that the track version is suffixed to the track name when enabled."""
+ report_with_version = create_report(version="Deluxe Edition").data
+ report_without_version = create_report(version=None).data
+
+ handler = DummyHandler(logging.getLogger(), ScrobblerConfig(suffix_version=True))
+ assert handler.get_name(report_with_version) == "track (Deluxe Edition)"
+ assert handler.get_name(report_without_version) == "track"
+
+ handler = DummyHandler(logging.getLogger(), ScrobblerConfig(suffix_version=False))
+ assert handler.get_name(report_with_version) == "track"
+ assert handler.get_name(report_without_version) == "track"
+
+
def create_report(
- duration: int, seconds_played: int, is_playing: bool = True, uri: str = "filesystem://track/1"
+ duration: int = 148,
+ seconds_played: int = 59,
+ is_playing: bool = True,
+ uri: str = "filesystem://track/1",
+ version: str | None = None,
) -> MassEvent:
"""Create the MediaItemPlaybackProgressReport and wrap it in a MassEvent."""
return wrap_event(
seconds_played=seconds_played,
fully_played=duration - seconds_played < 5,
is_playing=is_playing,
+ version=version,
)
)
projects / music-assistant-server.git / commitdiff