/
/
/
1"""Strict AI request and response helpers for Music Quiz distractors."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import unicodedata
8from collections.abc import Sequence
9from dataclasses import dataclass
10from typing import TYPE_CHECKING
11
12from music_assistant.helpers.json import JSON_DECODE_EXCEPTIONS, json_loads, strip_code_fence
13from music_assistant.helpers.plugin_engines import resolve_ai_engine
14from music_assistant.providers.music_quiz.ai_guards import (
15 ai_prompt_exceeds_limit,
16 validate_ai_response,
17)
18from music_assistant.providers.music_quiz.constants import AI_QUERY_TIMEOUT_SECONDS
19from music_assistant.providers.music_quiz.suggestions import answer_labels_are_too_close
20
21if TYPE_CHECKING:
22 from music_assistant.mass import MusicAssistant
23
24LOGGER = logging.getLogger(__name__)
25
26MAX_AI_CONTEXT_VALUE_LENGTH = 500
27MAX_AI_LABEL_LENGTH = 200
28MAX_AI_SYNTHETIC_COUNT = 12
29
30
31@dataclass(frozen=True, slots=True)
32class AISyntheticDistractor:
33 """A validated synthetic wrong-answer label returned by an AI provider."""
34
35 kind: str
36 label: str
37
38
39@dataclass(frozen=True, slots=True)
40class AIDistractorResponse:
41 """A validated AI ranking and synthetic distractor response."""
42
43 ranked_ids: tuple[str, ...]
44 synthetic: tuple[AISyntheticDistractor, ...]
45
46
47async def request_ai_distractors(
48 mass: MusicAssistant,
49 prompt: str,
50 *,
51 engine_uid: str | None,
52 timeout: float = AI_QUERY_TIMEOUT_SECONDS,
53) -> object | None:
54 """
55 Request distractors from the configured AI engine.
56
57 :param mass: Music Assistant instance used to discover AI engines.
58 :param prompt: Bounded prompt to submit.
59 :param engine_uid: The configured engine uid.
60 :param timeout: Maximum request duration in seconds.
61 :return: The untrusted engine response, or ``None`` when unavailable.
62 """
63 if ai_prompt_exceeds_limit(prompt):
64 return None
65 engine = await resolve_ai_engine(mass, engine_uid)
66 if engine is None:
67 return None
68 try:
69 async with asyncio.timeout(timeout):
70 return await engine.provider.ai_query(prompt, engine_id=engine.id)
71 except Exception as err:
72 LOGGER.debug(
73 "Music Quiz AI distractor request failed via %s (%s)",
74 engine.uid,
75 type(err).__name__,
76 )
77 return None
78
79
80def parse_ai_distractor_response(
81 response: object,
82 candidate_ids: Sequence[str],
83 expected_kinds: Sequence[str],
84) -> AIDistractorResponse:
85 """
86 Parse one exact AI distractor response.
87
88 :param response: Untrusted response returned by an AI provider.
89 :param candidate_ids: Server-owned candidate IDs the response must rank.
90 :param expected_kinds: Exact ordered synthetic distractor kinds requested.
91 :return: Strictly validated ranking and synthetic labels.
92 """
93 response_text = validate_ai_response(response)
94 if len(expected_kinds) > MAX_AI_SYNTHETIC_COUNT:
95 raise ValueError("too many synthetic distractors were requested")
96 if len(set(candidate_ids)) != len(candidate_ids):
97 raise ValueError("candidate IDs must be unique")
98 try:
99 payload = json_loads(strip_code_fence(response_text))
100 except JSON_DECODE_EXCEPTIONS as err:
101 raise ValueError("response is not valid JSON") from err
102 if not isinstance(payload, dict) or payload.keys() != {"ranked_ids", "synthetic"}:
103 raise ValueError("response must contain exactly ranked_ids and synthetic")
104
105 ranked_ids = payload["ranked_ids"]
106 expected_candidate_ids = set(candidate_ids)
107 if (
108 not isinstance(ranked_ids, list)
109 or len(ranked_ids) != len(candidate_ids)
110 or any(not isinstance(candidate_id, str) for candidate_id in ranked_ids)
111 or len(set(ranked_ids)) != len(ranked_ids)
112 or set(ranked_ids) != expected_candidate_ids
113 ):
114 raise ValueError("ranked_ids must be a complete candidate permutation")
115
116 synthetic = payload["synthetic"]
117 if not isinstance(synthetic, list) or len(synthetic) != len(expected_kinds):
118 raise ValueError("synthetic has an invalid shape")
119 parsed_synthetic: list[AISyntheticDistractor] = []
120 for raw_distractor, expected_kind in zip(synthetic, expected_kinds, strict=True):
121 if not isinstance(raw_distractor, dict) or raw_distractor.keys() != {"kind", "label"}:
122 raise ValueError("synthetic entries must contain exactly kind and label")
123 kind = raw_distractor["kind"]
124 if not isinstance(kind, str) or kind != expected_kind:
125 raise ValueError("synthetic distractor kind does not match the request")
126 label = _validate_ai_label(raw_distractor["label"])
127 if any(answer_labels_are_too_close(label, existing.label) for existing in parsed_synthetic):
128 raise ValueError("synthetic distractor labels must be distinct")
129 parsed_synthetic.append(AISyntheticDistractor(kind=kind, label=label))
130 return AIDistractorResponse(
131 ranked_ids=tuple(ranked_ids),
132 synthetic=tuple(parsed_synthetic),
133 )
134
135
136def bounded_ai_context(value: str | None) -> str | None:
137 """
138 Bound an untrusted metadata value before including it in an AI prompt.
139
140 :param value: Metadata value to bound.
141 :return: The bounded value, if present.
142 """
143 if value is None:
144 return None
145 return value[:MAX_AI_CONTEXT_VALUE_LENGTH]
146
147
148def _validate_ai_label(raw_label: object) -> str:
149 """Return a strict single-line synthetic label."""
150 if not isinstance(raw_label, str):
151 raise TypeError("synthetic distractor labels must be strings")
152 if not raw_label or raw_label != raw_label.strip():
153 raise ValueError("synthetic distractor labels must be non-empty and trimmed")
154 if len(raw_label) > MAX_AI_LABEL_LENGTH:
155 raise ValueError("synthetic distractor label exceeds the length limit")
156 if any(
157 unicodedata.category(character).startswith("C")
158 or unicodedata.category(character) in {"Zl", "Zp"}
159 for character in raw_label
160 ):
161 raise ValueError("synthetic distractor labels must not contain control characters")
162 return raw_label
163