music-assistant-server

1"""Emby Music Provider for MusicAssistant."""
2
3from __future__ import annotations
4
5import hashlib
6import socket
7from asyncio import TaskGroup
8from collections.abc import AsyncGenerator
9from typing import TYPE_CHECKING, Any
10from urllib.parse import urljoin
11
12from aiohttp import ClientResponseError
13from music_assistant_models.enums import (
14    MediaType,
15    ProviderFeature,
16    StreamType,
17)
18from music_assistant_models.errors import (
19    LoginFailed,
20    MediaNotFoundError,
21    ProviderPermissionDenied,
22)
23from music_assistant_models.media_items import (
24    Album,
25    Artist,
26    MediaItemType,
27    Playlist,
28    SearchResults,
29    Track,
30)
31from music_assistant_models.streamdetails import StreamDetails
32
33from music_assistant.controllers.cache import use_cache
34from music_assistant.mass import MusicAssistant
35from music_assistant.models import ProviderInstanceType
36from music_assistant.models.music_provider import MusicProvider
37from music_assistant.providers.emby.const import (
38    ALBUM_FIELDS,
39    ARTIST_FIELDS,
40    AUTH_ACCESS_TOKEN,
41    AUTH_USER,
42    ITEM_KEY_COLLECTION_TYPE,
43    ITEM_KEY_ID,
44    ITEM_KEY_MEDIA_STREAMS,
45    ITEM_KEY_RUNTIME_TICKS,
46    ITEM_LIMIT,
47    ITEMS,
48    TRACK_FIELDS,
49)
50from music_assistant.providers.emby.parsers import (
51    parse_album,
52    parse_artist,
53    parse_playlist,
54    parse_stream_details,
55    parse_track,
56)
57
58if TYPE_CHECKING:
59    from music_assistant_models.config_entries import (
60        ConfigEntry,
61        ProviderConfig,
62    )
63    from music_assistant_models.provider import ProviderManifest
64
65from music_assistant.constants import (
66    APPLICATION_NAME,
67    CONF_IP_ADDRESS,
68    CONF_PASSWORD,
69    CONF_USERNAME,
70)
71
72SUPPORTED_FEATURES = {
73    ProviderFeature.LIBRARY_ARTISTS,
74    ProviderFeature.LIBRARY_ALBUMS,
75    ProviderFeature.LIBRARY_TRACKS,
76    ProviderFeature.LIBRARY_PLAYLISTS,
77    ProviderFeature.BROWSE,
78    ProviderFeature.SEARCH,
79    ProviderFeature.ARTIST_ALBUMS,
80    ProviderFeature.ARTIST_TOPTRACKS,
81    ProviderFeature.SIMILAR_TRACKS,
82}
83
84
85async def setup(
86    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
87) -> ProviderInstanceType:
88    """Initialize provider(instance) with given configuration."""
89    return EmbyProvider(mass, manifest, config, SUPPORTED_FEATURES)
90
91
92class EmbyProvider(MusicProvider):
93    """Provider for an Emby music library (uses Emby REST API)."""
94
95    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
96        """Get configuration entries for provider setup."""
97        return ()
98
99    async def handle_async_init(self) -> None:
100        """Initialize provider(instance) with given configuration."""
101        username = str(self.get_setup_value(CONF_USERNAME))
102        password = str(self.get_setup_value(CONF_PASSWORD) or "")
103        self._base_url = str(self.get_setup_value(CONF_IP_ADDRESS)).rstrip("/") + "/"
104        self._session = self.mass.http_session
105
106        # stable device id
107        device_id = hashlib.sha256(f"{self.mass.server_id}+{username}".encode()).hexdigest()
108        self._device_id = device_id
109        self._device_name = socket.gethostname()
110
111        # authenticate against Emby /Users/AuthenticateByName
112        auth_url = urljoin(self._base_url, "Users/AuthenticateByName")
113        payload = {"Username": username, "Pw": password}
114        headers = {
115            "Accept": "application/json",
116            "X-Emby-Authorization": (
117                f'MediaBrowser Client="{APPLICATION_NAME}", '
118                f'Device="{self._device_name}", '
119                f'DeviceId="{device_id}", '
120                f'Version="{self.mass.version}"'
121            ),
122        }
123        try:
124            async with self._session.post(auth_url, json=payload, headers=headers) as resp:
125                resp.raise_for_status()
126                data = await resp.json()
127        except ClientResponseError as err:
128            if err.status == 401:
129                raise LoginFailed("Unauthorized: invalid credentials") from err
130            if err.status == 403:
131                raise ProviderPermissionDenied("Forbidden: insufficient permissions") from err
132            if err.status == 404:
133                raise MediaNotFoundError("Authentication endpoint not found") from err
134            raise
135
136        # store token and user id
137        token = data.get(AUTH_ACCESS_TOKEN)
138        user = data.get(AUTH_USER)
139        if not token or not user:
140            raise LoginFailed("Authentication failed: missing token/user in response")
141        self._token = token
142        self._user_id = user.get(ITEM_KEY_ID)
143        self._headers = {
144            "Accept": "application/json",
145            "X-Emby-Token": self._token,
146            "X-Emby-Authorization": (
147                f'MediaBrowser Client="{APPLICATION_NAME}", '
148                f'Device="{self._device_name}", '
149                f'DeviceId="{device_id}", '
150                f'Version="{self.mass.version}", '
151                f'Token="{self._token}"'
152            ),
153        }
154
155    @property
156    def is_streaming_provider(self) -> bool:
157        """Return True if provider supports streaming."""
158        return False
159
160    async def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
161        url = urljoin(self._base_url, path.lstrip("/"))
162        try:
163            async with self._session.get(url, headers=self._headers, params=params) as resp:
164                resp.raise_for_status()
165                return await resp.json()  # type: ignore[no-any-return]
166        except ClientResponseError as err:
167            if err.status == 401:
168                raise LoginFailed("Unauthorized: invalid credentials") from err
169            if err.status == 403:
170                raise ProviderPermissionDenied("Forbidden: insufficient permissions") from err
171            if err.status == 404:
172                raise MediaNotFoundError(f"Item {path} not found") from err
173            raise
174
175    async def _post(self, path: str, json: dict[str, Any] | None = None) -> None:
176        url = urljoin(self._base_url, path.lstrip("/"))
177        try:
178            async with self._session.post(url, headers=self._headers, json=json) as resp:
179                resp.raise_for_status()
180        except ClientResponseError as err:
181            if err.status == 401:
182                raise LoginFailed("Unauthorized: invalid credentials") from err
183            if err.status == 403:
184                raise ProviderPermissionDenied("Forbidden: insufficient permissions") from err
185            if err.status == 404:
186                raise MediaNotFoundError(f"Item {path} not found") from err
187            raise
188
189    async def _search_items(
190        self, search_query: str, include_types: str, fields: list[str], limit: int
191    ) -> list[dict[str, Any]]:
192        params = {
193            "SearchTerm": search_query,
194            "IncludeItemTypes": include_types,
195            "EnableUserData": "true",
196            "Fields": ",".join(fields or []),
197            "Limit": str(limit),
198            "Recursive": "true",
199        }
200        resp = await self._get(f"Users/{self._user_id}/Items", params=params)
201        return resp.get(ITEMS, [])  # type: ignore[no-any-return]
202
203    async def _search_track(self, search_query: str, limit: int) -> list[Track]:
204        items = await self._search_items(search_query, "Audio", TRACK_FIELDS, limit)
205        return [parse_track(self.instance_id, self, item) for item in items]
206
207    async def _search_album(self, search_query: str, limit: int) -> list[Album]:
208        albumname = search_query.split(" - ", 1)[1] if " - " in search_query else search_query
209        items = await self._search_items(albumname, "MusicAlbum", ALBUM_FIELDS, limit)
210        return [parse_album(self.instance_id, self, item) for item in items]
211
212    async def _search_artist(self, search_query: str, limit: int) -> list[Artist]:
213        items = await self._search_items(search_query, "MusicArtist", ARTIST_FIELDS, limit)
214        return [parse_artist(self.instance_id, self, item) for item in items]
215
216    async def _search_playlist(self, search_query: str, limit: int) -> list[Playlist]:
217        items = await self._search_items(search_query, "Playlist", [], limit)
218        return [parse_playlist(self.instance_id, self, item) for item in items]
219
220    @use_cache(60 * 15)
221    async def search(
222        self,
223        search_query: str,
224        media_types: list[MediaType],
225        limit: int = 20,
226    ) -> SearchResults:
227        """Search for media items in the Emby library."""
228        artists = None
229        albums = None
230        tracks = None
231        playlists = None
232
233        async with TaskGroup() as tg:
234            if MediaType.ARTIST in media_types:
235                artists = tg.create_task(self._search_artist(search_query, limit))
236            if MediaType.ALBUM in media_types:
237                albums = tg.create_task(self._search_album(search_query, limit))
238            if MediaType.TRACK in media_types:
239                tracks = tg.create_task(self._search_track(search_query, limit))
240            if MediaType.PLAYLIST in media_types:
241                playlists = tg.create_task(self._search_playlist(search_query, limit))
242
243        search_results = SearchResults()
244        if artists:
245            search_results.artists = artists.result()
246        if albums:
247            search_results.albums = albums.result()
248        if tracks:
249            search_results.tracks = tracks.result()
250        if playlists:
251            search_results.playlists = playlists.result()
252        return search_results
253
254    async def get_library_artists(self) -> AsyncGenerator[Artist]:
255        """Yield all artists from the music library."""
256        libs = await self._get_music_libraries()
257        for lib in libs:
258            params = {
259                "ParentId": lib[ITEM_KEY_ID],
260                "IncludeItemTypes": "MusicArtist",
261                "EnableUserData": "true",
262                "Fields": ",".join(ARTIST_FIELDS),
263                "Recursive": "true",
264            }
265            page = 0
266            while True:
267                params["StartIndex"] = str(page * ITEM_LIMIT)
268                params["Limit"] = ITEM_LIMIT
269                resp = await self._get("Artists", params=params)
270                items = resp.get(ITEMS, [])
271                if not items:
272                    break
273                for artist in items:
274                    yield parse_artist(self.instance_id, self, artist)
275                page += 1
276
277    async def get_library_albums(self) -> AsyncGenerator[Album]:
278        """Yield all albums from the music library."""
279        libs = await self._get_music_libraries()
280        for lib in libs:
281            params = {
282                "ParentId": lib[ITEM_KEY_ID],
283                "IncludeItemTypes": "MusicAlbum",
284                "EnableUserData": "true",
285                "Fields": ",".join(ALBUM_FIELDS),
286                "Recursive": "true",
287            }
288            page = 0
289            while True:
290                params["StartIndex"] = str(page * ITEM_LIMIT)
291                params["Limit"] = ITEM_LIMIT
292                resp = await self._get(f"Users/{self._user_id}/Items", params=params)
293                items = resp.get(ITEMS, [])
294                if not items:
295                    break
296                for album in items:
297                    yield parse_album(self.instance_id, self, album)
298                page += 1
299
300    async def get_library_tracks(self) -> AsyncGenerator[Track]:
301        """Yield all tracks from the music library."""
302        libs = await self._get_music_libraries()
303        for lib in libs:
304            params = {
305                "ParentId": lib[ITEM_KEY_ID],
306                "IncludeItemTypes": "Audio",
307                "EnableUserData": "true",
308                "Fields": ",".join(TRACK_FIELDS),
309                "Recursive": "true",
310            }
311            page = 0
312            while True:
313                params["StartIndex"] = str(page * ITEM_LIMIT)
314                params["Limit"] = ITEM_LIMIT
315                resp = await self._get(f"Users/{self._user_id}/Items", params=params)
316                items = resp.get(ITEMS, [])
317                if not items:
318                    break
319                for track in items:
320                    if not len(track.get(ITEM_KEY_MEDIA_STREAMS, [])):
321                        continue
322                    yield parse_track(self.instance_id, self, track)
323                page += 1
324
325    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
326        """Yield all playlists from the music library."""
327        libs = await self._get_music_libraries()
328        for lib in libs:
329            params = {
330                "ParentId": lib[ITEM_KEY_ID],
331                "IncludeItemTypes": "Playlist",
332                "EnableUserData": "true",
333                "Recursive": "true",
334            }
335            page = 0
336            while True:
337                params["StartIndex"] = str(page * ITEM_LIMIT)
338                params["Limit"] = ITEM_LIMIT
339                resp = await self._get(f"Users/{self._user_id}/Items", params=params)
340                items = resp.get(ITEMS, [])
341                if not items:
342                    break
343                for playlist in items:
344                    yield parse_playlist(self.instance_id, self, playlist)
345                page += 1
346
347    @use_cache(3600)
348    async def get_album(self, prov_album_id: str) -> Album:
349        """Get album by provider album id."""
350        album = await self._get(
351            f"Users/{self._user_id}/Items/{prov_album_id}",
352            params={
353                "EnableUserData": "true",
354                "Fields": ",".join(ALBUM_FIELDS),
355                "Recursive": "true",
356            },
357        )
358        return parse_album(self.instance_id, self, album)
359
360    @use_cache(3600)
361    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
362        """Get tracks for a given album by provider album id."""
363        params = {
364            "ParentId": prov_album_id,
365            "IncludeItemTypes": "Audio",
366            "EnableUserData": "true",
367            "Fields": ",".join(TRACK_FIELDS),
368            "Limit": ITEM_LIMIT,
369            "Recursive": "true",
370        }
371        resp = await self._get(f"Users/{self._user_id}/Items", params=params)
372        return [parse_track(self.instance_id, self, item) for item in resp.get(ITEMS, [])]
373
374    @use_cache(60 * 15)
375    async def get_artist(self, prov_artist_id: str) -> Artist:
376        """Get artist by provider artist id."""
377        artist_data = await self._get(
378            f"Users/{self._user_id}/Items/{prov_artist_id}",
379            params={"EnableUserData": "true", "Fields": ",".join(ARTIST_FIELDS)},
380        )
381
382        return parse_artist(self.instance_id, self, artist_data)
383
384    @use_cache(3600)
385    async def get_artist_toptracks(self, prov_artist_id: str, limit: int = 25) -> list[Track]:
386        """Get top tracks for a given artist by provider artist id."""
387        params = {
388            "ArtistIds": prov_artist_id,
389            "IncludeItemTypes": "Audio",
390            "EnableUserData": "true",
391            "Fields": ",".join(TRACK_FIELDS),
392            "Recursive": "true",
393            "Limit": str(limit),
394            "SortBy": "PlayCount",
395            "SortOrder": "Descending",
396        }
397        resp = await self._get(f"Users/{self._user_id}/Items", params=params)
398        return [parse_track(self.instance_id, self, item) for item in resp.get(ITEMS, [])]
399
400    @use_cache(60 * 15)
401    async def get_track(self, prov_track_id: str) -> Track:
402        """Get track by provider track id."""
403        track = await self._get(
404            f"Users/{self._user_id}/Items/{prov_track_id}",
405            params={"EnableUserData": "true", "Fields": ",".join(TRACK_FIELDS)},
406        )
407
408        return parse_track(self.instance_id, self, track)
409
410    @use_cache(60 * 15)
411    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
412        """Get playlist by provider playlist id."""
413        playlist = await self._get(
414            f"Users/{self._user_id}/Items/{prov_playlist_id}",
415            params={"EnableUserData": "true"},
416        )
417
418        return parse_playlist(self.instance_id, self, playlist)
419
420    @use_cache(3600)
421    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
422        """Get tracks for a given playlist by provider playlist id."""
423        result: list[Track] = []
424        params = {
425            "ParentId": prov_playlist_id,
426            "IncludeItemTypes": "Audio",
427            "EnableUserData": "true",
428            "Fields": ",".join(TRACK_FIELDS),
429            "Limit": ITEM_LIMIT,
430            "StartIndex": str(page * ITEM_LIMIT),
431        }
432        resp = await self._get(f"Users/{self._user_id}/Items", params=params)
433        for index, item in enumerate(resp.get(ITEMS, []), 1):
434            pos = (page * ITEM_LIMIT) + index
435            if track := parse_track(self.instance_id, self, item):
436                track.position = pos
437                result.append(track)
438
439        return result
440
441    @use_cache(3600)
442    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
443        """Get albums for a given artist by provider artist id."""
444        params = {
445            "AlbumArtistIds": prov_artist_id,
446            "IncludeItemTypes": "MusicAlbum",
447            "Fields": ",".join(ALBUM_FIELDS),
448            "EnableUserData": "true",
449            "Recursive": "true",
450        }
451        resp = await self._get(f"Users/{self._user_id}/Items", params=params)
452        return [parse_album(self.instance_id, self, album) for album in resp.get(ITEMS, [])]
453
454    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
455        """Get stream details for given item id and media type."""
456        track_data = await self._get(
457            f"Users/{self._user_id}/Items/{item_id}",
458            params={"EnableUserData": "true", "Fields": ",".join(TRACK_FIELDS)},
459        )
460
461        audio_format = parse_stream_details(track_data)
462
463        url = urljoin(self._base_url, f"Audio/{item_id}/universal")
464        params = {
465            "Container": audio_format.content_type,
466            "AudioCodec": audio_format.codec_type,
467            "AudioSampleRate": audio_format.sample_rate,
468            "AudioChannels": audio_format.channels,
469            "Static": "true",
470            "api_key": self._token,
471        }
472        query = "&".join([f"{k}={v}" for k, v in params.items()])
473
474        duration = int(
475            track_data.get(ITEM_KEY_RUNTIME_TICKS, 0) / 10000000
476        )  # Convert ticks to seconds
477
478        return StreamDetails(
479            item_id=item_id,
480            provider=self.instance_id,
481            audio_format=audio_format,
482            stream_type=StreamType.HTTP,
483            duration=duration,
484            path=f"{url}?{query}",
485            can_seek=True,
486            allow_seek=True,
487        )
488
489    @use_cache(3600)
490    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
491        """Get similar tracks."""
492        resp = await self._get(
493            f"Items/{prov_track_id}/Similar",
494            params={"Limit": str(limit), "Fields": ",".join(TRACK_FIELDS)},
495        )
496
497        return [parse_track(self.instance_id, self, t) for t in resp.get(ITEMS, [])]
498
499    async def _get_music_libraries(self) -> list[dict[str, Any]]:
500        resp = await self._get("Library/MediaFolders")
501        libs = resp.get(ITEMS, [])
502        result = []
503        for library in libs:
504            if ITEM_KEY_COLLECTION_TYPE in library:
505                collection_type = library.get(ITEM_KEY_COLLECTION_TYPE, "").lower()
506                if collection_type == "music":
507                    result.append(library)
508        return result
509
510    async def on_played(
511        self,
512        media_type: MediaType,
513        prov_item_id: str,
514        fully_played: bool,
515        position: int,
516        media_item: MediaItemType,
517        is_playing: bool = False,
518    ) -> None:
519        """Handle media item played event."""
520        if media_type != MediaType.TRACK:
521            return
522        if fully_played:
523            await self._post(f"Users/{self._user_id}/PlayedItems/{prov_item_id}")
524        if is_playing:
525            await self._post(
526                f"/Users/{self._user_id}/Items/{prov_item_id}/UserData",
527                json={"PlaybackPositionTicks": position * 10000000},
528            )
529        if not fully_played and position == 0:
530            await self._post(f"/Users/{self._user_id}/PlayedItems/{prov_item_id}/Delete")
531