/
/
/
1"""Helper functions for the music controller."""
2
3from __future__ import annotations
4
5from dataclasses import fields
6from typing import TYPE_CHECKING, Any, Final
7
8from music_assistant_models.helpers import create_safe_string
9from music_assistant_models.media_items import (
10 Artist,
11 ItemMapping,
12 MediaItemMetadata,
13 MediaItemType,
14 ProviderMapping,
15 SearchResults,
16)
17from music_assistant_models.unique_list import UniqueList
18
19if TYPE_CHECKING:
20 from collections.abc import Iterable, Sequence
21
22 from music_assistant_models.enums import MediaType
23
24# the trigram tokenizer of the FTS5 search index cannot match
25# search terms shorter than 3 characters
26MIN_FTS_TERM_LENGTH: Final[int] = 3
27
28
29def search_name_match_clause(
30 db_table: str, search_term: str, param_name: str, query_params: dict[str, Any]
31) -> str:
32 """
33 Return a SQL WHERE fragment matching ``search_term`` as substring of the search_name column.
34
35 :param db_table: The media item table to match against.
36 :param search_term: The (already normalized) search term to match.
37 :param param_name: Name of the query parameter to bind the search term to.
38 :param query_params: Query parameters dict the search term is bound into.
39 """
40 if len(search_term) < MIN_FTS_TERM_LENGTH:
41 # terms too short for the trigram index fall back to a LIKE scan
42 query_params[param_name] = f"%{search_term}%"
43 return f"{db_table}.search_name LIKE :{param_name}"
44 # quote the term so it is interpreted as a plain (sub)string instead of
45 # FTS5 query syntax; normalized search terms are alphanumeric only so
46 # they can never contain quotes themselves
47 query_params[param_name] = f'"{search_term}"'
48 return (
49 f"{db_table}.item_id IN "
50 f"(SELECT rowid FROM {db_table}_fts WHERE {db_table}_fts MATCH :{param_name})"
51 )
52
53
54def sort_search_result[SortItemT: MediaItemType | ItemMapping](
55 search_query: str,
56 items: Sequence[SortItemT],
57) -> UniqueList[SortItemT]:
58 """Sort search results on priority/preference."""
59 scored_items: list[tuple[int, SortItemT]] = []
60 # search results are already sorted by (streaming) providers on relevance
61 # but we prefer exact name matches and library items so we simply put those
62 # on top of the list.
63 safe_title_str = create_safe_string(search_query)
64 if " - " in search_query:
65 artist_name, title_alt = search_query.split(" - ", 1)
66 safe_title_alt = create_safe_string(title_alt)
67 safe_artist_str = create_safe_string(artist_name)
68 else:
69 safe_artist_str = None
70 safe_title_alt = None
71 for item in items:
72 score = 0
73 if create_safe_string(item.name) not in (safe_title_str, safe_title_alt):
74 # literal name match is mandatory to get a score at all
75 continue
76 # bonus point if artist provided and exact match
77 if safe_artist_str:
78 artist: Artist | ItemMapping
79 for artist in getattr(item, "artists", []):
80 if create_safe_string(artist.name) == safe_artist_str:
81 score += 1
82 # bonus point for library items
83 if item.provider == "library":
84 score += 1
85 scored_items.append((score, item))
86 scored_items.sort(key=lambda x: x[0], reverse=True)
87 # combine it all with uniquelist, so this will deduplicated by default
88 # note that streaming provider results are already (most likely) sorted on relevance
89 # so we add all remaining items in their original order. We just prioritize
90 # exact name matches and library items.
91 return UniqueList([*[x[1] for x in scored_items], *items])
92
93
94def filter_search_results(
95 results: SearchResults,
96 provider_domain: str,
97 skip_item_ids: set[tuple[MediaType, str, str]] | None,
98) -> SearchResults:
99 """
100 Return a copy of the given search results without the items in skip_item_ids.
101
102 :param results: The search results to filter.
103 :param provider_domain: Domain of the provider the results originate from.
104 :param skip_item_ids: Set of (media_type, provider_domain, item_id) tuples to filter out.
105 """
106 if not skip_item_ids:
107 return results
108
109 def _keep(item: MediaItemType | ItemMapping) -> bool:
110 return (item.media_type, provider_domain, item.item_id) not in skip_item_ids
111
112 # build a new SearchResults object as the original may be a (shared) cached object
113 return SearchResults(
114 artists=[x for x in results.artists if _keep(x)],
115 albums=[x for x in results.albums if _keep(x)],
116 genres=[x for x in results.genres if _keep(x)],
117 tracks=[x for x in results.tracks if _keep(x)],
118 playlists=[x for x in results.playlists if _keep(x)],
119 radio=[x for x in results.radio if _keep(x)],
120 audiobooks=[x for x in results.audiobooks if _keep(x)],
121 podcasts=[x for x in results.podcasts if _keep(x)],
122 sound_effects=[x for x in results.sound_effects if _keep(x)],
123 )
124
125
126def metadata_for_update(
127 stored: MediaItemMetadata, update: MediaItemMetadata, overwrite: bool
128) -> MediaItemMetadata:
129 """
130 Return the metadata to store for a library item update.
131
132 An overwrite replaces the stored metadata, unless the given item carries none at
133 all: providers embed a bare stub of an album or artist in their track payloads.
134
135 :param stored: Metadata currently stored for the library item.
136 :param update: Metadata of the item as delivered by the provider.
137 :param overwrite: Whether the given item replaces the stored one.
138 """
139 if overwrite and any(getattr(update, field.name) for field in fields(update)):
140 return update
141 return stored.update(update)
142
143
144def provider_mappings_for_update(
145 stored: Iterable[ProviderMapping], update: Iterable[ProviderMapping], overwrite: bool
146) -> set[ProviderMapping]:
147 """
148 Return the provider mappings to store for a library item update.
149
150 An overwrite replaces the mappings of the providers the given item comes from, so a
151 changed item id (a moved file) drops its stale row, and keeps the mappings written
152 by the other providers the item is linked to.
153
154 :param stored: Provider mappings currently stored for the library item.
155 :param update: Provider mappings of the item as delivered by the provider.
156 :param overwrite: Whether the given item replaces the stored one.
157 """
158 if not overwrite:
159 return {*update, *stored}
160 updated_instances = {mapping.provider_instance for mapping in update}
161 return {
162 *update,
163 *(mapping for mapping in stored if mapping.provider_instance not in updated_instances),
164 }
165