/
/
1"""Tune-In music provider support for MusicAssistant."""
2
3from __future__ import annotations
4
5import random
6from typing import TYPE_CHECKING, Any
7from urllib.parse import unquote
8
9from music_assistant_models.enums import (
10 ContentType,
11 ImageType,
12 MediaType,
13 ProviderFeature,
14 StreamType,
15)
16from music_assistant_models.errors import InvalidDataError, LoginFailed, MediaNotFoundError
17from music_assistant_models.media_items import (
18 AudioFormat,
19 BrowseFolder,
20 ItemMapping,
21 MediaItemImage,
22 MediaItemType,
23 ProviderMapping,
24 Radio,
25 RecommendationFolder,
26 SearchResults,
27 UniqueList,
28)
29from music_assistant_models.streamdetails import StreamDetails
30
31from music_assistant.constants import CONF_USERNAME
32from music_assistant.controllers.cache import use_cache
33from music_assistant.helpers.throttle_retry import Throttler
34from music_assistant.models.music_provider import MusicProvider
35
36if TYPE_CHECKING:
37 from collections.abc import AsyncGenerator
38
39 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
40 from music_assistant_models.provider import ProviderManifest
41
42 from music_assistant import MusicAssistant
43 from music_assistant.models import ProviderInstanceType
44
45
46CACHE_CATEGORY_STREAMS = 1
47CACHE_CATEGORY_BROWSE_MAP = 2
48
49SUPPORTED_FEATURES = {
50 ProviderFeature.LIBRARY_RADIOS,
51 ProviderFeature.BROWSE,
52 ProviderFeature.SEARCH,
53 ProviderFeature.RECOMMENDATIONS,
54}
55
56
57async def setup(
58 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
59) -> ProviderInstanceType:
60 """Initialize provider(instance) with given configuration."""
61 return TuneInProvider(mass, manifest, config, SUPPORTED_FEATURES)
62
63
64class TuneInProvider(MusicProvider):
65 """Provider implementation for Tune In."""
66
67 _throttler: Throttler
68 _browse_url_map: dict[str, str]
69
70 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
71 """Return Config entries to configure this provider."""
72 return ()
73
74 async def handle_async_init(self) -> None:
75 """Handle async initialization of the provider."""
76 self._throttler = Throttler(rate_limit=1, period=2)
77 self._browse_url_map = {}
78 username = self.get_setup_value(CONF_USERNAME)
79 if not username:
80 msg = "Username is invalid"
81 raise LoginFailed(msg)
82 if isinstance(username, str) and "@" in username:
83 self.logger.warning(
84 "Email address detected instead of username, "
85 "it is advised to use the tunein username instead of email."
86 )
87
88 async def browse(self, path: str) -> list[MediaItemType | BrowseFolder]:
89 """Browse TuneIn catalog."""
90 instance_base = path.split("://", 1)[0] + "://"
91 sub_path = path.split("://", 1)[1] if "://" in path else ""
92
93 if not sub_path:
94 # top-level: fetch TuneIn catalog root categories
95 data = await self.__get_data("Browse.ashx")
96 if not data or "body" not in data:
97 return []
98 result: list[MediaItemType | BrowseFolder] = []
99 for item in data["body"]:
100 if item.get("type") == "link" and item.get("key") != "podcast":
101 key = self._browse_path_key(item["text"])
102 folder_path = f"{instance_base}{key}"
103 self._browse_url_map[folder_path] = item["URL"]
104 await self.mass.cache.set(
105 key=folder_path,
106 data=item["URL"],
107 provider=self.instance_id,
108 category=CACHE_CATEGORY_BROWSE_MAP,
109 )
110 result.append(
111 BrowseFolder(
112 item_id=item["key"],
113 provider=self.instance_id,
114 path=folder_path,
115 name=item["text"],
116 )
117 )
118 return result
119
120 # sub-level: resolve TuneIn URL from map (populated during navigation)
121 tunein_url = self._browse_url_map.get(path)
122 if not tunein_url:
123 # try persistent cache so deep links work after a restart
124 tunein_url = await self.mass.cache.get(
125 path, provider=self.instance_id, category=CACHE_CATEGORY_BROWSE_MAP
126 )
127 if tunein_url:
128 self._browse_url_map[path] = tunein_url
129 if not tunein_url:
130 # backward-compat: old paths had the URL-encoded TuneIn URL as sub_path
131 if "%3A" in sub_path or sub_path.startswith("http"):
132 tunein_url = unquote(sub_path)
133 else:
134 return []
135
136 data = await self.__get_data(tunein_url, render="json")
137 if not data or "body" not in data:
138 return []
139
140 result = []
141 for item in data["body"]:
142 item_type = item.get("type", "")
143 if item_type == "audio" and "preset_id" in item:
144 result.append(self._parse_radio_lazy(item))
145 elif item_type == "link" and item.get("key") != "podcast":
146 key = item.get("key") or self._browse_path_key(item["text"])
147 folder_path = f"{path}/{key}"
148 self._browse_url_map[folder_path] = item["URL"]
149 await self.mass.cache.set(
150 key=folder_path,
151 data=item["URL"],
152 provider=self.instance_id,
153 category=CACHE_CATEGORY_BROWSE_MAP,
154 )
155 result.append(
156 BrowseFolder(
157 item_id=key,
158 provider=self.instance_id,
159 path=folder_path,
160 name=item["text"],
161 )
162 )
163 elif item.get("children"):
164 # inline children group (e.g. "Stations", "Local Stations" on genre pages)
165 for child in item["children"]:
166 if child.get("type") == "audio" and "preset_id" in child:
167 result.append(self._parse_radio_lazy(child))
168 return result
169
170 async def get_library_radios(self) -> AsyncGenerator[Radio]:
171 """Retrieve library/subscribed radio stations from the provider."""
172
173 async def parse_items(
174 items: list[dict[str, Any]], folder: str | None = None
175 ) -> AsyncGenerator[Radio]:
176 for item in items:
177 item_type = item.get("type", "")
178 if "unavailable" in item.get("key", ""):
179 continue
180 if not item.get("is_available", True):
181 continue
182 if item_type == "audio":
183 if "preset_id" not in item:
184 continue
185 # each radio station can have multiple streams add each one as different quality
186 stream_info = await self._get_stream_info(item["preset_id"])
187 yield self._parse_radio(item, stream_info, folder)
188 elif item_type == "link" and item.get("item") == "url":
189 # custom url
190 try:
191 yield self._parse_radio(item)
192 except InvalidDataError as err:
193 # there may be invalid custom urls, ignore those
194 self.logger.warning(str(err))
195 elif item_type == "link":
196 # stations are in sublevel (new style)
197 if sublevel := await self.__get_data(item["URL"], render="json"):
198 async for subitem in parse_items(sublevel["body"], item["text"]):
199 yield subitem
200 elif item.get("children"):
201 # stations are in sublevel (old style ?)
202 async for subitem in parse_items(item["children"], item["text"]):
203 yield subitem
204
205 data = await self.__get_data("Browse.ashx", c="presets")
206 if data and "body" in data:
207 async for item in parse_items(data["body"]):
208 yield item
209
210 @use_cache(3600 * 24 * 30) # Cache for 30 days
211 async def get_radio(self, prov_radio_id: str) -> Radio:
212 """Get radio station details."""
213 if not prov_radio_id.startswith("http"):
214 if "--" in prov_radio_id:
215 # handle this for backwards compatibility
216 prov_radio_id = prov_radio_id.split("--", maxsplit=1)[0]
217 params = {"c": "composite", "detail": "listing", "id": prov_radio_id}
218 result = await self.__get_data("Describe.ashx", **params)
219 if result and result.get("body") and result["body"][0].get("children"):
220 item = result["body"][0]["children"][0]
221 stream_info = await self._get_stream_info(prov_radio_id)
222 return self._parse_radio(item, stream_info)
223 # fallback - e.g. for handle custom urls ...
224 async for radio in self.get_library_radios():
225 if radio.item_id == prov_radio_id:
226 return radio
227 msg = f"Item {prov_radio_id} not found"
228 raise MediaNotFoundError(msg)
229
230 async def get_recommendations(self) -> list[RecommendationFolder]:
231 """Get this provider's available recommendation rows, without items."""
232 return [
233 RecommendationFolder(
234 item_id="trending",
235 provider=self.instance_id,
236 name="Trending",
237 translation_key="trending_stations",
238 )
239 ]
240
241 async def get_recommendation_items(
242 self, item_id: str
243 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
244 """
245 Get the items for a single recommendation row.
246
247 :param item_id: The item_id of the row, as returned by get_recommendations.
248 """
249 if item_id != "trending":
250 return UniqueList()
251 folder = await self._get_trending_folder()
252 if folder is None:
253 return UniqueList()
254 return folder.items
255
256 @use_cache(3600 * 3, base_class=RecommendationFolder)
257 async def _get_trending_folder(self) -> RecommendationFolder | None:
258 """Get the trending stations row with items, or None if unavailable."""
259 data = await self.__get_data("Browse.ashx", c="trending")
260 if not data or "body" not in data:
261 return None
262 all_stations = [
263 self._parse_radio_lazy(item)
264 for item in data["body"]
265 if item.get("type") == "audio" and "preset_id" in item
266 ]
267 return RecommendationFolder(
268 item_id="trending",
269 provider=self.instance_id,
270 name="Trending",
271 translation_key="trending_stations",
272 items=UniqueList(random.sample(all_stations, min(20, len(all_stations)))),
273 )
274
275 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
276 """Get stream details for a radio station."""
277 if item_id.startswith("http"):
278 # custom url
279 return StreamDetails(
280 provider=self.instance_id,
281 item_id=item_id,
282 audio_format=AudioFormat(
283 content_type=ContentType.UNKNOWN,
284 ),
285 media_type=MediaType.RADIO,
286 stream_type=StreamType.HTTP,
287 path=item_id,
288 allow_seek=False,
289 can_seek=False,
290 )
291 if "--" in item_id:
292 # handle this for backwards compatibility
293 item_id = item_id.split("--", maxsplit=1)[0]
294 if stream_info := await self._get_stream_info(item_id):
295 # assuming here that the streams are sorted by quality (bitrate)
296 # and the first one is the best quality
297 preferred_stream = stream_info[0]
298 return StreamDetails(
299 provider=self.instance_id,
300 item_id=item_id,
301 # set contenttype to unknown so ffmpeg can auto detect it
302 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
303 media_type=MediaType.RADIO,
304 stream_type=StreamType.HTTP,
305 path=preferred_stream["url"],
306 allow_seek=False,
307 can_seek=False,
308 )
309 msg = f"Unable to retrieve stream details for {item_id}"
310 raise MediaNotFoundError(msg)
311
312 @use_cache(3600)
313 async def search(
314 self, search_query: str, media_types: list[MediaType], limit: int = 10
315 ) -> SearchResults:
316 """Perform search on Tune-in music provider."""
317 result = SearchResults()
318 if MediaType.RADIO not in media_types:
319 return result
320 data = await self.__get_data("Search.ashx", query=search_query)
321 radios = []
322 if data and "body" in data:
323 count = 0
324 for item in data["body"]:
325 if item.get("type") == "audio" and "preset_id" in item:
326 radios.append(self._parse_radio_lazy(item))
327 count += 1
328 if count >= limit:
329 break
330 result.radio = radios
331 return result
332
333 @staticmethod
334 def _browse_path_key(text: str) -> str:
335 """Convert a display name to a URL-safe path segment."""
336 if " (" in text:
337 text = text[: text.rfind(" (")]
338 result = "".join(c if c.isalnum() else "_" for c in text.strip())
339 while "__" in result:
340 result = result.replace("__", "_")
341 return result.strip("_") or "item"
342
343 def _parse_radio_lazy(self, details: dict[str, Any]) -> Radio:
344 """Create a Radio from a browse item without fetching stream info."""
345 if "name" in details:
346 name = details["name"]
347 else:
348 name = details["text"]
349 if " | " in name:
350 name = name.split(" | ")[1]
351 name = name.split(" (")[0]
352 radio = Radio(
353 item_id=details["preset_id"],
354 provider=self.instance_id,
355 name=name,
356 provider_mappings={
357 ProviderMapping(
358 item_id=details["preset_id"],
359 provider_domain=self.domain,
360 provider_instance=self.instance_id,
361 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
362 details=details.get("URL", details["preset_id"]),
363 available=details.get("is_available", True),
364 )
365 },
366 )
367 if img := details.get("image") or details.get("logo"):
368 radio.metadata.images = UniqueList(
369 [
370 MediaItemImage(
371 type=ImageType.THUMB,
372 path=img.replace("http://", "https://", 1),
373 provider=self.instance_id,
374 remotely_accessible=True,
375 )
376 ]
377 )
378 return radio
379
380 def _parse_radio(
381 self,
382 details: dict[str, Any],
383 stream_info: list[dict[str, Any]] | None = None,
384 folder: str | None = None,
385 ) -> Radio:
386 """Parse Radio object from json obj returned from api."""
387 if "name" in details:
388 name = details["name"]
389 else:
390 # parse name from text attr
391 name = details["text"]
392 if " | " in name:
393 name = name.split(" | ")[1]
394 name = name.split(" (")[0]
395
396 if stream_info is not None:
397 # stream info is provided: parse first stream into provider mapping
398 # assuming here that the streams are sorted by quality (bitrate)
399 # and the first one is the best quality
400 preferred_stream = stream_info[0]
401 radio = Radio(
402 item_id=details["preset_id"],
403 provider=self.instance_id,
404 name=name,
405 provider_mappings={
406 ProviderMapping(
407 item_id=details["preset_id"],
408 provider_domain=self.domain,
409 provider_instance=self.instance_id,
410 audio_format=AudioFormat(
411 content_type=ContentType.try_parse(preferred_stream["media_type"]),
412 bit_rate=preferred_stream.get("bitrate", 128),
413 ),
414 details=preferred_stream["url"],
415 available=details.get("is_available", True),
416 )
417 },
418 )
419 else:
420 # custom url (no stream object present)
421 radio = Radio(
422 item_id=details["URL"],
423 provider=self.instance_id,
424 name=name,
425 provider_mappings={
426 ProviderMapping(
427 item_id=details["URL"],
428 provider_domain=self.domain,
429 provider_instance=self.instance_id,
430 audio_format=AudioFormat(
431 content_type=ContentType.UNKNOWN,
432 ),
433 details=details["URL"],
434 available=details.get("is_available", True),
435 )
436 },
437 )
438
439 # preset number is used for sorting (not present at stream time)
440 preset_number = details.get("preset_number", 0)
441 radio.position = preset_number
442 if "text" in details:
443 radio.metadata.description = details["text"]
444 # image
445 if img := details.get("image") or details.get("logo"):
446 radio.metadata.images = UniqueList(
447 [
448 MediaItemImage(
449 type=ImageType.THUMB,
450 path=img.replace("http://", "https://", 1),
451 provider=self.instance_id,
452 remotely_accessible=True,
453 )
454 ]
455 )
456 return radio
457
458 async def _get_stream_info(self, preset_id: str) -> list[dict[str, Any]]:
459 """Get stream info for a radio station."""
460 cached_data = await self.mass.cache.get(
461 preset_id, provider=self.instance_id, category=CACHE_CATEGORY_STREAMS
462 )
463 if cached_data is not None:
464 # We know from cache this is the right type
465 assert isinstance(cached_data, list)
466 return cached_data
467
468 data = await self.__get_data("Tune.ashx", id=preset_id)
469 if not data:
470 return []
471
472 body_data = data["body"]
473 assert isinstance(body_data, list)
474
475 await self.mass.cache.set(
476 key=preset_id,
477 data=body_data,
478 provider=self.instance_id,
479 category=CACHE_CATEGORY_STREAMS,
480 )
481 return body_data
482
483 async def __get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any] | None:
484 """Get data from api."""
485 if endpoint.startswith("http"):
486 url = endpoint
487 else:
488 url = f"https://opml.radiotime.com/{endpoint}"
489 kwargs["formats"] = "ogg,aac,wma,mp3,hls"
490 kwargs["username"] = self.get_setup_value(CONF_USERNAME)
491 kwargs["partnerId"] = "1"
492 kwargs["render"] = "json"
493 locale = self.mass.metadata.locale.replace("_", "-")
494 language = locale.split("-")[0]
495 headers = {"Accept-Language": f"{locale}, {language};q=0.9, *;q=0.5"}
496 async with (
497 self._throttler,
498 self.mass.http_session.get(url, params=kwargs, headers=headers, ssl=False) as response,
499 ):
500 result: Any = await response.json()
501 if not result or "error" in result:
502 self.logger.error(url)
503 self.logger.error(kwargs)
504 return None
505 assert isinstance(result, dict)
506 return result
507