/
/
/
1"""Helper class to aid scrobblers."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, 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
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 tracks."""
25
26 logger: logging.Logger
27 config: ScrobblerConfig
28 currently_playing: str | None = None
29 last_scrobbled: str | None = None
30
31 def __init__(self, logger: logging.Logger, config: ScrobblerConfig | None = None) -> None:
32 """Initialize."""
33 self.logger = logger
34 self.config = config or ScrobblerConfig(suffix_version=False)
35
36 def _is_configured(self) -> bool:
37 """Override if subclass needs specific configuration."""
38 return True
39
40 def get_name(self, report: MediaItemPlaybackProgressReport) -> str:
41 """Get the track name to use for scrobbling, possibly appended with version info."""
42 if self.config.suffix_version and report.version:
43 return f"{report.name} ({report.version})"
44
45 return report.name
46
47 async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
48 """Send a Now Playing update to the scrobbling service."""
49
50 async def _scrobble(self, report: MediaItemPlaybackProgressReport) -> None:
51 """Scrobble."""
52
53 async def _on_mass_media_item_played(self, event: MassEvent) -> None:
54 """Media item has finished playing, we'll scrobble the track."""
55 if not self._is_configured():
56 return
57
58 report: MediaItemPlaybackProgressReport = event.data
59
60 # handle optional user_id filtering
61 if self.config.mass_userids and report.userid not in self.config.mass_userids:
62 self.logger.debug("skipped scrobbling for user %s due to user filter", report.userid)
63 return
64
65 # poor mans attempt to detect a song on loop
66 if not report.fully_played and report.uri == self.last_scrobbled:
67 self.logger.debug(
68 "reset _last_scrobbled and _currently_playing because the song was restarted"
69 )
70 self.last_scrobbled = None
71 # reset currently playing to avoid it expiring when looping single songs
72 self.currently_playing = None
73
74 async def update_now_playing() -> None:
75 try:
76 await self._update_now_playing(report)
77 self.logger.debug(f"track {report.uri} marked as 'now playing'")
78 self.currently_playing = report.uri
79 except Exception as err:
80 # TODO: try to make this a more specific exception instead of a generic one
81 self.logger.exception(err)
82
83 async def scrobble() -> None:
84 try:
85 await self._scrobble(report)
86 self.last_scrobbled = report.uri
87 except Exception as err:
88 # TODO: try to make this a more specific exception instead of a generic one
89 self.logger.exception(err)
90
91 # update now playing if needed
92 if report.is_playing and (
93 self.currently_playing is None or self.currently_playing != report.uri
94 ):
95 await update_now_playing()
96
97 if self.should_scrobble(report):
98 await scrobble()
99
100 def should_scrobble(self, report: MediaItemPlaybackProgressReport) -> bool:
101 """Determine if a track should be scrobbled, to be extended later."""
102 if self.last_scrobbled == report.uri:
103 self.logger.debug("skipped scrobbling due to duplicate event")
104 return False
105
106 # ideally we want more precise control
107 # but because the event is triggered every 30s
108 # and we don't have full queue details to determine
109 # the exact context in which the event was fired
110 # we can only rely on fully_played for now
111 return bool(report.fully_played)
112
113
114CONF_VERSION_SUFFIX = "suffix_version"
115CONF_SCROBBLE_USERS = "scrobble_users"
116
117
118class ScrobblerConfig:
119 """Shared configuration options for scrobblers."""
120
121 def __init__(self, suffix_version: bool, mass_userids: list[str] | None = None) -> None:
122 """Initialize."""
123 self.suffix_version = suffix_version
124 self.mass_userids = mass_userids or []
125
126 @staticmethod
127 def get_shared_config_entries(values: dict[str, ConfigValueType] | None) -> list[ConfigEntry]:
128 """Shared config entries."""
129 return [
130 ConfigEntry(
131 key=CONF_VERSION_SUFFIX,
132 type=ConfigEntryType.BOOLEAN,
133 label="Suffix version to track names",
134 required=True,
135 description="Whether to add the version as suffix to track names,"
136 "e.g. 'Amazing Track (Live)'.",
137 default_value=True,
138 value=values.get(CONF_VERSION_SUFFIX) if values else None,
139 )
140 ]
141
142 @staticmethod
143 def create_from_config(config: Config) -> ScrobblerConfig:
144 """Extract relevant shared config values."""
145 return ScrobblerConfig(
146 suffix_version=bool(config.get_value(CONF_VERSION_SUFFIX, True)),
147 mass_userids=cast("list[str]", config.get_value(CONF_SCROBBLE_USERS, [])),
148 )
149
150
151async def create_scrobble_users_config_entry(mass: MusicAssistant) -> ConfigEntry:
152 """Create a reusable configentry to specify a userlist for scrobbling providers."""
153 # User options for scrobble filtering
154 ma_user_list = await mass.webserver.auth.list_users() # excludes system users
155 ma_user_list = [user for user in ma_user_list if user.enabled]
156 user_options = [
157 ConfigValueOption(title=user.display_name or user.username, value=user.user_id)
158 for user in ma_user_list
159 ]
160 return ConfigEntry(
161 key=CONF_SCROBBLE_USERS,
162 type=ConfigEntryType.STRING,
163 label="Scrobble for users",
164 required=False,
165 description="Only register scrobbles for the selected users. "
166 "Leave empty to scrobble for all users.",
167 options=user_options,
168 multi_value=True,
169 default_value=[],
170 )
171