music-assistant-server

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