music-assistant-server

1"""Tune-In music provider support for MusicAssistant."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6from urllib.parse import quote
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
9from music_assistant_models.enums import (
10    ConfigEntryType,
11    ContentType,
12    ImageType,
13    MediaType,
14    ProviderFeature,
15    StreamType,
16)
17from music_assistant_models.errors import InvalidDataError, LoginFailed, MediaNotFoundError
18from music_assistant_models.media_items import (
19    AudioFormat,
20    MediaItemImage,
21    ProviderMapping,
22    Radio,
23    SearchResults,
24    UniqueList,
25)
26from music_assistant_models.streamdetails import StreamDetails
27
28from music_assistant.constants import CONF_USERNAME
29from music_assistant.controllers.cache import use_cache
30from music_assistant.helpers.throttle_retry import Throttler
31from music_assistant.models.music_provider import MusicProvider
32
33if TYPE_CHECKING:
34    from collections.abc import AsyncGenerator
35
36    from music_assistant_models.config_entries import ProviderConfig
37    from music_assistant_models.provider import ProviderManifest
38
39    from music_assistant import MusicAssistant
40    from music_assistant.models import ProviderInstanceType
41
42
43CACHE_CATEGORY_STREAMS = 1
44
45SUPPORTED_FEATURES = {
46    ProviderFeature.LIBRARY_RADIOS,
47    ProviderFeature.BROWSE,
48    ProviderFeature.SEARCH,
49}
50
51
52async def setup(
53    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
54) -> ProviderInstanceType:
55    """Initialize provider(instance) with given configuration."""
56    if not config.get_value(CONF_USERNAME):
57        msg = "Username is invalid"
58        raise LoginFailed(msg)
59
60    return TuneInProvider(mass, manifest, config, SUPPORTED_FEATURES)
61
62
63async def get_config_entries(
64    mass: MusicAssistant,
65    instance_id: str | None = None,
66    action: str | None = None,
67    values: dict[str, ConfigValueType] | None = None,
68) -> tuple[ConfigEntry, ...]:
69    """
70    Return Config entries to setup this provider.
71
72    instance_id: id of an existing provider instance (None if new instance setup).
73    action: [optional] action key called from config entries UI.
74    values: the (intermediate) raw values for config entries sent with the action.
75    """
76    # ruff: noqa: ARG001
77    return (
78        ConfigEntry(
79            key=CONF_USERNAME,
80            type=ConfigEntryType.STRING,
81            label="Username",
82            required=True,
83        ),
84    )
85
86
87class TuneInProvider(MusicProvider):
88    """Provider implementation for Tune In."""
89
90    _throttler: Throttler
91
92    async def handle_async_init(self) -> None:
93        """Handle async initialization of the provider."""
94        self._throttler = Throttler(rate_limit=1, period=2)
95        username = self.config.get_value(CONF_USERNAME)
96        if isinstance(username, str) and "@" in username:
97            self.logger.warning(
98                "Email address detected instead of username, "
99                "it is advised to use the tunein username instead of email."
100            )
101
102    async def get_library_radios(self) -> AsyncGenerator[Radio, None]:
103        """Retrieve library/subscribed radio stations from the provider."""
104
105        async def parse_items(
106            items: list[dict[str, Any]], folder: str | None = None
107        ) -> AsyncGenerator[Radio, None]:
108            for item in items:
109                item_type = item.get("type", "")
110                if "unavailable" in item.get("key", ""):
111                    continue
112                if not item.get("is_available", True):
113                    continue
114                if item_type == "audio":
115                    if "preset_id" not in item:
116                        continue
117                    # each radio station can have multiple streams add each one as different quality
118                    stream_info = await self._get_stream_info(item["preset_id"])
119                    yield self._parse_radio(item, stream_info, folder)
120                elif item_type == "link" and item.get("item") == "url":
121                    # custom url
122                    try:
123                        yield self._parse_radio(item)
124                    except InvalidDataError as err:
125                        # there may be invalid custom urls, ignore those
126                        self.logger.warning(str(err))
127                elif item_type == "link":
128                    # stations are in sublevel (new style)
129                    if sublevel := await self.__get_data(item["URL"], render="json"):
130                        async for subitem in parse_items(sublevel["body"], item["text"]):
131                            yield subitem
132                elif item.get("children"):
133                    # stations are in sublevel (old style ?)
134                    async for subitem in parse_items(item["children"], item["text"]):
135                        yield subitem
136
137        data = await self.__get_data("Browse.ashx", c="presets")
138        if data and "body" in data:
139            async for item in parse_items(data["body"]):
140                yield item
141
142    @use_cache(3600 * 24 * 30)  # Cache for 30 days
143    async def get_radio(self, prov_radio_id: str) -> Radio:
144        """Get radio station details."""
145        if not prov_radio_id.startswith("http"):
146            if "--" in prov_radio_id:
147                # handle this for backwards compatibility
148                prov_radio_id = prov_radio_id.split("--")[0]
149            params = {"c": "composite", "detail": "listing", "id": prov_radio_id}
150            result = await self.__get_data("Describe.ashx", **params)
151            if result and result.get("body") and result["body"][0].get("children"):
152                item = result["body"][0]["children"][0]
153                stream_info = await self._get_stream_info(prov_radio_id)
154                return self._parse_radio(item, stream_info)
155        # fallback - e.g. for handle custom urls ...
156        async for radio in self.get_library_radios():
157            if radio.item_id == prov_radio_id:
158                return radio
159        msg = f"Item {prov_radio_id} not found"
160        raise MediaNotFoundError(msg)
161
162    def _parse_radio(
163        self,
164        details: dict[str, Any],
165        stream_info: list[dict[str, Any]] | None = None,
166        folder: str | None = None,
167    ) -> Radio:
168        """Parse Radio object from json obj returned from api."""
169        if "name" in details:
170            name = details["name"]
171        else:
172            # parse name from text attr
173            name = details["text"]
174            if " | " in name:
175                name = name.split(" | ")[1]
176            name = name.split(" (")[0]
177
178        if stream_info is not None:
179            # stream info is provided: parse first stream into provider mapping
180            # assuming here that the streams are sorted by quality (bitrate)
181            # and the first one is the best quality
182            preferred_stream = stream_info[0]
183            radio = Radio(
184                item_id=details["preset_id"],
185                provider=self.instance_id,
186                name=name,
187                provider_mappings={
188                    ProviderMapping(
189                        item_id=details["preset_id"],
190                        provider_domain=self.domain,
191                        provider_instance=self.instance_id,
192                        audio_format=AudioFormat(
193                            content_type=ContentType.try_parse(preferred_stream["media_type"]),
194                            bit_rate=preferred_stream.get("bitrate", 128),
195                        ),
196                        details=preferred_stream["url"],
197                        available=details.get("is_available", True),
198                    )
199                },
200            )
201        else:
202            # custom url (no stream object present)
203            radio = Radio(
204                item_id=details["URL"],
205                provider=self.instance_id,
206                name=name,
207                provider_mappings={
208                    ProviderMapping(
209                        item_id=details["URL"],
210                        provider_domain=self.domain,
211                        provider_instance=self.instance_id,
212                        audio_format=AudioFormat(
213                            content_type=ContentType.UNKNOWN,
214                        ),
215                        details=details["URL"],
216                        available=details.get("is_available", True),
217                    )
218                },
219            )
220
221        # preset number is used for sorting (not present at stream time)
222        preset_number = details.get("preset_number", 0)
223        radio.position = preset_number
224        if "text" in details:
225            radio.metadata.description = details["text"]
226        # image
227        if img := details.get("image") or details.get("logo"):
228            radio.metadata.images = UniqueList(
229                [
230                    MediaItemImage(
231                        type=ImageType.THUMB,
232                        path=img,
233                        provider=self.instance_id,
234                        remotely_accessible=True,
235                    )
236                ]
237            )
238        return radio
239
240    async def _get_stream_info(self, preset_id: str) -> list[dict[str, Any]]:
241        """Get stream info for a radio station."""
242        cached_data = await self.mass.cache.get(
243            preset_id, provider=self.instance_id, category=CACHE_CATEGORY_STREAMS
244        )
245        if cached_data is not None:
246            # We know from cache this is the right type
247            assert isinstance(cached_data, list)
248            return cached_data
249
250        data = await self.__get_data("Tune.ashx", id=preset_id)
251        if not data:
252            return []
253
254        body_data = data["body"]
255        assert isinstance(body_data, list)
256
257        await self.mass.cache.set(
258            key=preset_id,
259            data=body_data,
260            provider=self.instance_id,
261            category=CACHE_CATEGORY_STREAMS,
262        )
263        return body_data
264
265    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
266        """Get stream details for a radio station."""
267        if item_id.startswith("http"):
268            # custom url
269            return StreamDetails(
270                provider=self.instance_id,
271                item_id=item_id,
272                audio_format=AudioFormat(
273                    content_type=ContentType.UNKNOWN,
274                ),
275                media_type=MediaType.RADIO,
276                stream_type=StreamType.HTTP,
277                path=item_id,
278                allow_seek=False,
279                can_seek=False,
280            )
281        if "--" in item_id:
282            # handle this for backwards compatibility
283            item_id = item_id.split("--")[0]
284        if stream_info := await self._get_stream_info(item_id):
285            # assuming here that the streams are sorted by quality (bitrate)
286            # and the first one is the best quality
287            preferred_stream = stream_info[0]
288            return StreamDetails(
289                provider=self.instance_id,
290                item_id=item_id,
291                # set contenttype to unknown so ffmpeg can auto detect it
292                audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
293                media_type=MediaType.RADIO,
294                stream_type=StreamType.HTTP,
295                path=preferred_stream["url"],
296                allow_seek=False,
297                can_seek=False,
298            )
299        msg = f"Unable to retrieve stream details for {item_id}"
300        raise MediaNotFoundError(msg)
301
302    @use_cache(3600 * 24 * 7)  # Cache for 7 days
303    async def search(
304        self, search_query: str, media_types: list[MediaType], limit: int = 10
305    ) -> SearchResults:
306        """Perform search on Tune-in music provider."""
307        result = SearchResults()
308        if MediaType.RADIO not in media_types:
309            return result
310        params = {
311            "query": quote(search_query),
312            "formats": "ogg,aac,wma,mp3,hls",
313            "username": self.config.get_value(CONF_USERNAME),
314            "partnerId": "1",
315            "render": "json",
316        }
317        data = await self.__get_data("search.ashx", **params)
318        radios = []
319        if data and "body" in data:
320            count = 0
321            for item in data["body"]:
322                if item.get("type") == "audio" and "preset_id" in item:
323                    try:
324                        stream_info = await self._get_stream_info(item["preset_id"])
325                        radios.append(self._parse_radio(item, stream_info))
326                        count += 1
327                        if count >= limit:
328                            break
329                    except Exception as err:
330                        self.logger.debug("Failed to parse radio: %s", err)
331        result.radio = radios
332        return result
333
334    async def __get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any] | None:
335        """Get data from api."""
336        if endpoint.startswith("http"):
337            url = endpoint
338        else:
339            url = f"https://opml.radiotime.com/{endpoint}"
340            kwargs["formats"] = "ogg,aac,wma,mp3,hls"
341            kwargs["username"] = self.config.get_value(CONF_USERNAME)
342            kwargs["partnerId"] = "1"
343            kwargs["render"] = "json"
344        locale = self.mass.metadata.locale.replace("_", "-")
345        language = locale.split("-")[0]
346        headers = {"Accept-Language": f"{locale}, {language};q=0.9, *;q=0.5"}
347        async with (
348            self._throttler,
349            self.mass.http_session.get(url, params=kwargs, headers=headers, ssl=False) as response,
350        ):
351            result: Any = await response.json()
352            if not result or "error" in result:
353                self.logger.error(url)
354                self.logger.error(kwargs)
355                return None
356            assert isinstance(result, dict)
357            return result
358