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