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