music-assistant-server

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