/
/
/
1"""Allows scrobbling of tracks back to the Subsonic media server."""
2
3import logging
4import time
5from collections.abc import Callable
6
7from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, ProviderConfig
8from music_assistant_models.enums import EventType, MediaType
9from music_assistant_models.errors import SetupFailedError
10from music_assistant_models.media_items import Audiobook, PodcastEpisode, Track
11from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
12from music_assistant_models.provider import ProviderManifest
13
14from music_assistant.helpers.scrobbler import ScrobblerHelper
15from music_assistant.helpers.uri import parse_uri
16from music_assistant.mass import MusicAssistant
17from music_assistant.models import ProviderInstanceType
18from music_assistant.models.plugin import PluginProvider
19from music_assistant.providers.opensubsonic.parsers import EP_CHAN_SEP
20from music_assistant.providers.opensubsonic.sonic_provider import OpenSonicProvider
21
22
23async def setup(
24 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
25) -> ProviderInstanceType:
26 """Initialize provider(instance) with given configuration."""
27 sonic_prov = mass.get_provider("opensubsonic")
28 if not sonic_prov or not isinstance(sonic_prov, OpenSonicProvider):
29 raise SetupFailedError("A Open Subsonic Music provider must be configured first.")
30
31 return SubsonicScrobbleProvider(mass, manifest, config)
32
33
34class SubsonicScrobbleProvider(PluginProvider):
35 """Plugin provider to support scrobbling of tracks."""
36
37 def __init__(
38 self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
39 ) -> None:
40 """Initialize MusicProvider."""
41 super().__init__(mass, manifest, config)
42 self._on_unload: list[Callable[[], None]] = []
43
44 async def loaded_in_mass(self) -> None:
45 """Call after the provider has been loaded."""
46 await super().loaded_in_mass()
47
48 handler = SubsonicScrobbleEventHandler(self.mass, self.logger)
49
50 # subscribe to media_item_played event
51 self._on_unload.append(
52 self.mass.subscribe(handler._on_mass_media_item_played, EventType.MEDIA_ITEM_PLAYED)
53 )
54
55 async def unload(self, is_removed: bool = False) -> None:
56 """
57 Handle unload/close of the provider.
58
59 Called when provider is deregistered (e.g. MA exiting or config reloading).
60 """
61 for unload_cb in self._on_unload:
62 unload_cb()
63
64
65class SubsonicScrobbleEventHandler(ScrobblerHelper):
66 """Handles the scrobbling event handling."""
67
68 def __init__(self, mass: MusicAssistant, logger: logging.Logger) -> None:
69 """Initialize."""
70 super().__init__(logger)
71 self.mass = mass
72
73 def _is_scrobblable_media_type(self, media_type: MediaType) -> bool:
74 """Return true if the given OpenSubsonic media type can be scrobbled, false otherwise."""
75 return media_type in (
76 MediaType.TRACK,
77 MediaType.AUDIOBOOK,
78 MediaType.PODCAST_EPISODE,
79 )
80
81 async def _get_subsonic_provider_and_item_id(
82 self, media_type: MediaType, provider_instance_id_or_domain: str, item_id: str
83 ) -> tuple[None | OpenSonicProvider, str]:
84 """Return a OpenSonicProvider or None if no subsonic provider, and the Subsonic item_id.
85
86 Returns:
87 Tuple[OpenSonicProvider | None, str]: The provider or None, and the Subsonic item_id.
88 """
89 if provider_instance_id_or_domain == "library":
90 # unwrap library item to check if we have a subsonic mapping...
91 library_item = await self.mass.music.get_library_item_by_prov_id(
92 media_type, item_id, provider_instance_id_or_domain
93 )
94 if library_item is None:
95 return None, item_id
96 assert isinstance(library_item, Track | Audiobook | PodcastEpisode)
97 for mapping in library_item.provider_mappings:
98 if mapping.provider_domain.startswith("opensubsonic"):
99 # found a subsonic mapping, proceed...
100 prov = self.mass.get_provider(mapping.provider_instance)
101 assert isinstance(prov, OpenSonicProvider)
102 # Because there is no way to retrieve a single podcast episode in vanilla
103 # subsonic, we have to carry around the channel id as well. See
104 # opensubsonic.parsers.parse_episode.
105 if isinstance(library_item, PodcastEpisode) and EP_CHAN_SEP in mapping.item_id:
106 _, ret_id = mapping.item_id.split(EP_CHAN_SEP)
107 else:
108 ret_id = mapping.item_id
109 return prov, ret_id
110 # no subsonic mapping has been found in library item, ignore...
111 return None, item_id
112 if provider_instance_id_or_domain.startswith("opensubsonic"):
113 # found a subsonic mapping, proceed...
114 prov = self.mass.get_provider(provider_instance_id_or_domain)
115 assert isinstance(prov, OpenSonicProvider)
116 if media_type == MediaType.PODCAST_EPISODE and EP_CHAN_SEP in item_id:
117 _, ret_id = item_id.split(EP_CHAN_SEP)
118 return prov, ret_id
119 return prov, item_id
120 # not an item from subsonic provider, ignore...
121 return None, item_id
122
123 async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
124 media_type, provider_instance_id_or_domain, item_id = await parse_uri(report.uri)
125 if not self._is_scrobblable_media_type(media_type):
126 return
127 prov, item_id = await self._get_subsonic_provider_and_item_id(
128 media_type, provider_instance_id_or_domain, item_id
129 )
130 if not prov:
131 return
132
133 try:
134 self.logger.info("scrobble play now event")
135 await prov.conn.scrobble(item_id, submission=False)
136 self.logger.debug("track %s marked as 'now playing'", report.uri)
137 self.currently_playing = report.uri
138 except Exception as err:
139 self.logger.exception(err)
140
141 async def _scrobble(self, report: MediaItemPlaybackProgressReport) -> None:
142 media_type, provider_instance_id_or_domain, item_id = await parse_uri(report.uri)
143 if not self._is_scrobblable_media_type(media_type):
144 return
145 prov, item_id = await self._get_subsonic_provider_and_item_id(
146 media_type, provider_instance_id_or_domain, item_id
147 )
148 if not prov:
149 return
150
151 try:
152 await prov.conn.scrobble(item_id, submission=True, listen_time=int(time.time()))
153 self.logger.debug("track %s marked as 'played'", report.uri)
154 self.last_scrobbled = report.uri
155 except Exception as err:
156 self.logger.exception(err)
157
158
159async def get_config_entries(
160 mass: MusicAssistant,
161 instance_id: str | None = None,
162 action: str | None = None,
163 values: dict[str, ConfigValueType] | None = None,
164) -> tuple[ConfigEntry, ...]:
165 """Return Config entries to setup this provider."""
166 # ruff: noqa: ARG001
167 return ()
168