/
/
/
1"""Suggestion helpers for the Music Quiz provider."""
2
3from __future__ import annotations
4
5import random
6import re
7import secrets
8import unicodedata
9from collections.abc import Iterable, Sequence
10from dataclasses import dataclass
11from difflib import SequenceMatcher
12
13from music_assistant.providers.music_quiz.models import MultipleChoiceSuggestion
14
15# collapse runs of non-word characters and underscores so filename-style titles
16# ("Foo_Bar") normalize like their spaced form ("Foo Bar"); \W keeps this
17# Unicode-aware so non-Latin titles (e.g. CJK, Cyrillic) survive instead of becoming ""
18NORMALIZE_PATTERN = re.compile(r"[\W_]+")
19MAX_LABEL_SIMILARITY = 0.78
20MAX_TOKEN_CONTAINMENT = 0.85
21SYSTEM_RANDOM = secrets.SystemRandom()
22
23
24@dataclass(frozen=True)
25class SuggestionCandidate:
26 """A candidate track answer for Music Quiz suggestions."""
27
28 label: str
29 uri: str | None = None
30 title: str | None = None
31 artist_names: tuple[str, ...] = ()
32
33
34@dataclass(frozen=True)
35class OpaqueOption:
36 """An answer option with an opaque client-visible identity."""
37
38 option_id: str
39 label: str
40 uri: str | None
41 is_correct: bool
42
43
44def normalize_answer_label(label: str) -> str:
45 """
46 Normalize an answer label for duplicate detection.
47
48 :param label: Answer label to normalize.
49 """
50 normalized_label = unicodedata.normalize("NFKC", label).casefold()
51 return NORMALIZE_PATTERN.sub(" ", normalized_label).strip()
52
53
54def answer_labels_are_too_close(first_label: str, second_label: str) -> bool:
55 """
56 Return if two answer labels are too similar to use together.
57
58 :param first_label: First answer label to compare.
59 :param second_label: Second answer label to compare.
60 """
61 first = normalize_answer_label(first_label)
62 second = normalize_answer_label(second_label)
63 if not first or not second:
64 return False
65 if first == second:
66 return True
67
68 similarity = SequenceMatcher(None, first, second).ratio()
69 if similarity >= MAX_LABEL_SIMILARITY:
70 return True
71
72 first_tokens = set(first.split())
73 second_tokens = set(second.split())
74 shared_tokens = first_tokens & second_tokens
75 token_containment = len(shared_tokens) / min(len(first_tokens), len(second_tokens))
76 return token_containment >= MAX_TOKEN_CONTAINMENT
77
78
79def suggestion_candidates_are_too_close(
80 first: SuggestionCandidate,
81 second: SuggestionCandidate,
82) -> bool:
83 """
84 Return if two candidates are too similar to use together.
85
86 Prefer comparing raw track titles when available so artists with similar
87 names do not dominate the distance check.
88 """
89 if first.title and second.title:
90 return answer_labels_are_too_close(first.title, second.title)
91 return answer_labels_are_too_close(first.label, second.label)
92
93
94def build_answer_label(artist: str | None, title: str) -> str:
95 """
96 Build the displayed answer label.
97
98 :param artist: Artist name.
99 :param title: Track title.
100 """
101 if artist:
102 return f"{artist} - {title}"
103 return title
104
105
106def build_suggestions(
107 correct: SuggestionCandidate,
108 distractors: Iterable[SuggestionCandidate],
109 suggestion_count: int,
110 *,
111 rng: random.Random | None = None,
112) -> list[MultipleChoiceSuggestion]:
113 """
114 Build shuffled suggestions containing exactly one correct answer.
115
116 :param correct: Correct answer candidate.
117 :param distractors: Wrong answer candidates.
118 :param suggestion_count: Total number of suggestions to return.
119 :param rng: Optional random generator.
120 """
121 return [
122 MultipleChoiceSuggestion(
123 suggestion_id=option.option_id,
124 label=option.label,
125 uri=option.uri,
126 is_correct=option.is_correct,
127 )
128 for option in build_opaque_options(
129 correct,
130 distractors,
131 suggestion_count,
132 rng=rng,
133 )
134 ]
135
136
137def build_opaque_options(
138 correct: SuggestionCandidate,
139 distractors: Iterable[SuggestionCandidate],
140 option_count: int,
141 *,
142 rng: random.Random | None = None,
143) -> list[OpaqueOption]:
144 """
145 Build shuffled answer options with opaque IDs and one correct answer.
146
147 :param correct: Correct answer candidate.
148 :param distractors: Wrong answer candidates.
149 :param option_count: Total number of options to return.
150 :param rng: Optional random generator.
151 """
152 if option_count < 2:
153 msg = "Suggestion count must be at least 2"
154 raise ValueError(msg)
155
156 selected = _select_distractors(correct, distractors, option_count - 1)
157 # option IDs are sent to guests while the answer is still secret: they
158 # must be opaque, never semantic ("correct"/"wrong_x" would leak the answer)
159 options = [
160 OpaqueOption(
161 option_id=secrets.token_hex(8),
162 label=correct.label,
163 uri=correct.uri,
164 is_correct=True,
165 ),
166 *[
167 OpaqueOption(
168 option_id=secrets.token_hex(8),
169 label=candidate.label,
170 uri=candidate.uri,
171 is_correct=False,
172 )
173 for candidate in selected
174 ],
175 ]
176 (rng or SYSTEM_RANDOM).shuffle(options)
177 return options
178
179
180def has_enough_distractors(
181 correct: SuggestionCandidate,
182 distractors: Iterable[SuggestionCandidate],
183 option_count: int,
184) -> bool:
185 """
186 Return whether candidates can fill every wrong option.
187
188 :param correct: Correct answer candidate.
189 :param distractors: Wrong answer candidates.
190 :param option_count: Total number of options that must be built.
191 """
192 if option_count < 2:
193 return False
194 try:
195 _select_distractors(correct, distractors, option_count - 1)
196 except ValueError:
197 return False
198 return True
199
200
201def filter_suggestion_candidates(
202 correct: SuggestionCandidate,
203 distractors: Iterable[SuggestionCandidate],
204 *,
205 limit: int | None = None,
206) -> list[SuggestionCandidate]:
207 """
208 Return ordered distractors that stay distinct from the answer and each other.
209
210 :param correct: Correct answer candidate.
211 :param distractors: Ordered wrong-answer candidates to filter.
212 :param limit: Maximum number of candidates to return.
213 """
214 if limit is not None and limit < 1:
215 return []
216 correct_label = normalize_answer_label(correct.label)
217 seen_labels = {correct_label}
218 seen_uris = {correct.uri} if correct.uri else set()
219 selected: list[SuggestionCandidate] = []
220 for candidate in distractors:
221 candidate_label = normalize_answer_label(candidate.label)
222 if not candidate_label or candidate_label in seen_labels:
223 continue
224 if any(
225 suggestion_candidates_are_too_close(candidate, selected_candidate)
226 for selected_candidate in (correct, *selected)
227 ):
228 continue
229 if candidate.uri and candidate.uri in seen_uris:
230 continue
231 seen_labels.add(candidate_label)
232 if candidate.uri:
233 seen_uris.add(candidate.uri)
234 selected.append(candidate)
235 if limit is not None and len(selected) >= limit:
236 break
237 return selected
238
239
240def _select_distractors(
241 correct: SuggestionCandidate,
242 distractors: Iterable[SuggestionCandidate],
243 needed_count: int,
244) -> Sequence[SuggestionCandidate]:
245 """Return unique distractors that do not match the correct answer."""
246 selected = filter_suggestion_candidates(correct, distractors, limit=needed_count)
247 if len(selected) < needed_count:
248 msg = "Not enough distractors to build suggestions"
249 raise ValueError(msg)
250 return selected
251