music-assistant-server
17.6 KB•PY
radio.py
17.6 KB • 406 lines • python
1"""Manage MediaItems of type Radio."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from dataclasses import replace
8from typing import TYPE_CHECKING, Any, cast
9
10from music_assistant_models.auth import Scope
11from music_assistant_models.enums import MediaType, ProviderFeature
12from music_assistant_models.errors import (
13 InvalidDataError,
14 MusicAssistantError,
15 ProviderUnavailableError,
16 UnsupportedFeaturedException,
17)
18from music_assistant_models.helpers import create_safe_string
19from music_assistant_models.media_items import ProviderMapping, Radio, RadioSummary, Track
20
21from music_assistant.constants import DB_TABLE_RADIOS
22from music_assistant.controllers.tasks.context import (
23 report_current_task_failure,
24 update_current_task_progress_from_index,
25)
26from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
27from music_assistant.helpers.compare import (
28 compare_media_item,
29 compare_radio,
30 loose_compare_strings,
31)
32from music_assistant.helpers.database import UNSET
33from music_assistant.helpers.json import serialize_to_json
34from music_assistant.helpers.playlists import (
35 PlaylistItem,
36 ProviderMappingInfo,
37 construct_media_item_from_playlist_item,
38 generate_m3u,
39 media_item_to_playlist_item,
40 parse_m3u,
41)
42from music_assistant.helpers.uri import parse_uri
43from music_assistant.models.music_provider import MusicProvider
44
45from .base import MediaControllerBase
46
47if TYPE_CHECKING:
48 from collections.abc import Mapping
49
50 from music_assistant_models.background_task import BackgroundTask
51
52 from music_assistant import MusicAssistant
53
54
55class RadioController(MediaControllerBase[Radio]):
56 """Controller managing MediaItems of type Radio."""
57
58 db_table = DB_TABLE_RADIOS
59 media_type = MediaType.RADIO
60 item_cls = Radio
61 summary_item_cls = RadioSummary
62
63 def __init__(self, mass: MusicAssistant) -> None:
64 """Initialize class."""
65 super().__init__(mass)
66 # register (extra) api handlers
67 api_base = self.api_base
68 self.mass.register_api_command(
69 f"music/{api_base}/radio_versions", self.versions, required_scope=Scope.LIBRARY_READ
70 )
71 self.mass.register_api_command(
72 f"music/{api_base}/radio_tracks",
73 self.radio_tracks,
74 required_scope=Scope.LIBRARY_READ,
75 )
76 self.mass.register_api_command(
77 f"music/{api_base}/export_radios", self.export_radios, required_scope=Scope.LIBRARY_READ
78 )
79 self.mass.register_api_command(
80 f"music/{api_base}/import_radios",
81 self.import_radios,
82 required_scope=Scope.LIBRARY_WRITE,
83 )
84
85 @property
86 def summary_query(self) -> tuple[str, dict[str, Any]]:
87 """Return the slim SELECT query used for radio summary listings."""
88 query = f"""
89 SELECT
90 {self._summary_base_columns()},
91 {self.db_table}.is_dynamic,
92 json_extract({self.db_table}.metadata, '$.description') AS description,
93 {self._provider_mappings_query()} AS provider_mappings
94 FROM {self.db_table}"""
95 return query, {}
96
97 async def radio_tracks(self, item_id: str, provider_instance_id_or_domain: str) -> list[Track]:
98 """
99 Return a fresh batch of tracks for a dynamic radio station.
100
101 :param item_id: The provider (or library) item id of the station.
102 :param provider_instance_id_or_domain: The provider instance id or domain the
103 item id belongs to ("library" for a library item).
104 """
105 radio = await self.get_provider_item(item_id, provider_instance_id_or_domain)
106 return await self.dynamic_tracks(radio)
107
108 async def dynamic_tracks(self, radio: Radio) -> list[Track]:
109 """
110 Return a fresh batch of tracks for an already resolved dynamic radio station.
111
112 :param radio: The dynamic station to fetch the next batch for.
113 """
114 if not radio.is_dynamic:
115 raise UnsupportedFeaturedException(f"{radio.name} is not a dynamic radio station")
116 provider_instance_id_or_domain, item_id = (
117 self._select_provider_id(radio)
118 if radio.provider == "library"
119 else (radio.provider, radio.item_id)
120 )
121 if not (provider := self.mass.get_provider(provider_instance_id_or_domain)):
122 raise ProviderUnavailableError(f"{provider_instance_id_or_domain} is not available")
123 return await cast("MusicProvider", provider).get_dynamic_radio_tracks(item_id)
124
125 async def export_radios(self) -> str:
126 """Export all library radio stations to M3U8 format."""
127 items: list[PlaylistItem] = []
128 async for radio in self.iter_library_items():
129 entry = media_item_to_playlist_item(radio)
130 if radio.favorite:
131 # favorite is library-level user state, so only a library export carries it
132 entry.metadata = {**(entry.metadata or {}), "favorite": "true"}
133 items.append(entry)
134 return generate_m3u("Radio Stations", items)
135
136 async def import_radios(self, m3u_data: str) -> BackgroundTask:
137 """
138 Queue importing radio stations from M3U8 format.
139
140 Any station that can not be imported is reported as a failure on the returned
141 task, and the remaining stations are still imported.
142
143 :param m3u_data: The M3U8 data as a string.
144 :return: Managed background task performing the import.
145 :raises InvalidDataError: The M3U data holds no entries.
146 """
147 parsed_items = parse_m3u(m3u_data)
148 if not parsed_items:
149 msg = "No items found in M3U data"
150 raise InvalidDataError(msg)
151 user = get_current_user()
152 return self.mass.tasks.run_background_task(
153 name=f"Import {len(parsed_items)} radio stations",
154 handler=lambda: self._handle_import_radios(parsed_items),
155 translation_key="import_radios",
156 translation_owner=self.translation_owner,
157 translation_args=[len(parsed_items)],
158 user_id=user.user_id if user else None,
159 metadata={
160 "task_domain": "radio_import",
161 "item_count": len(parsed_items),
162 },
163 allow_retry=True,
164 priority=True,
165 )
166
167 async def versions(
168 self,
169 item_id: str,
170 provider_instance_id_or_domain: str,
171 ) -> list[Radio]:
172 """Return all versions of a radio station we can find on all providers."""
173 radio = await self.get(item_id, provider_instance_id_or_domain)
174 if radio.is_dynamic:
175 # a dynamic station is its provider's own, so a same-named station is a different one
176 return []
177 # perform a search on all provider(types) to collect all versions/variants
178 all_versions = {
179 prov_item.item_id: prov_item
180 for prov_items in await asyncio.gather(
181 *[
182 self.search(radio.name, provider_domain)
183 for provider_domain in self.mass.music.get_unique_providers()
184 ]
185 )
186 for prov_item in prov_items
187 if loose_compare_strings(radio.name, prov_item.name)
188 }
189 # make sure that the 'base' version is NOT included
190 for prov_version in radio.provider_mappings:
191 all_versions.pop(prov_version.item_id, None)
192
193 # return the aggregated result
194 return list(all_versions.values())
195
196 async def match_provider(
197 self, db_radio: Radio, provider: MusicProvider, strict: bool = True
198 ) -> list[ProviderMapping]:
199 """
200 Try to find match on (streaming) provider for the provided (database) radio.
201
202 This is used to link objects of different providers/qualities together.
203 """
204 self.logger.debug(
205 "Trying to match radio %s on provider %s",
206 db_radio.name,
207 provider.name,
208 )
209 matches: list[ProviderMapping] = []
210 search_str = db_radio.name
211 search_result = await self.search(search_str, provider.instance_id)
212 for search_result_item in search_result:
213 if not search_result_item.available:
214 continue
215 if not compare_media_item(db_radio, search_result_item, strict=strict):
216 continue
217 # we must fetch the full radio version, search results can be simplified objects
218 prov_radio = await self.get_provider_item(
219 search_result_item.item_id,
220 search_result_item.provider,
221 fallback=search_result_item,
222 )
223 if compare_radio(db_radio, prov_radio, strict=strict):
224 # 100% match
225 matches.extend(prov_radio.provider_mappings)
226 if not matches:
227 self.logger.debug(
228 "Could not find match for Radio %s on provider %s",
229 db_radio.name,
230 provider.name,
231 )
232 return matches
233
234 async def match_providers(self, db_radio: Radio) -> None:
235 """
236 Try to find match on all (streaming) providers for the provided (database) radio.
237
238 This is used to link objects of different providers/qualities together.
239 """
240 if db_radio.provider != "library":
241 return # Matching only supported for database items
242 if db_radio.is_dynamic:
243 # matching a dynamic station by name would link an unrelated radio stream to it
244 return
245
246 # try to find match on all providers
247 cur_provider_domains = {x.provider_domain for x in db_radio.provider_mappings}
248 for provider in self.mass.music.providers:
249 if provider.domain in cur_provider_domains:
250 continue
251 if ProviderFeature.SEARCH not in provider.supported_features:
252 continue
253 if MediaType.RADIO not in provider.supported_media_types:
254 continue
255 if not provider.is_streaming_provider:
256 # matching on unique providers is pointless as they push (all) their content to MA
257 continue
258 if match := await self.match_provider(db_radio, provider):
259 # 100% match, we update the db with the additional provider mapping(s)
260 await self.add_provider_mappings(db_radio.item_id, match)
261 cur_provider_domains.add(provider.domain)
262
263 async def _add_library_item(self, item: Radio, overwrite_existing: bool = False) -> int:
264 """Add a new item record to the database."""
265 assert self.mass.music.database is not None # For type checking
266 db_id = await self.mass.music.database.insert(
267 self.db_table,
268 {
269 "name": item.name,
270 "sort_name": item.sort_name,
271 "favorite": item.favorite,
272 "metadata": serialize_to_json(item.metadata),
273 "search_name": create_safe_string(item.name, True, True),
274 "search_sort_name": create_safe_string(
275 item.sort_name if item.sort_name is not None else "", True, True
276 ),
277 "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
278 "is_dynamic": item.is_dynamic,
279 },
280 )
281 # update/set external id lookup table
282 await self.set_external_ids(db_id, item.external_ids)
283 # update/set provider_mappings table
284 await self.set_provider_mappings(db_id, item.provider_mappings)
285 self.logger.debug("added %s to database (id: %s)", item.name, db_id)
286 return db_id
287
288 async def _update_library_item(
289 self, item_id: str | int, update: Radio, overwrite: bool = False
290 ) -> None:
291 """Update existing record in the database."""
292 db_id = int(item_id) # ensure integer
293 cur_item = await self.get_library_item(db_id)
294 metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
295 cur_item.external_ids.update(update.external_ids)
296 match = {"item_id": db_id}
297 name = update.name if overwrite else cur_item.name
298 sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
299 assert self.mass.music.database is not None # For type checking
300 await self.mass.music.database.update(
301 self.db_table,
302 match,
303 {
304 # always prefer name from updated item here
305 "name": name,
306 "sort_name": sort_name,
307 "metadata": serialize_to_json(metadata),
308 "search_name": create_safe_string(name, True, True),
309 "search_sort_name": create_safe_string(sort_name or "", True, True),
310 "timestamp_added": int(update.date_added.timestamp())
311 if update.date_added
312 else UNSET,
313 "is_dynamic": update.is_dynamic,
314 },
315 )
316 # update/set external id lookup table
317 await self.set_external_ids(
318 db_id, update.external_ids if overwrite else cur_item.external_ids
319 )
320 # update/set provider_mappings table
321 provider_mappings = (
322 update.provider_mappings
323 if overwrite
324 else {*update.provider_mappings, *cur_item.provider_mappings}
325 )
326 await self.set_provider_mappings(db_id, provider_mappings, overwrite)
327 self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
328
329 def _parse_summary_row(self, db_row: Mapping[str, Any]) -> RadioSummary:
330 """Parse a raw summary db row into a RadioSummary object."""
331 item = cast("RadioSummary", super()._parse_summary_row(db_row))
332 item.is_dynamic = bool(db_row["is_dynamic"])
333 item.metadata.description = db_row["description"]
334 return item
335
336 async def _handle_import_radios(self, parsed_items: list[PlaylistItem]) -> None:
337 """Add the parsed M3U entries to the library, one station at a time."""
338 total = len(parsed_items)
339 for index, item in enumerate(parsed_items):
340 update_current_task_progress_from_index(
341 index, total, f"Importing station {index + 1}/{total}"
342 )
343 label = (item.metadata or {}).get("name") or item.title or item.path
344 try:
345 await self._import_radio_item(item)
346 except MusicAssistantError as err:
347 report_current_task_failure(f"{label}: {err}")
348 except Exception as err:
349 # a transient fault such as a provider timeout must not abandon the
350 # stations queued behind it, but it is a real fault worth a log line
351 self.logger.warning(
352 "Error importing radio station %s: %s",
353 label,
354 str(err),
355 exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
356 )
357 report_current_task_failure(f"{label}: {err}")
358 update_current_task_progress_from_index(total, total, "Import complete")
359
360 async def _import_radio_item(self, item: PlaylistItem) -> None:
361 """Resolve a single parsed M3U entry against its provider and add it to the library."""
362 if not item.providers:
363 # a plain third-party M3U carries no #EXTPROV, so recover the mapping from the
364 # path, which parse_uri normalises into a provider and its native item_id
365 if not item.is_url:
366 msg = f"{item.path} is not a stream URL"
367 raise InvalidDataError(msg)
368 media_type, prov_lookup, prov_item_id = await parse_uri(item.path)
369 if media_type not in (MediaType.RADIO, MediaType.UNKNOWN):
370 msg = f"{item.path} is a {media_type.value}, not a radio station"
371 raise InvalidDataError(msg)
372 provider = self.mass.get_provider(prov_lookup, provider_type=MusicProvider)
373 if not provider:
374 msg = f"Provider {prov_lookup} is not available"
375 raise ProviderUnavailableError(msg)
376 # a retry re-runs this handler over the same parsed items, so build a new
377 # entry rather than writing the resolved mapping back into the shared one
378 item = replace(
379 item,
380 providers=[
381 ProviderMappingInfo(
382 domain=provider.domain,
383 item_id=prov_item_id,
384 instance_id=provider.instance_id,
385 )
386 ],
387 )
388 radio = construct_media_item_from_playlist_item(item, self.mass, MediaType.RADIO)
389 if not isinstance(radio, Radio):
390 msg = f"{item.path} is not a radio station"
391 raise InvalidDataError(msg)
392 if not any(mapping.available for mapping in radio.provider_mappings):
393 msg = f"No available provider for radio station {radio.name}"
394 raise ProviderUnavailableError(msg)
395 library_item = await self.mass.music.add_item_to_library(radio)
396 # a station owned by another provider is refetched from it, so the library item comes
397 # back with only that provider's mapping; reattach the ones the file recorded for the
398 # other providers it is loaded on, which the refetch has no way to know about
399 await self.add_provider_mappings(
400 library_item.item_id, [pm for pm in radio.provider_mappings if pm.available]
401 )
402 # the refetch also discards anything set on the object passed in, so the exported
403 # favorite goes onto the library item
404 if (item.metadata or {}).get("favorite") == "true":
405 await self.set_favorite(library_item.item_id, True)
406