/
/
/
1"""Helper class to aid scrobblers."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, ClassVar, cast
7
8from music_assistant_models.config_entries import (
9 Config,
10 ConfigEntry,
11 ConfigValueOption,
12 ConfigValueType,
13)
14from music_assistant_models.enums import ConfigEntryType, MediaType
15
16from music_assistant.helpers.config_entries import PLAYBACK_TARGET_TYPES
17
18if TYPE_CHECKING:
19 from music_assistant_models.event import MassEvent
20 from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
21
22 from music_assistant import MusicAssistant
23
24
25class ScrobblerHelper:
26 """Base class to aid scrobbling media items."""
27
28 logger: logging.Logger
29 config: ScrobblerConfig
30 supported_media_types: frozenset[MediaType] | None
31 currently_playing: str | None = None
32 last_scrobbled: str | None = None
33 # Exceptions the concrete scrobble client raises when a submission can't reach
34 # the service (network blips, service-side errors). Subclasses set this to their
35 # client library's error hierarchy so those are logged and swallowed, while any
36 # exception outside the set surfaces as the bug it is.
37 scrobble_exceptions: ClassVar[tuple[type[Exception], ...]] = ()
38
39 def __init__(
40 self,
41 logger: logging.Logger,
42 config: ScrobblerConfig | None = None,
43 supported_media_types: frozenset[MediaType] | None = None,
44 ) -> None:
45 """Initialize."""
46 self.logger = logger
47 self.config = config or ScrobblerConfig(suffix_version=False)
48 self.supported_media_types = supported_media_types
49
50 def get_name(self, report: MediaItemPlaybackProgressReport) -> str:
51 """Get the track name to use for scrobbling, possibly appended with version info."""
52 if self.config.suffix_version and report.version:
53 return f"{report.name} ({report.version})"
54
55 return report.name
56
57 def should_scrobble(self, report: MediaItemPlaybackProgressReport) -> bool:
58 """Determine if a track should be scrobbled, to be extended later."""
59 if self.last_scrobbled == report.uri:
60 self.logger.debug("skipped scrobbling due to duplicate event")
61 return False
62
63 # ideally we want more precise control
64 # but because the event is triggered every 30s
65 # and we don't have full queue details to determine
66 # the exact context in which the event was fired
67 # we can only rely on fully_played for now
68 return bool(report.fully_played)
69
70 def _is_configured(self) -> bool:
71 """Override if subclass needs specific configuration."""
72 return True
73
74 async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
75 """Send a Now Playing update to the scrobbling service."""
76
77 async def _scrobble(self, report: MediaItemPlaybackProgressReport) -> None:
78 """Scrobble."""
79
80 async def _on_mass_media_item_played(self, event: MassEvent) -> None:
81 """Media item has finished playing, we'll scrobble the item."""
82 if not self._is_configured():
83 return
84
85 report: MediaItemPlaybackProgressReport = event.data
86
87 if self.supported_media_types and report.media_type not in self.supported_media_types:
88 self.logger.debug("skipped scrobbling for unsupported media type %s", report.media_type)
89 return
90
91 # handle optional user_id filtering
92 if self.config.mass_userids and report.userid not in self.config.mass_userids:
93 self.logger.debug("skipped scrobbling for user %s due to user filter", report.userid)
94 return
95
96 # handle optional player_id filtering
97 if self.config.mass_playerids and report.player_id not in self.config.mass_playerids:
98 self.logger.debug(
99 "skipped scrobbling for player %s due to player filter", report.player_id
100 )
101 return
102
103 # poor mans attempt to detect a song on loop
104 if not report.fully_played and report.uri == self.last_scrobbled:
105 self.logger.debug(
106 "reset _last_scrobbled and _currently_playing because the song was restarted"
107 )
108 self.last_scrobbled = None
109 # reset currently playing to avoid it expiring when looping single songs
110 self.currently_playing = None
111
112 async def update_now_playing() -> None:
113 try:
114 await self._update_now_playing(report)
115 self.logger.debug(f"track {report.uri} marked as 'now playing'")
116 self.currently_playing = report.uri
117 except self.scrobble_exceptions:
118 self.logger.exception("Error while marking track as 'now playing'")
119
120 async def scrobble() -> None:
121 try:
122 await self._scrobble(report)
123 self.last_scrobbled = report.uri
124 except self.scrobble_exceptions:
125 self.logger.exception("Error while scrobbling track")
126
127 # update now playing if needed
128 if report.is_playing and (
129 self.currently_playing is None or self.currently_playing != report.uri
130 ):
131 await update_now_playing()
132
133 if self.should_scrobble(report):
134 await scrobble()
135
136
137CONF_VERSION_SUFFIX = "suffix_version"
138CONF_SCROBBLE_USERS = "scrobble_users"
139CONF_SCROBBLE_PLAYERS = "scrobble_players"
140
141
142class ScrobblerConfig:
143 """Shared configuration options for scrobblers."""
144
145 def __init__(
146 self,
147 suffix_version: bool,
148 mass_userids: list[str] | None = None,
149 mass_playerids: list[str] | None = None,
150 ) -> None:
151 """Initialize."""
152 self.suffix_version = suffix_version
153 self.mass_userids = mass_userids or []
154 self.mass_playerids = mass_playerids or []
155
156 @staticmethod
157 async def get_shared_config_entries(
158 mass: MusicAssistant, values: dict[str, ConfigValueType] | None
159 ) -> list[ConfigEntry]:
160 """Shared config entries."""
161 return [
162 ConfigEntry(
163 key=CONF_VERSION_SUFFIX,
164 type=ConfigEntryType.BOOLEAN,
165 required=True,
166 default_value=True,
167 value=values.get(CONF_VERSION_SUFFIX) if values else None,
168 ),
169 # User and player filter options for scrobbling providers
170 await create_scrobble_users_config_entry(mass),
171 create_scrobble_players_config_entry(mass),
172 ]
173
174 @staticmethod
175 def create_from_config(config: Config) -> ScrobblerConfig:
176 """Extract relevant shared config values."""
177 return ScrobblerConfig(
178 suffix_version=bool(config.get_value(CONF_VERSION_SUFFIX, True)),
179 mass_userids=cast("list[str]", config.get_value(CONF_SCROBBLE_USERS, [])),
180 mass_playerids=cast("list[str]", config.get_value(CONF_SCROBBLE_PLAYERS, [])),
181 )
182
183
184async def create_scrobble_users_config_entry(mass: MusicAssistant) -> ConfigEntry:
185 """Create a reusable configentry to specify a userlist for scrobbling providers."""
186 # User options for scrobble filtering
187 ma_user_list = await mass.webserver.auth.list_users() # excludes system users
188 ma_user_list = [user for user in ma_user_list if user.enabled]
189 user_options = [
190 ConfigValueOption(user.user_id, title=user.display_name or user.username)
191 for user in ma_user_list
192 ]
193 return ConfigEntry(
194 key=CONF_SCROBBLE_USERS,
195 type=ConfigEntryType.STRING,
196 required=False,
197 options=user_options,
198 multi_value=True,
199 default_value=[],
200 )
201
202
203def create_scrobble_players_config_entry(mass: MusicAssistant) -> ConfigEntry:
204 """Create a reusable configentry to specify a player list for scrobbling providers."""
205 ma_player_list = sorted(
206 mass.players.all_players(return_unavailable=True, return_disabled=False),
207 key=lambda player: player.display_name.lower(),
208 )
209 player_options = [
210 ConfigValueOption(player.player_id, title=player.display_name)
211 for player in ma_player_list
212 if player.type in PLAYBACK_TARGET_TYPES
213 ]
214 return ConfigEntry(
215 key=CONF_SCROBBLE_PLAYERS,
216 type=ConfigEntryType.STRING,
217 required=False,
218 options=player_options,
219 multi_value=True,
220 default_value=[],
221 )
222