/
/
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
178 async def parse_items(
179 items: list[dict[str, Any]], folder: str | None = None
180 ) -> AsyncGenerator[Radio]:
181 for item in items:
182 item_type = item.get("type", "")
183 if "unavailable" in item.get("key", ""):
184 continue
185 if not item.get("is_available", True):
186 continue
187 if item_type == "audio":
188 if "preset_id" not in item:
189 continue
190 # each radio station can have multiple streams add each one as different quality
191 stream_info = await self._get_stream_info(item["preset_id"])
192 yield self._parse_radio(item, stream_info, folder)
193 elif item_type == "link" and item.get("item") == "url":
194 # custom url
195 try:
196 yield self._parse_radio(item)
197 except InvalidDataError as err:
198 # there may be invalid custom urls, ignore those
199 self.report_skipped_sync_item(MediaType.RADIO, item["URL"], err)
200 elif item_type == "link":
201 # stations are in sublevel (new style)
202 if sublevel := await self.__get_data(item["URL"], render="json"):
203 async for subitem in parse_items(sublevel["body"], item["text"]):
204 yield subitem
205 elif item.get("children"):
206 # stations are in sublevel (old style ?)
207 async for subitem in parse_items(item["children"], item["text"]):
208 yield subitem
209
210 data = await self.__get_data("Browse.ashx", c="presets")
211 if data and "body" in data:
212 async for item in parse_items(data["body"]):
213 yield item
214
215 @use_cache(3600 * 24 * 30) # Cache for 30 days
216 async def get_radio(self, prov_radio_id: str) -> Radio:
217 """Get radio station details."""
218 if not prov_radio_id.startswith("http"):
219 if "--" in prov_radio_id:
220 # handle this for backwards compatibility
221 prov_radio_id = prov_radio_id.split("--", maxsplit=1)[0]
222 params = {"c": "composite", "detail": "listing", "id": prov_radio_id}
223 result = await self.__get_data("Describe.ashx", **params)
224 if result and result.get("body") and result["body"][0].get("children"):
225 item = result["body"][0]["children"][0]
226 stream_info = await self._get_stream_info(prov_radio_id)
227 return self._parse_radio(item, stream_info)
228 # fallback - e.g. for handle custom urls ...
229 async for radio in self.get_library_radios():
230 if radio.item_id == prov_radio_id:
231 return radio
232 msg = f"Item {prov_radio_id} not found"
233 raise MediaNotFoundError(msg)
234
235 async def get_recommendations(self) -> list[RecommendationFolder]:
236 """Get this provider's available recommendation rows, without items."""
237 return [
238 RecommendationFolder(
239 item_id="trending",
240 provider=self.instance_id,
241 name="Trending",
242 translation_key="trending_stations",
243 )
244 ]
245
246 async def get_recommendation_items(
247 self, item_id: str
248 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
249 """
250 Get the items for a single recommendation row.
251
252 :param item_id: The item_id of the row, as returned by get_recommendations.
253 """
254 if item_id != "trending":
255 return UniqueList()
256 folder = await self._get_trending_folder()
257 if folder is None:
258 return UniqueList()
259 return folder.items
260
261 @use_cache(3600 * 3, base_class=RecommendationFolder)
262 async def _get_trending_folder(self) -> RecommendationFolder | None:
263 """Get the trending stations row with items, or None if unavailable."""
264 data = await self.__get_data("Browse.ashx", c="trending")
265 if not data or "body" not in data:
266 return None
267 all_stations = [
268 self._parse_radio_lazy(item)
269 for item in data["body"]
270 if item.get("type") == "audio" and "preset_id" in item
271 ]
272 return RecommendationFolder(
273 item_id="trending",
274 provider=self.instance_id,
275 name="Trending",
276 translation_key="trending_stations",
277 items=UniqueList(random.sample(all_stations, min(20, len(all_stations)))),
278 )
279
280 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
281 """Get stream details for a radio station."""
282 if item_id.startswith("http"):
283 # custom url
284 return StreamDetails(
285 provider=self.instance_id,
286 item_id=item_id,
287 audio_format=AudioFormat(
288 content_type=ContentType.UNKNOWN,
289 ),
290 media_type=MediaType.RADIO,
291 stream_type=StreamType.HTTP,
292 path=item_id,
293 allow_seek=False,
294 can_seek=False,
295 )
296 if "--" in item_id:
297 # handle this for backwards compatibility
298 item_id = item_id.split("--", maxsplit=1)[0]
299 if stream_info := await self._get_stream_info(item_id):
300 # assuming here that the streams are sorted by quality (bitrate)
301 # and the first one is the best quality
302 preferred_stream = stream_info[0]
303 return StreamDetails(
304 provider=self.instance_id,
305 item_id=item_id,
306 # set contenttype to unknown so ffmpeg can auto detect it
307 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
308 media_type=MediaType.RADIO,
309 stream_type=StreamType.HTTP,
310 path=preferred_stream["url"],
311 allow_seek=False,
312 can_seek=False,
313 )
314 msg = f"Unable to retrieve stream details for {item_id}"
315 raise MediaNotFoundError(msg)
316
317 @use_cache(3600)
318 async def search(
319 self, search_query: str, media_types: list[MediaType], limit: int = 10
320 ) -> SearchResults:
321 """Perform search on Tune-in music provider."""
322 result = SearchResults()
323 if MediaType.RADIO not in media_types:
324 return result
325 data = await self.__get_data("Search.ashx", query=search_query)
326 radios = []
327 if data and "body" in data:
328 count = 0
329 for item in data["body"]:
330 if item.get("type") == "audio" and "preset_id" in item:
331 radios.append(self._parse_radio_lazy(item))
332 count += 1
333 if count >= limit:
334 break
335 result.radio = radios
336 return result
337
338 @staticmethod
339 def _browse_path_key(text: str) -> str:
340 """Convert a display name to a URL-safe path segment."""
341 if " (" in text:
342 text = text[: text.rfind(" (")]
343 result = "".join(c if c.isalnum() else "_" for c in text.strip())
344 while "__" in result:
345 result = result.replace("__", "_")
346 return result.strip("_") or "item"
347
348 def _parse_radio_lazy(self, details: dict[str, Any]) -> Radio:
349 """Create a Radio from a browse item without fetching stream info."""
350 if "name" in details:
351 name = details["name"]
352 else:
353 name = details["text"]
354 if " | " in name:
355 name = name.split(" | ")[1]
356 name = name.split(" (")[0]
357 radio = Radio(
358 item_id=details["preset_id"],
359 provider=self.instance_id,
360 name=name,
361 provider_mappings={
362 ProviderMapping(
363 item_id=details["preset_id"],
364 provider_domain=self.domain,
365 provider_instance=self.instance_id,
366 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
367 details=details.get("URL", details["preset_id"]),
368 available=details.get("is_available", True),
369 )
370 },
371 )
372 if img := details.get("image") or details.get("logo"):
373 radio.metadata.images = UniqueList(
374 [
375 MediaItemImage(
376 type=ImageType.THUMB,
377 path=img.replace("http://", "https://", 1),
378 provider=self.instance_id,
379 remotely_accessible=True,
380 )
381 ]
382 )
383 return radio
384
385 def _parse_radio(
386 self,
387 details: dict[str, Any],
388 stream_info: list[dict[str, Any]] | None = None,
389 folder: str | None = None,
390 ) -> Radio:
391 """Parse Radio object from json obj returned from api."""
392 if "name" in details:
393 name = details["name"]
394 else:
395 # parse name from text attr
396 name = details["text"]
397 if " | " in name:
398 name = name.split(" | ")[1]
399 name = name.split(" (")[0]
400
401 if stream_info is not None:
402 # stream info is provided: parse first stream into provider mapping
403 # assuming here that the streams are sorted by quality (bitrate)
404 # and the first one is the best quality
405 preferred_stream = stream_info[0]
406 radio = Radio(
407 item_id=details["preset_id"],
408 provider=self.instance_id,
409 name=name,
410 provider_mappings={
411 ProviderMapping(
412 item_id=details["preset_id"],
413 provider_domain=self.domain,
414 provider_instance=self.instance_id,
415 audio_format=AudioFormat(
416 content_type=ContentType.try_parse(preferred_stream["media_type"]),
417 bit_rate=preferred_stream.get("bitrate", 128),
418 ),
419 details=preferred_stream["url"],
420 available=details.get("is_available", True),
421 )
422 },
423 )
424 else:
425 # custom url (no stream object present)
426 radio = Radio(
427 item_id=details["URL"],
428 provider=self.instance_id,
429 name=name,
430 provider_mappings={
431 ProviderMapping(
432 item_id=details["URL"],
433 provider_domain=self.domain,
434 provider_instance=self.instance_id,
435 audio_format=AudioFormat(
436 content_type=ContentType.UNKNOWN,
437 ),
438 details=details["URL"],
439 available=details.get("is_available", True),
440 )
441 },
442 )
443
444 # preset number is used for sorting (not present at stream time)
445 preset_number = details.get("preset_number", 0)
446 radio.position = preset_number
447 if "text" in details:
448 radio.metadata.description = details["text"]
449 # image
450 if img := details.get("image") or details.get("logo"):
451 radio.metadata.images = UniqueList(
452 [
453 MediaItemImage(
454 type=ImageType.THUMB,
455 path=img.replace("http://", "https://", 1),
456 provider=self.instance_id,
457 remotely_accessible=True,
458 )
459 ]
460 )
461 return radio
462
463 async def _get_stream_info(self, preset_id: str) -> list[dict[str, Any]]:
464 """Get stream info for a radio station."""
465 cached_data = await self.mass.cache.get(
466 preset_id, provider=self.instance_id, category=CACHE_CATEGORY_STREAMS
467 )
468 if cached_data is not None:
469 # We know from cache this is the right type
470 assert isinstance(cached_data, list)
471 return cached_data
472
473 data = await self.__get_data("Tune.ashx", id=preset_id)
474 if not data:
475 return []
476
477 body_data = data["body"]
478 assert isinstance(body_data, list)
479
480 await self.mass.cache.set(
481 key=preset_id,
482 data=body_data,
483 provider=self.instance_id,
484 category=CACHE_CATEGORY_STREAMS,
485 )
486 return body_data
487
488 async def __get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any] | None:
489 """Get data from api."""
490 if endpoint.startswith("http"):
491 url = endpoint
492 else:
493 url = f"https://opml.radiotime.com/{endpoint}"
494 kwargs["formats"] = "ogg,aac,wma,mp3,hls"
495 kwargs["username"] = self.get_setup_value(CONF_USERNAME)
496 kwargs["partnerId"] = "1"
497 kwargs["render"] = "json"
498 locale = self.mass.metadata.locale.replace("_", "-")
499 language = locale.split("-")[0]
500 headers = {"Accept-Language": f"{locale}, {language};q=0.9, *;q=0.5"}
501 async with (
502 self._throttler,
503 self.mass.http_session.get(url, params=kwargs, headers=headers, ssl=False) as response,
504 ):
505 result: Any = await response.json()
506 if not result or "error" in result:
507 self.logger.error(url)
508 self.logger.error(kwargs)
509 return None
510 assert isinstance(result, dict)
511 return result
512