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