/
/
/
1"""Helper functions for Phish.in provider."""
2
3from __future__ import annotations
4
5import contextlib
6from collections.abc import Callable
7from typing import TYPE_CHECKING, Any
8
9import aiohttp
10from music_assistant_models.enums import AlbumType, ContentType, ExternalID, ImageType, MediaType
11from music_assistant_models.errors import MediaNotFoundError, ProviderUnavailableError
12from music_assistant_models.media_items import (
13 Album,
14 Artist,
15 AudioFormat,
16 ItemMapping,
17 MediaItemImage,
18 MediaItemMetadata,
19 Playlist,
20 ProviderMapping,
21 Track,
22)
23from music_assistant_models.unique_list import UniqueList
24
25from .constants import (
26 API_BASE_URL,
27 FALLBACK_ALBUM_IMAGE,
28 PHISH_ARTIST_ID,
29 PHISH_ARTIST_NAME,
30 PHISH_DISCOGS_ID,
31 PHISH_GENRES,
32 PHISH_MUSICBRAINZ_ID,
33 PHISH_TADB_ID,
34 REQUEST_TIMEOUT,
35)
36
37if TYPE_CHECKING:
38 from music_assistant.models.music_provider import MusicProvider
39
40
41async def api_request(
42 provider: MusicProvider,
43 endpoint: str,
44 params: dict[str, Any] | None = None,
45) -> Any:
46 """Make an API request to Phish.in."""
47 url = f"{API_BASE_URL}{endpoint}"
48
49 try:
50 async with provider.mass.http_session.get(
51 url,
52 params=params,
53 timeout=aiohttp.ClientTimeout(total=REQUEST_TIMEOUT),
54 ) as response:
55 if response.status == 404:
56 raise MediaNotFoundError(f"Resource not found: {url}")
57 response.raise_for_status()
58 return await response.json()
59 except MediaNotFoundError:
60 raise
61 except aiohttp.ClientError as err:
62 provider.logger.error("API request failed for %s: %s", url, err)
63 raise ProviderUnavailableError(f"Phish.in API unavailable: {err}") from err
64
65
66def show_to_album(
67 provider: MusicProvider,
68 show_data: dict[str, Any],
69 *,
70 available: bool | None = None,
71) -> Album:
72 """
73 Convert a Phish.in show to a Music Assistant Album.
74
75 :param available: Override album availability; defaults to the show's audio_status.
76 """
77 show_date = show_data.get("date", "")
78 venue_data = show_data.get("venue", {})
79 venue_name = venue_data.get("name", "Unknown Venue")
80 location = venue_data.get("location", "")
81
82 album_name = f"{show_date} - {venue_name}"
83 if location:
84 album_name += f", {location}"
85
86 # Create metadata with image
87 album_cover_url = show_data.get("album_cover_url") or FALLBACK_ALBUM_IMAGE
88 metadata = MediaItemMetadata(
89 images=UniqueList(
90 [
91 MediaItemImage(
92 type=ImageType.THUMB,
93 path=album_cover_url,
94 provider=provider.instance_id,
95 remotely_accessible=True,
96 )
97 ]
98 )
99 )
100
101 # Parse year from date string (YYYY-MM-DD format)
102 year = None
103 if show_date and "-" in show_date:
104 with contextlib.suppress(ValueError, IndexError):
105 year = int(show_date.split("-")[0])
106
107 # Create details string for provider mapping
108 details_parts = [f"venue:{venue_name}"]
109 if location:
110 details_parts.append(f"location:{location}")
111 if show_data.get("duration"):
112 details_parts.append(f"duration:{show_data.get('duration')}")
113
114 audio_status = show_data.get("audio_status", "missing")
115 details_parts.append(f"audio_status:{audio_status}")
116 is_available = available if available is not None else audio_status in ["complete", "partial"]
117
118 if show_data.get("tour_name"):
119 details_parts.append(f"tour:{show_data.get('tour_name')}")
120
121 # Create ItemMapping for Phish artist
122 phish_artist = ItemMapping(
123 item_id=PHISH_ARTIST_ID,
124 provider=provider.instance_id,
125 name=PHISH_ARTIST_NAME,
126 media_type=MediaType.ARTIST,
127 available=True,
128 )
129
130 return Album(
131 item_id=show_date,
132 provider=provider.instance_id,
133 name=album_name,
134 artists=UniqueList([phish_artist]),
135 year=year,
136 album_type=AlbumType.LIVE,
137 metadata=metadata,
138 provider_mappings={
139 ProviderMapping(
140 item_id=show_date,
141 provider_domain=provider.domain,
142 provider_instance=provider.instance_id,
143 available=is_available,
144 audio_format=AudioFormat(content_type=ContentType.MP3),
145 details="|".join(details_parts),
146 )
147 },
148 )
149
150
151async def get_phish_artist(provider: MusicProvider) -> Artist:
152 """Get the main Phish artist object."""
153 artist = Artist(
154 item_id=PHISH_ARTIST_ID,
155 provider=provider.instance_id,
156 name=PHISH_ARTIST_NAME,
157 provider_mappings={
158 ProviderMapping(
159 item_id=PHISH_ARTIST_ID,
160 provider_domain=provider.domain,
161 provider_instance=provider.instance_id,
162 available=True,
163 )
164 },
165 )
166
167 # Add external IDs for metadata enrichment
168 artist.add_external_id(ExternalID.MB_ARTIST, PHISH_MUSICBRAINZ_ID)
169 artist.add_external_id(ExternalID.DISCOGS, PHISH_DISCOGS_ID)
170 artist.add_external_id(ExternalID.TADB, PHISH_TADB_ID)
171
172 return artist
173
174
175def _extract_version_from_title(full_title: str) -> tuple[str, str]:
176 """
177 Extract song title and version from full title with performance indicators.
178
179 Returns:
180 Tuple of (clean_song_title, version_string)
181 """
182 song_title = full_title
183 version = None
184 performance_indicators = ["set", "soundcheck", "check", "encore"]
185
186 # Check for prefix: "(Check) Song Name"
187 if full_title.startswith("(") and ") " in full_title:
188 end_paren = full_title.index(") ")
189 prefix = full_title[1:end_paren]
190 if any(indicator in prefix.lower() for indicator in performance_indicators):
191 version = prefix
192 song_title = full_title[end_paren + 2 :]
193
194 # Check for suffix: "Song Name (Soundcheck)"
195 if " (" in song_title and song_title.endswith(")"):
196 base_title, suffix = song_title.rsplit(" (", 1)
197 suffix = suffix.rstrip(")")
198 if any(indicator in suffix.lower() for indicator in performance_indicators):
199 version = f"{version}, {suffix}" if version else suffix
200 song_title = base_title
201
202 return song_title, version or ""
203
204
205def _build_track_details(
206 track_data: dict[str, Any],
207 song_data: dict[str, Any],
208 show_date: str,
209 set_name: str,
210 venue_name: str,
211) -> str:
212 """Build details string for provider mapping."""
213 details_parts = [f"song_slug:{song_data.get('slug', '')}"]
214
215 if set_name:
216 details_parts.append(f"set_name:{set_name}")
217 if show_date:
218 details_parts.append(f"show_date:{show_date}")
219 if venue_name:
220 details_parts.append(f"venue:{venue_name}")
221 if track_data.get("tags"):
222 tag_names = [tag.get("name", "") for tag in track_data.get("tags", [])]
223 details_parts.append(f"tags:{','.join(tag_names)}")
224 if track_data.get("likes_count"):
225 details_parts.append(f"likes_count:{track_data.get('likes_count', 0)}")
226
227 return "|".join(details_parts)
228
229
230def _flat_track_show(track_data: dict[str, Any]) -> dict[str, Any]:
231 """
232 Build a show_data dict from a track's flat fields.
233
234 Some endpoints (search results, playlist entries) return track objects
235 without a nested ``show`` object, exposing the show details as flat fields.
236 """
237 return {
238 "date": track_data.get("show_date"),
239 "album_cover_url": track_data.get("show_album_cover_url"),
240 "venue": {
241 "name": track_data.get("venue_name"),
242 "location": track_data.get("venue_location"),
243 },
244 }
245
246
247def track_to_ma_track(
248 provider: MusicProvider,
249 track_data: dict[str, Any],
250 show_data: dict[str, Any] | None = None,
251) -> Track:
252 """Convert a Phish.in track to a Music Assistant Track."""
253 track_id = str(track_data.get("id", ""))
254
255 # Extract song info and version
256 songs = track_data.get("songs", [])
257 song_data = songs[0] if songs else {}
258 full_title = track_data.get("title", "Unknown Song")
259 song_title, version = _extract_version_from_title(full_title)
260
261 # Extract basic track info
262 duration_ms = track_data.get("duration")
263 duration = int(duration_ms / 1000) if duration_ms else 0
264 position = track_data.get("position")
265 track_number = int(position) if position is not None else 0
266 set_name = track_data.get("set_name", "")
267
268 # Get show information; fall back to the nested show or flat track fields
269 if show_data is None:
270 show_data = track_data.get("show") or _flat_track_show(track_data)
271 show_date = show_data.get("date", "")
272 venue_name = show_data.get("venue", {}).get("name", "")
273
274 # Create artist mapping
275 phish_artist = ItemMapping(
276 item_id=PHISH_ARTIST_ID,
277 provider=provider.instance_id,
278 name=PHISH_ARTIST_NAME,
279 media_type=MediaType.ARTIST,
280 available=True,
281 )
282
283 # Build the parent album (full Album so the track shows under "Appears On").
284 # Tracks we surface always have audio, so force the album available.
285 track_album = show_to_album(provider, show_data, available=True) if show_date else None
286
287 # Build details string
288 details = _build_track_details(track_data, song_data, show_date, set_name, venue_name)
289
290 # Create metadata with genres and image
291 metadata = MediaItemMetadata(genres=set(PHISH_GENRES))
292 if show_data and (image_url := show_data.get("album_cover_url")):
293 metadata.images = UniqueList(
294 [
295 MediaItemImage(
296 type=ImageType.THUMB,
297 path=image_url,
298 provider=provider.instance_id,
299 remotely_accessible=True,
300 )
301 ]
302 )
303
304 return Track(
305 item_id=track_id,
306 provider=provider.instance_id,
307 name=song_title,
308 version=version,
309 artists=UniqueList([phish_artist]),
310 album=track_album,
311 duration=duration,
312 track_number=track_number,
313 metadata=metadata,
314 provider_mappings={
315 ProviderMapping(
316 item_id=track_id,
317 provider_domain=provider.domain,
318 provider_instance=provider.instance_id,
319 available=bool(track_data.get("mp3_url")),
320 audio_format=AudioFormat(content_type=ContentType.MP3),
321 url=track_data.get("mp3_url"),
322 details=details,
323 )
324 },
325 )
326
327
328def playlist_to_ma_playlist(provider: MusicProvider, playlist_data: dict[str, Any]) -> Playlist:
329 """Convert phish.in playlist data to Music Assistant Playlist."""
330 playlist_id = str(playlist_data["id"])
331
332 metadata = MediaItemMetadata(
333 description=playlist_data.get("description"),
334 images=UniqueList(
335 [
336 MediaItemImage(
337 type=ImageType.THUMB,
338 path=FALLBACK_ALBUM_IMAGE,
339 provider=provider.instance_id,
340 remotely_accessible=True,
341 )
342 ]
343 ),
344 )
345
346 return Playlist(
347 item_id=playlist_id,
348 provider=provider.instance_id,
349 name=playlist_data.get("name", ""),
350 owner=playlist_data.get("username", ""),
351 is_editable=False,
352 metadata=metadata,
353 provider_mappings={
354 ProviderMapping(
355 item_id=playlist_id,
356 provider_domain=provider.domain,
357 provider_instance=provider.instance_id,
358 available=True,
359 )
360 },
361 )
362
363
364def get_main_artist_mapping(provider: MusicProvider) -> ProviderMapping:
365 """Get artist mapping for Phish."""
366 return ProviderMapping(
367 item_id=PHISH_ARTIST_ID,
368 provider_domain=provider.domain,
369 provider_instance=provider.instance_id,
370 available=True,
371 )
372
373
374def get_album_mapping(provider: MusicProvider, show_date: str) -> ProviderMapping:
375 """Get album mapping for a show date."""
376 return ProviderMapping(
377 item_id=show_date,
378 provider_domain=provider.domain,
379 provider_instance=provider.instance_id,
380 available=True,
381 )
382
383
384def parse_search_results(
385 provider: MusicProvider,
386 search_data: dict[str, Any],
387 media_types: list[MediaType],
388 search_query: str,
389) -> tuple[list[Artist], list[Album], list[Track], list[Playlist]]:
390 """Parse search results into MA media items."""
391 search_term = search_query.lower()
392
393 def contains_search_term(text: str | None) -> bool:
394 return search_term in text.lower() if text else False
395
396 def strip_performance_indicators(title: str) -> str:
397 """Strip performance indicators like (Set1), (Soundcheck), etc. from title."""
398 song_title = title
399 performance_indicators = ["set", "soundcheck", "check", "encore"]
400
401 # Check for prefix: "(Check) Song"
402 if song_title.startswith("(") and ") " in song_title:
403 end_paren = song_title.index(") ")
404 prefix = song_title[1:end_paren]
405 if any(indicator in prefix.lower() for indicator in performance_indicators):
406 song_title = song_title[end_paren + 2 :]
407
408 # Check for suffix: "Song (Set1)"
409 if " (" in song_title and song_title.endswith(")"):
410 base_title, suffix = song_title.rsplit(" (", 1)
411 suffix = suffix.rstrip(")")
412 if any(indicator in suffix.lower() for indicator in performance_indicators):
413 song_title = base_title
414
415 return song_title
416
417 artists: list[Artist] = _parse_artists(provider, media_types, search_term)
418 albums: list[Album] = _parse_albums(provider, search_data, media_types, contains_search_term)
419 tracks: list[Track] = _parse_tracks(
420 provider, search_data, media_types, contains_search_term, strip_performance_indicators
421 )
422 playlists: list[Playlist] = _parse_playlists(
423 provider, search_data, media_types, contains_search_term
424 )
425
426 return artists, albums, tracks, playlists
427
428
429def _parse_artists(
430 provider: MusicProvider, media_types: list[MediaType], search_term: str
431) -> list[Artist]:
432 """Parse artists from search results."""
433 artists: list[Artist] = []
434 # Phish.in hosts a single artist and its api has no artist search, so only
435 # claim a match when the query is the start of its name.
436 if MediaType.ARTIST in media_types and PHISH_ARTIST_NAME.lower().startswith(search_term):
437 metadata = MediaItemMetadata(
438 images=UniqueList(
439 [
440 MediaItemImage(
441 type=ImageType.THUMB,
442 path=FALLBACK_ALBUM_IMAGE,
443 provider=provider.instance_id,
444 remotely_accessible=True,
445 )
446 ]
447 )
448 )
449
450 phish_artist_full = Artist(
451 item_id=PHISH_ARTIST_ID,
452 provider=provider.instance_id,
453 name=PHISH_ARTIST_NAME,
454 metadata=metadata,
455 provider_mappings={
456 ProviderMapping(
457 item_id=PHISH_ARTIST_ID,
458 provider_domain=provider.domain,
459 provider_instance=provider.instance_id,
460 available=True,
461 )
462 },
463 )
464 artists.append(phish_artist_full)
465
466 return artists
467
468
469def _parse_albums(
470 provider: MusicProvider,
471 search_data: dict[str, Any],
472 media_types: list[MediaType],
473 contains_search_term: Callable[[str | None], bool],
474) -> list[Album]:
475 """Parse albums from search results."""
476 albums: list[Album] = []
477 if MediaType.ALBUM not in media_types:
478 return albums
479
480 # Add exact show if present
481 if search_data.get("exact_show"):
482 show = search_data["exact_show"]
483 venue_name = show.get("venue_name", "")
484 if contains_search_term(venue_name):
485 albums.append(show_to_album(provider, show))
486
487 # Add other shows
488 for show in search_data.get("other_shows", []):
489 venue_name = show.get("venue_name", "")
490 if contains_search_term(venue_name):
491 albums.append(show_to_album(provider, show))
492
493 # Add venue shows (from additional API calls)
494 for show in search_data.get("venue_shows", []):
495 venue_name = show.get("venue_name", "")
496 if contains_search_term(venue_name):
497 albums.append(show_to_album(provider, show))
498
499 return albums
500
501
502def _parse_tracks(
503 provider: MusicProvider,
504 search_data: dict[str, Any],
505 media_types: list[MediaType],
506 contains_search_term: Callable[[str | None], bool],
507 strip_performance_indicators: Callable[[str], str],
508) -> list[Track]:
509 """Parse tracks from search results."""
510 tracks: list[Track] = []
511 if MediaType.TRACK not in media_types:
512 return tracks
513
514 for track_data in search_data.get("tracks", []):
515 full_title = track_data.get("title", "")
516 # Strip performance indicators to get base song name for matching
517 clean_title = strip_performance_indicators(full_title)
518
519 if contains_search_term(clean_title):
520 tracks.append(track_to_ma_track(provider, track_data))
521
522 # Deduplicate by album - only return one track per show
523 seen_albums = set()
524 unique_tracks = []
525 for track in tracks:
526 album_id = track.album.item_id if track.album else None
527 if album_id and album_id not in seen_albums:
528 seen_albums.add(album_id)
529 unique_tracks.append(track)
530 elif not album_id:
531 unique_tracks.append(track)
532
533 return unique_tracks
534
535
536def _parse_playlists(
537 provider: MusicProvider,
538 search_data: dict[str, Any],
539 media_types: list[MediaType],
540 contains_search_term: Callable[[str | None], bool],
541) -> list[Playlist]:
542 """Parse playlists from search results."""
543 playlists: list[Playlist] = []
544 if MediaType.PLAYLIST in media_types:
545 for playlist_data in search_data.get("playlists", []):
546 playlist_name = playlist_data.get("name", "")
547 if contains_search_term(playlist_name):
548 playlists.append(playlist_to_ma_playlist(provider, playlist_data))
549
550 return playlists
551