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