/
/
/
1"""Helper functions for the music controller."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any, Final
6
7from music_assistant_models.media_items import Artist, ItemMapping, MediaItemType, SearchResults
8from music_assistant_models.unique_list import UniqueList
9
10from music_assistant.helpers.compare import create_safe_string
11
12if TYPE_CHECKING:
13 from collections.abc import Sequence
14
15 from music_assistant_models.enums import MediaType
16
17# the trigram tokenizer of the FTS5 search index cannot match
18# search terms shorter than 3 characters
19MIN_FTS_TERM_LENGTH: Final[int] = 3
20
21
22def search_name_match_clause(
23 db_table: str, search_term: str, param_name: str, query_params: dict[str, Any]
24) -> str:
25 """
26 Return a SQL WHERE fragment matching ``search_term`` as substring of the search_name column.
27
28 :param db_table: The media item table to match against.
29 :param search_term: The (already normalized) search term to match.
30 :param param_name: Name of the query parameter to bind the search term to.
31 :param query_params: Query parameters dict the search term is bound into.
32 """
33 if len(search_term) < MIN_FTS_TERM_LENGTH:
34 # terms too short for the trigram index fall back to a LIKE scan
35 query_params[param_name] = f"%{search_term}%"
36 return f"{db_table}.search_name LIKE :{param_name}"
37 # quote the term so it is interpreted as a plain (sub)string instead of
38 # FTS5 query syntax; normalized search terms are alphanumeric only so
39 # they can never contain quotes themselves
40 query_params[param_name] = f'"{search_term}"'
41 return (
42 f"{db_table}.item_id IN "
43 f"(SELECT rowid FROM {db_table}_fts WHERE {db_table}_fts MATCH :{param_name})"
44 )
45
46
47def sort_search_result[SortItemT: MediaItemType | ItemMapping](
48 search_query: str,
49 items: Sequence[SortItemT],
50) -> UniqueList[SortItemT]:
51 """Sort search results on priority/preference."""
52 scored_items: list[tuple[int, SortItemT]] = []
53 # search results are already sorted by (streaming) providers on relevance
54 # but we prefer exact name matches and library items so we simply put those
55 # on top of the list.
56 safe_title_str = create_safe_string(search_query)
57 if " - " in search_query:
58 artist_name, title_alt = search_query.split(" - ", 1)
59 safe_title_alt = create_safe_string(title_alt)
60 safe_artist_str = create_safe_string(artist_name)
61 else:
62 safe_artist_str = None
63 safe_title_alt = None
64 for item in items:
65 score = 0
66 if create_safe_string(item.name) not in (safe_title_str, safe_title_alt):
67 # literal name match is mandatory to get a score at all
68 continue
69 # bonus point if artist provided and exact match
70 if safe_artist_str:
71 artist: Artist | ItemMapping
72 for artist in getattr(item, "artists", []):
73 if create_safe_string(artist.name) == safe_artist_str:
74 score += 1
75 # bonus point for library items
76 if item.provider == "library":
77 score += 1
78 scored_items.append((score, item))
79 scored_items.sort(key=lambda x: x[0], reverse=True)
80 # combine it all with uniquelist, so this will deduplicated by default
81 # note that streaming provider results are already (most likely) sorted on relevance
82 # so we add all remaining items in their original order. We just prioritize
83 # exact name matches and library items.
84 return UniqueList([*[x[1] for x in scored_items], *items])
85
86
87def filter_search_results(
88 results: SearchResults,
89 provider_domain: str,
90 skip_item_ids: set[tuple[MediaType, str, str]] | None,
91) -> SearchResults:
92 """
93 Return a copy of the given search results without the items in skip_item_ids.
94
95 :param results: The search results to filter.
96 :param provider_domain: Domain of the provider the results originate from.
97 :param skip_item_ids: Set of (media_type, provider_domain, item_id) tuples to filter out.
98 """
99 if not skip_item_ids:
100 return results
101
102 def _keep(item: MediaItemType | ItemMapping) -> bool:
103 return (item.media_type, provider_domain, item.item_id) not in skip_item_ids
104
105 # build a new SearchResults object as the original may be a (shared) cached object
106 return SearchResults(
107 artists=[x for x in results.artists if _keep(x)],
108 albums=[x for x in results.albums if _keep(x)],
109 genres=[x for x in results.genres if _keep(x)],
110 tracks=[x for x in results.tracks if _keep(x)],
111 playlists=[x for x in results.playlists if _keep(x)],
112 radio=[x for x in results.radio if _keep(x)],
113 audiobooks=[x for x in results.audiobooks if _keep(x)],
114 podcasts=[x for x in results.podcasts if _keep(x)],
115 sound_effects=[x for x in results.sound_effects if _keep(x)],
116 )
117