/
/
/
1"""Helpers for the Smart Playlist plugin: rules dataclass, validation, and JSON I/O."""
2
3from __future__ import annotations
4
5import asyncio
6from contextlib import suppress
7from dataclasses import dataclass, field
8from pathlib import Path
9from typing import Any, cast
10
11import aiofiles
12from music_assistant_models.enums import AlbumType
13from music_assistant_models.errors import InvalidDataError
14
15from music_assistant.helpers.json import json_dumps, json_loads
16
17LOGIC_AND = "AND"
18LOGIC_OR = "OR"
19DEFAULT_TRACK_LIMIT = 100
20MAX_SEEDS = 10
21RULES_FILENAME = "smart_playlist_rules.json"
22
23
24def _coerce_int(value: Any, field_name: str) -> int:
25 try:
26 return int(value)
27 except (TypeError, ValueError) as err:
28 raise InvalidDataError(f"Invalid value for {field_name}: {value!r}") from err
29
30
31def _coerce_optional_int(value: Any, field_name: str) -> int | None:
32 if value is None:
33 return None
34 try:
35 return int(value)
36 except (TypeError, ValueError) as err:
37 raise InvalidDataError(f"Invalid value for {field_name}: {value!r}") from err
38
39
40def _coerce_optional_bool(value: Any, field_name: str) -> bool | None:
41 """Coerce a value to bool | None, raising InvalidDataError on bad input."""
42 if value is None:
43 return None
44 if isinstance(value, bool):
45 return value
46 raise InvalidDataError(
47 f"Invalid value for {field_name}: {value!r}. Expected True, False, or None."
48 )
49
50
51def _coerce_id_list(value: Any, field_name: str) -> list[int]:
52 """Coerce a value to a list of ints, raising InvalidDataError on bad input."""
53 if value is None:
54 return []
55 if not isinstance(value, list):
56 raise InvalidDataError(f"Expected list for {field_name}, got {type(value).__name__}")
57 result = []
58 for item in value:
59 try:
60 result.append(int(item))
61 except (TypeError, ValueError) as err:
62 raise InvalidDataError(f"Invalid id in {field_name}: {item!r}") from err
63 return result
64
65
66def _coerce_str_list(value: Any, field_name: str) -> list[str]:
67 """Coerce a value to a list of strs, raising InvalidDataError on bad input."""
68 if value is None:
69 return []
70 if not isinstance(value, list):
71 raise InvalidDataError(f"Expected list for {field_name}, got {type(value).__name__}")
72 return [str(item) for item in value]
73
74
75def _coerce_int_keyed_dict(value: Any, field_name: str) -> dict[int, str]:
76 """Coerce a value to a dict[int, str], raising InvalidDataError on bad input."""
77 if value is None:
78 return {}
79 if not isinstance(value, dict):
80 raise InvalidDataError(f"Expected dict for {field_name}, got {type(value).__name__}")
81 result = {}
82 for k, v in value.items():
83 try:
84 result[int(k)] = str(v)
85 except (TypeError, ValueError) as err:
86 raise InvalidDataError(f"Invalid key in {field_name}: {k!r}") from err
87 return result
88
89
90def _coerce_str_keyed_dict(value: Any, field_name: str) -> dict[str, str]:
91 """Coerce a value to a dict[str, str], raising InvalidDataError on bad input."""
92 if value is None:
93 return {}
94 if not isinstance(value, dict):
95 raise InvalidDataError(f"Expected dict for {field_name}, got {type(value).__name__}")
96 return {str(k): str(v) for k, v in value.items()}
97
98
99@dataclass
100class SmartPlaylistRules:
101 """Rules that define which tracks are included in a smart playlist."""
102
103 genre_ids: list[int] = field(default_factory=list)
104 artist_ids: list[int] = field(default_factory=list)
105 album_ids: list[int] = field(default_factory=list)
106 favorites_only: bool = False
107 seed_track_uris: list[str] = field(default_factory=list)
108 seed_artist_uris: list[str] = field(default_factory=list)
109 seed_album_uris: list[str] = field(default_factory=list)
110 seed_playlist_uris: list[str] = field(default_factory=list)
111 seed_names: dict[str, str] = field(default_factory=dict)
112 min_popularity: int | None = None
113 logic: str = LOGIC_AND
114 limit: int = DEFAULT_TRACK_LIMIT
115 is_dynamic: bool = True
116 genre_names: dict[int, str] = field(default_factory=dict)
117 artist_names: dict[int, str] = field(default_factory=dict)
118 album_names: dict[int, str] = field(default_factory=dict)
119 year_from: int | None = None
120 year_to: int | None = None
121 excluded_artist_ids: list[int] = field(default_factory=list)
122 excluded_album_ids: list[int] = field(default_factory=list)
123 excluded_track_uris: list[str] = field(default_factory=list)
124 excluded_artist_names: dict[int, str] = field(default_factory=dict)
125 excluded_album_names: dict[int, str] = field(default_factory=dict)
126 excluded_genre_ids: list[int] = field(default_factory=list)
127 excluded_genre_names: dict[int, str] = field(default_factory=dict)
128 album_types: list[str] = field(default_factory=list)
129 excluded_album_types: list[str] = field(default_factory=list)
130 explicit: bool | None = None
131 min_duration: int | None = None
132 max_duration: int | None = None
133 last_played_before_value: int | None = None
134 last_played_before_unit: str | None = None # "hours", "days", "weeks", "months"
135
136 def all_seed_uris(self) -> list[str]:
137 """Return every seed URI across the four seed lists, deduplicated, original order."""
138 seen: set[str] = set()
139 result: list[str] = []
140 for uri in (
141 *self.seed_track_uris,
142 *self.seed_artist_uris,
143 *self.seed_album_uris,
144 *self.seed_playlist_uris,
145 ):
146 if uri and uri not in seen:
147 seen.add(uri)
148 result.append(uri)
149 return result
150
151 def to_dict(self) -> dict[str, Any]:
152 """Serialize to dictionary."""
153 return {
154 "genre_ids": self.genre_ids,
155 "artist_ids": self.artist_ids,
156 "album_ids": self.album_ids,
157 "favorites_only": self.favorites_only,
158 "seed_track_uris": self.seed_track_uris,
159 "seed_artist_uris": self.seed_artist_uris,
160 "seed_album_uris": self.seed_album_uris,
161 "seed_playlist_uris": self.seed_playlist_uris,
162 "seed_names": self.seed_names,
163 "min_popularity": self.min_popularity,
164 "logic": self.logic,
165 "limit": self.limit,
166 "is_dynamic": self.is_dynamic,
167 "genre_names": {str(k): v for k, v in self.genre_names.items()},
168 "artist_names": {str(k): v for k, v in self.artist_names.items()},
169 "album_names": {str(k): v for k, v in self.album_names.items()},
170 "year_from": self.year_from,
171 "year_to": self.year_to,
172 "excluded_artist_ids": self.excluded_artist_ids,
173 "excluded_album_ids": self.excluded_album_ids,
174 "excluded_track_uris": self.excluded_track_uris,
175 "excluded_artist_names": {str(k): v for k, v in self.excluded_artist_names.items()},
176 "excluded_album_names": {str(k): v for k, v in self.excluded_album_names.items()},
177 "excluded_genre_ids": self.excluded_genre_ids,
178 "excluded_genre_names": {str(k): v for k, v in self.excluded_genre_names.items()},
179 "album_types": self.album_types,
180 "excluded_album_types": self.excluded_album_types,
181 "explicit": self.explicit,
182 "min_duration": self.min_duration,
183 "max_duration": self.max_duration,
184 "last_played_before_value": self.last_played_before_value,
185 "last_played_before_unit": self.last_played_before_unit,
186 }
187
188 @classmethod
189 def from_dict(cls, data: dict[str, Any]) -> SmartPlaylistRules:
190 """Deserialize from dictionary."""
191 return cls(
192 genre_ids=_coerce_id_list(data.get("genre_ids"), "genre_ids"),
193 artist_ids=_coerce_id_list(data.get("artist_ids"), "artist_ids"),
194 album_ids=_coerce_id_list(data.get("album_ids"), "album_ids"),
195 favorites_only=data.get("favorites_only", False),
196 seed_track_uris=_coerce_str_list(data.get("seed_track_uris"), "seed_track_uris"),
197 seed_artist_uris=_coerce_str_list(data.get("seed_artist_uris"), "seed_artist_uris"),
198 seed_album_uris=_coerce_str_list(data.get("seed_album_uris"), "seed_album_uris"),
199 seed_playlist_uris=_coerce_str_list(
200 data.get("seed_playlist_uris"), "seed_playlist_uris"
201 ),
202 seed_names=_coerce_str_keyed_dict(data.get("seed_names"), "seed_names"),
203 min_popularity=_coerce_optional_int(data.get("min_popularity"), "min_popularity"),
204 logic=data.get("logic", LOGIC_AND),
205 limit=_coerce_int(data.get("limit", DEFAULT_TRACK_LIMIT), "limit"),
206 is_dynamic=data.get("is_dynamic", True),
207 genre_names=_coerce_int_keyed_dict(data.get("genre_names"), "genre_names"),
208 artist_names=_coerce_int_keyed_dict(data.get("artist_names"), "artist_names"),
209 album_names=_coerce_int_keyed_dict(data.get("album_names"), "album_names"),
210 year_from=_coerce_optional_int(data.get("year_from"), "year_from"),
211 year_to=_coerce_optional_int(data.get("year_to"), "year_to"),
212 excluded_artist_ids=_coerce_id_list(
213 data.get("excluded_artist_ids"), "excluded_artist_ids"
214 ),
215 excluded_album_ids=_coerce_id_list(
216 data.get("excluded_album_ids"), "excluded_album_ids"
217 ),
218 excluded_track_uris=_coerce_str_list(
219 data.get("excluded_track_uris"), "excluded_track_uris"
220 ),
221 excluded_artist_names=_coerce_int_keyed_dict(
222 data.get("excluded_artist_names"), "excluded_artist_names"
223 ),
224 excluded_album_names=_coerce_int_keyed_dict(
225 data.get("excluded_album_names"), "excluded_album_names"
226 ),
227 excluded_genre_ids=_coerce_id_list(
228 data.get("excluded_genre_ids"), "excluded_genre_ids"
229 ),
230 excluded_genre_names=_coerce_int_keyed_dict(
231 data.get("excluded_genre_names"), "excluded_genre_names"
232 ),
233 album_types=_coerce_str_list(data.get("album_types"), "album_types"),
234 excluded_album_types=_coerce_str_list(
235 data.get("excluded_album_types"), "excluded_album_types"
236 ),
237 explicit=_coerce_optional_bool(data.get("explicit"), "explicit"),
238 min_duration=_coerce_optional_int(data.get("min_duration"), "min_duration"),
239 max_duration=_coerce_optional_int(data.get("max_duration"), "max_duration"),
240 last_played_before_value=_coerce_optional_int(
241 data.get("last_played_before_value"), "last_played_before_value"
242 ),
243 last_played_before_unit=data.get("last_played_before_unit"),
244 )
245
246 def human_readable(self) -> str:
247 """Return a human-readable summary of the rules."""
248 parts: list[str] = []
249 if self.favorites_only:
250 parts.append("Favorites only")
251 if self.explicit is True:
252 parts.append("Explicit only")
253 elif self.explicit is False:
254 parts.append("No explicit content")
255 if self.genre_ids:
256 names = [self.genre_names.get(gid, str(gid)) for gid in self.genre_ids]
257 parts.append(f"Genres: {', '.join(names)}")
258 if self.artist_ids:
259 names = [self.artist_names.get(aid, str(aid)) for aid in self.artist_ids]
260 parts.append(f"Artists: {', '.join(names)}")
261 if self.album_ids:
262 names = [self.album_names.get(aid, str(aid)) for aid in self.album_ids]
263 parts.append(f"Albums: {', '.join(names)}")
264 seed_uris = self.all_seed_uris()
265 if seed_uris:
266 labels = [self.seed_names.get(uri, uri) for uri in seed_uris]
267 parts.append(f"Similar to: {', '.join(labels)}")
268 if self.excluded_artist_ids:
269 names = [
270 self.excluded_artist_names.get(aid, str(aid)) for aid in self.excluded_artist_ids
271 ]
272 parts.append(f"Excl. artists: {', '.join(names)}")
273 if self.excluded_album_ids:
274 names = [
275 self.excluded_album_names.get(aid, str(aid)) for aid in self.excluded_album_ids
276 ]
277 parts.append(f"Excl. albums: {', '.join(names)}")
278 if self.excluded_genre_ids:
279 names = [
280 self.excluded_genre_names.get(gid, str(gid)) for gid in self.excluded_genre_ids
281 ]
282 parts.append(f"Excl. genres: {', '.join(names)}")
283 if self.excluded_track_uris:
284 parts.append(f"Excl. {len(self.excluded_track_uris)} track(s)")
285 if self.album_types:
286 parts.append(f"Album types: {', '.join(self.album_types)}")
287 if self.excluded_album_types:
288 parts.append(f"Excl. album types: {', '.join(self.excluded_album_types)}")
289 if self.min_popularity is not None:
290 parts.append(f"Min. popularity: {self.min_popularity}")
291 if year_range := self._format_year_range():
292 parts.append(year_range)
293 if duration_range := self._format_duration_range():
294 parts.append(duration_range)
295 if last_played_str := self._format_last_played():
296 parts.append(last_played_str)
297 if not parts:
298 return "No rules (all library tracks)"
299 connector = f" {self.logic} "
300 return connector.join(parts)
301
302 def _format_year_range(self) -> str | None:
303 """Format year filter as human-readable string."""
304 if self.year_from is not None and self.year_to is not None:
305 return f"Year: {self.year_from}-{self.year_to}"
306 if self.year_from is not None:
307 return f"Year: from {self.year_from}"
308 if self.year_to is not None:
309 return f"Year: to {self.year_to}"
310 return None
311
312 def _format_duration_range(self) -> str | None:
313 """Format duration filter as human-readable string."""
314 if self.min_duration is not None and self.max_duration is not None:
315 return f"Duration: {self.min_duration}-{self.max_duration}s"
316 if self.min_duration is not None:
317 return f"Duration: â¥{self.min_duration}s"
318 if self.max_duration is not None:
319 return f"Duration: â¤{self.max_duration}s"
320 return None
321
322 def _format_last_played(self) -> str | None:
323 """Format last played filter as human-readable string."""
324 if self.last_played_before_value is None or self.last_played_before_unit is None:
325 return None
326 unit_display = self.last_played_before_unit # hours, days, weeks, months
327 return f"Not played in {self.last_played_before_value} {unit_display}"
328
329
330def validate_rules(rules: SmartPlaylistRules) -> None:
331 """Raise InvalidDataError if any rule field is out of allowed range."""
332 if rules.logic not in (LOGIC_AND, LOGIC_OR):
333 msg = f"Invalid logic operator: {rules.logic}. Must be AND or OR."
334 raise InvalidDataError(msg)
335 if rules.limit < 1 or rules.limit > 2000:
336 msg = f"Track limit must be between 1 and 2000, got {rules.limit}"
337 raise InvalidDataError(msg)
338 if rules.min_popularity is not None and not (0 <= rules.min_popularity <= 100):
339 msg = f"min_popularity must be between 0 and 100, got {rules.min_popularity}"
340 raise InvalidDataError(msg)
341 if (
342 rules.year_from is not None
343 and rules.year_to is not None
344 and rules.year_from > rules.year_to
345 ):
346 msg = (
347 f"year_from must be less than or equal to year_to, got "
348 f"{rules.year_from}>{rules.year_to}"
349 )
350 raise InvalidDataError(msg)
351 total_seeds = len(rules.all_seed_uris())
352 if total_seeds > MAX_SEEDS:
353 msg = f"Too many seeds: {total_seeds} > {MAX_SEEDS}"
354 raise InvalidDataError(msg)
355 valid_album_types = {t.value for t in AlbumType}
356 for at in rules.album_types:
357 if at not in valid_album_types:
358 msg = f"Invalid album_types value: {at!r}. Must be one of {sorted(valid_album_types)}"
359 raise InvalidDataError(msg)
360 for at in rules.excluded_album_types:
361 if at not in valid_album_types:
362 msg = f"Invalid excluded_album_types value: {at!r}. Must be one of {sorted(valid_album_types)}"
363 raise InvalidDataError(msg)
364 if rules.min_duration is not None and rules.min_duration < 0:
365 msg = f"min_duration must be >= 0, got {rules.min_duration}"
366 raise InvalidDataError(msg)
367 if rules.max_duration is not None and rules.max_duration < 0:
368 msg = f"max_duration must be >= 0, got {rules.max_duration}"
369 raise InvalidDataError(msg)
370 if (
371 rules.min_duration is not None
372 and rules.max_duration is not None
373 and rules.min_duration > rules.max_duration
374 ):
375 msg = f"min_duration must be <= max_duration, got {rules.min_duration}>{rules.max_duration}"
376 raise InvalidDataError(msg)
377
378 # Validate last_played fields
379 if (rules.last_played_before_value is None) != (rules.last_played_before_unit is None):
380 msg = (
381 "last_played_before_value and last_played_before_unit must both be set or both be None"
382 )
383 raise InvalidDataError(msg)
384 if rules.last_played_before_value is not None and rules.last_played_before_value < 1:
385 msg = f"last_played_before_value must be >= 1, got {rules.last_played_before_value}"
386 raise InvalidDataError(msg)
387 if rules.last_played_before_unit is not None and rules.last_played_before_unit not in (
388 "hours",
389 "days",
390 "weeks",
391 "months",
392 ):
393 msg = f"last_played_before_unit must be hours/days/weeks/months, got {rules.last_played_before_unit}"
394 raise InvalidDataError(msg)
395
396
397async def read_json(path: str) -> dict[str, Any]:
398 """Read a JSON file and return its contents."""
399 async with aiofiles.open(path, encoding="utf-8") as fh:
400 return cast("dict[str, Any]", json_loads(await fh.read()))
401
402
403def _atomic_write(path: str, payload: str) -> None:
404 """Write payload to a temp file and atomically replace path, cleaning up on failure."""
405 tmp = Path(f"{path}.tmp")
406 replaced = False
407 try:
408 with tmp.open("w", encoding="utf-8") as fh:
409 fh.write(payload)
410 tmp.replace(path)
411 replaced = True
412 finally:
413 # On any failure mid-write, don't leave a stray temp file behind to accumulate.
414 if not replaced:
415 with suppress(OSError):
416 tmp.unlink(missing_ok=True)
417
418
419async def write_json(path: str, data: dict[str, Any]) -> None:
420 """Write data as JSON to a file atomically (temp file + replace, off the event loop)."""
421 # The whole write+rename+cleanup runs in one thread so a cancelled await can't race the
422 # rename against cleanup; the rename itself is atomic, so the destination is never truncated.
423 payload = json_dumps(data, indent=True)
424 await asyncio.to_thread(_atomic_write, path, payload)
425