/
/
1"""Phish.in Music Provider for Music Assistant."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from typing import TYPE_CHECKING, Any
7from urllib.parse import quote
8
9from music_assistant_models.enums import (
10 ContentType,
11 MediaType,
12 StreamType,
13)
14from music_assistant_models.errors import MediaNotFoundError, ProviderUnavailableError
15from music_assistant_models.media_items import (
16 Album,
17 Artist,
18 AudioFormat,
19 BrowseFolder,
20 ItemMapping,
21 Playlist,
22 SearchResults,
23 Track,
24)
25from music_assistant_models.streamdetails import StreamDetails
26
27from music_assistant.controllers.cache import use_cache
28from music_assistant.helpers.datetime import now
29from music_assistant.models.music_provider import MusicProvider
30
31from .constants import (
32 ENDPOINTS,
33 MAX_SEARCH_RESULTS,
34 PHISH_ARTIST_ID,
35)
36from .helpers import (
37 api_request,
38 get_phish_artist,
39 parse_search_results,
40 playlist_to_ma_playlist,
41 show_to_album,
42 track_to_ma_track,
43)
44
45if TYPE_CHECKING:
46 from collections.abc import Sequence
47
48 from music_assistant_models.config_entries import ConfigEntry
49 from music_assistant_models.media_items import MediaItemType
50
51
52class PhishInProvider(MusicProvider):
53 """Phish.in music provider."""
54
55 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
56 """Return Config entries to setup this provider."""
57 return ()
58
59 @property
60 def is_streaming_provider(self) -> bool:
61 """Return True if the provider is a streaming provider."""
62 return True
63
64 async def search(
65 self,
66 search_query: str,
67 media_types: list[MediaType],
68 limit: int = MAX_SEARCH_RESULTS,
69 ) -> SearchResults:
70 """Perform search on Phish.in."""
71 # Handle "Artist - Track" format by extracting just the track name
72 if " - " in search_query:
73 parts = search_query.split(" - ", 1)
74 if parts[0].strip().lower() in ["phish", "the phish"]:
75 search_query = parts[1].strip()
76
77 if len(search_query.strip()) < 3:
78 return SearchResults()
79
80 try:
81 # encode the term so values with "/" etc. don't break the URL path
82 endpoint = ENDPOINTS["search"].format(term=quote(search_query, safe=""))
83 search_data = await api_request(
84 self, endpoint, params={"audio_status": "complete_or_partial"}
85 )
86
87 # If we got song matches, fetch all performances of those songs
88 if MediaType.TRACK in media_types and search_data.get("songs"):
89 all_track_results = []
90 for song in search_data.get("songs", [])[:3]: # Limit to first 3 songs
91 song_slug = song.get("slug")
92 if song_slug:
93 tracks_data = await api_request(
94 self,
95 "/tracks",
96 params={
97 "song_slug": song_slug,
98 "audio_status": "complete_or_partial",
99 "per_page": limit,
100 "sort": "likes_count:desc",
101 },
102 )
103 all_track_results.extend(tracks_data.get("tracks", []))
104
105 # Replace with comprehensive song_slug results
106 if all_track_results:
107 search_data["tracks"] = all_track_results[:limit]
108
109 # Handle venue album searches
110 if MediaType.ALBUM in media_types and search_data.get("venues"):
111 venue_shows: list[dict[str, Any]] = []
112 for venue in search_data.get("venues", []):
113 venue_slug = venue["slug"]
114 page = 1
115 while len(venue_shows) < limit:
116 shows_data = await api_request(
117 self, "/shows", params={"venue_slug": venue_slug, "page": page}
118 )
119 shows_on_page = shows_data.get("shows", [])
120 if not shows_on_page:
121 break
122 remaining_slots = limit - len(venue_shows)
123 venue_shows.extend(shows_on_page[:remaining_slots])
124 current_page = shows_data.get("current_page", 1)
125 total_pages = shows_data.get("total_pages", 1)
126 if current_page >= total_pages or len(venue_shows) >= limit:
127 break
128 page += 1
129 if venue_shows:
130 search_data["venue_shows"] = venue_shows
131
132 artists, albums, tracks, playlists = parse_search_results(
133 self, search_data, media_types, search_query.lower()
134 )
135
136 return SearchResults(
137 artists=artists[:limit] if MediaType.ARTIST in media_types else [],
138 albums=albums[:limit] if MediaType.ALBUM in media_types else [],
139 tracks=tracks[:limit] if MediaType.TRACK in media_types else [],
140 playlists=playlists[:limit] if MediaType.PLAYLIST in media_types else [],
141 )
142 except MediaNotFoundError:
143 raise
144 except Exception as err:
145 self.logger.error("Search failed for query '%s': %s", search_query, err)
146 raise ProviderUnavailableError(f"Search error: {err}") from err
147
148 async def get_library_artists(self) -> AsyncGenerator[Artist]:
149 """Retrieve library artists from the provider."""
150 yield await get_phish_artist(self)
151
152 async def get_artist(self, prov_artist_id: str) -> Artist:
153 """Get full artist details by id."""
154 if prov_artist_id == PHISH_ARTIST_ID:
155 return await get_phish_artist(self)
156 raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
157
158 @use_cache(expiration=86400) # 24 hours - albums (ie. shows) could update daily
159 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
160 """Get a list of all albums for the given artist."""
161 if prov_artist_id != PHISH_ARTIST_ID:
162 raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
163
164 albums = []
165 page = 1
166 per_page = 750 # Phish.in limit is 1000 but this caused asyncio warnings
167
168 try:
169 while True:
170 shows_data = await api_request(
171 self,
172 ENDPOINTS["shows"],
173 params={
174 "page": page,
175 "per_page": per_page,
176 "audio_status": "complete_or_partial",
177 },
178 )
179
180 shows = shows_data.get("shows", [])
181 if not shows:
182 break
183
184 for show in shows:
185 if show.get("audio_status") in ["complete", "partial"]:
186 albums.append(show_to_album(self, show))
187
188 if len(shows) < per_page:
189 break
190
191 page += 1
192
193 return albums
194
195 except MediaNotFoundError, ProviderUnavailableError:
196 raise
197 except Exception as err:
198 self.logger.error("Failed to get artist albums: %s", err)
199 raise ProviderUnavailableError(f"Artist albums error: {err}") from err
200
201 @use_cache(expiration=2592000) # 30 days - Top tracks won't change that often as its voted on
202 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
203 """Get a list of most popular tracks for the given artist."""
204 if prov_artist_id != PHISH_ARTIST_ID:
205 raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
206
207 # single ranked page, mirroring Phish.in's own top-tracks page
208 tracks_data = await api_request(
209 self,
210 ENDPOINTS["tracks"],
211 params={
212 "page": 1,
213 "per_page": 50,
214 "sort": "likes_count:desc",
215 "audio_status": "complete_or_partial",
216 },
217 )
218 return [track_to_ma_track(self, track_data) for track_data in tracks_data.get("tracks", [])]
219
220 @use_cache(expiration=86400) # 24 hours - new performances can be added as shows are uploaded
221 async def get_artist_tracks(self, prov_artist_id: str) -> list[Track]:
222 """Get a list of all tracks for the given artist."""
223 if prov_artist_id != PHISH_ARTIST_ID:
224 raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
225
226 all_tracks: list[Track] = []
227 page = 1
228 per_page = 500
229 max_pages = 5 # cap at 2500 tracks for UI performance
230
231 while page <= max_pages:
232 tracks_data = await api_request(
233 self,
234 ENDPOINTS["tracks"],
235 params={
236 "page": page,
237 "per_page": per_page,
238 "sort": "likes_count:desc",
239 "audio_status": "complete_or_partial",
240 },
241 )
242
243 tracks_on_page = tracks_data.get("tracks", [])
244 if not tracks_on_page:
245 break
246
247 for track_data in tracks_on_page:
248 all_tracks.append(track_to_ma_track(self, track_data))
249
250 # short page means the last page was reached
251 if len(tracks_on_page) < per_page:
252 break
253
254 page += 1
255
256 return all_tracks
257
258 @use_cache(expiration=2592000) # 30 days - Show details from specific dates never change
259 async def get_album(self, prov_album_id: str) -> Album:
260 """Get full album details by id (show date)."""
261 try:
262 endpoint = ENDPOINTS["show_by_date"].format(date=prov_album_id)
263 show_data = await api_request(self, endpoint)
264
265 if not show_data:
266 raise MediaNotFoundError(f"Show {prov_album_id} not found")
267
268 return show_to_album(self, show_data)
269
270 except MediaNotFoundError:
271 raise
272 except Exception as err:
273 self.logger.error("Failed to get album %s: %s", prov_album_id, err)
274 raise ProviderUnavailableError(f"Album error: {err}") from err
275
276 @use_cache(expiration=2592000) # 30 days - Individual tracks never change once recorded
277 async def get_track(self, prov_track_id: str) -> Track:
278 """Get full track details by id."""
279 try:
280 endpoint = ENDPOINTS["track_by_id"].format(id=prov_track_id)
281 track_data = await api_request(self, endpoint)
282
283 if not track_data:
284 raise MediaNotFoundError(f"Track {prov_track_id} not found")
285
286 # Extract show data from the track response
287 show_data = track_data.get("show")
288
289 return track_to_ma_track(self, track_data, show_data)
290
291 except MediaNotFoundError:
292 raise
293 except Exception as err:
294 self.logger.error("Failed to get track %s: %s", prov_track_id, err)
295 raise ProviderUnavailableError(f"Track error: {err}") from err
296
297 @use_cache(expiration=2592000) # 30 days - Track listings for historical shows never change
298 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
299 """Get album tracks for given album id (show date)."""
300 try:
301 endpoint = ENDPOINTS["show_by_date"].format(date=prov_album_id)
302 show_data = await api_request(self, endpoint)
303
304 if not show_data:
305 raise MediaNotFoundError(f"Show {prov_album_id} not found")
306
307 tracks = []
308 for track_data in show_data.get("tracks", []):
309 track = track_to_ma_track(self, track_data, show_data)
310 tracks.append(track)
311
312 return tracks
313
314 except MediaNotFoundError:
315 raise
316 except Exception as err:
317 self.logger.error("Failed to get album tracks for %s: %s", prov_album_id, err)
318 raise ProviderUnavailableError(f"Album tracks error: {err}") from err
319
320 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
321 """Get streamdetails for a track."""
322 if media_type != MediaType.TRACK:
323 raise MediaNotFoundError(f"Streaming not supported for {media_type}")
324
325 try:
326 track = await self.get_track(item_id)
327
328 mp3_url = None
329 for mapping in track.provider_mappings:
330 if mapping.provider_instance == self.instance_id and mapping.url:
331 mp3_url = mapping.url
332 break
333
334 if not mp3_url:
335 raise MediaNotFoundError(f"No audio URL found for track {item_id}")
336
337 return StreamDetails(
338 provider=self.instance_id,
339 item_id=item_id,
340 audio_format=AudioFormat(
341 content_type=ContentType.MP3,
342 sample_rate=44100,
343 bit_depth=16,
344 channels=2,
345 ),
346 media_type=MediaType.TRACK,
347 stream_type=StreamType.HTTP,
348 path=mp3_url,
349 allow_seek=True,
350 can_seek=True,
351 )
352
353 except MediaNotFoundError, ProviderUnavailableError:
354 raise
355 except Exception as err:
356 self.logger.error("Failed to get stream details for %s: %s", item_id, err)
357 raise ProviderUnavailableError(f"Stream error: {err}") from err
358
359 @use_cache(expiration=86400) # 24 hours - Current year gets new shows added throughout the year
360 async def _get_years_data(self) -> Any:
361 """Get years data with caching."""
362 return await api_request(self, ENDPOINTS["years"])
363
364 @use_cache(expiration=86400) # 24 hours - recent shows could update daily
365 async def _get_recent_shows(self) -> Any:
366 """Get recent shows with caching."""
367 return await api_request(
368 self,
369 ENDPOINTS["shows"],
370 params={"per_page": 20, "sort": "date:desc", "audio_status": "complete_or_partial"},
371 )
372
373 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
374 """Retrieve library playlists from the provider."""
375 try:
376 for playlist_data in await self._get_playlists():
377 if playlist_data.get("tracks_count", 0) > 0:
378 yield playlist_to_ma_playlist(self, playlist_data)
379 except MediaNotFoundError, ProviderUnavailableError:
380 raise
381 except Exception as err:
382 self.logger.error("Failed to get library playlists: %s", err)
383 raise ProviderUnavailableError(f"Library playlists error: {err}") from err
384
385 @use_cache(expiration=86400) # 24 hours - Playlist metadata might be updated by users
386 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
387 """Get full playlist details by id."""
388 try:
389 playlist_info = await self._resolve_playlist(prov_playlist_id)
390 if not playlist_info:
391 raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
392 return playlist_to_ma_playlist(self, playlist_info)
393 except MediaNotFoundError:
394 raise
395 except Exception as err:
396 self.logger.error("Failed to get playlist %s: %s", prov_playlist_id, err)
397 raise ProviderUnavailableError(f"Playlist error: {err}") from err
398
399 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
400 """Get playlist tracks for given playlist id."""
401 if page > 0:
402 return []
403 try:
404 playlist_info = await self._resolve_playlist(prov_playlist_id)
405 if not playlist_info:
406 return []
407
408 playlist_data = await api_request(
409 self, ENDPOINTS["playlist_by_slug"].format(slug=playlist_info["slug"])
410 )
411
412 tracks = []
413 for entry in playlist_data.get("entries", []):
414 track_data = entry.get("track")
415 if track_data and track_data.get("mp3_url"):
416 tracks.append(track_to_ma_track(self, track_data))
417
418 return tracks
419
420 except MediaNotFoundError, ProviderUnavailableError:
421 raise
422 except Exception as err:
423 self.logger.error("Failed to get playlist tracks for %s: %s", prov_playlist_id, err)
424 raise ProviderUnavailableError(f"Playlist tracks error: {err}") from err
425
426 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
427 """Browse this provider's items."""
428 path_parts = [] if "://" not in path else path.split("://")[1].split("/")
429 subpath = path_parts[0] if path_parts else ""
430 subsubpath = "/".join(path_parts[1:]) if len(path_parts) > 1 else ""
431
432 if not subpath:
433 return self._browse_root(path)
434
435 if subpath == "playlists":
436 playlists = []
437 async for playlist in self.get_library_playlists():
438 playlists.append(playlist)
439 if len(playlists) >= 50:
440 break
441 return playlists
442 if subpath == "years":
443 return await self._browse_years(path, subsubpath)
444 if subpath == "recent":
445 return await self._browse_recent()
446 if subpath == "random":
447 return await self._browse_random()
448 if subpath == "today":
449 return await self._browse_today()
450 if subpath == "venues":
451 return await self._browse_venues(path, subsubpath)
452 if subpath == "tags":
453 return await self._browse_tags(path, subsubpath)
454 if subpath == "top_shows":
455 return await self._browse_top_shows()
456 if subpath == "top_tracks":
457 return await self._browse_top_tracks()
458
459 return []
460
461 def _browse_root(self, path: str) -> list[BrowseFolder]:
462 """Root level browse options."""
463 return [
464 BrowseFolder(
465 item_id="years",
466 provider=self.domain,
467 path=path + "years",
468 name="Browse by Year",
469 translation_key="browse_by_year",
470 ),
471 BrowseFolder(
472 item_id="today",
473 provider=self.domain,
474 path=path + "today",
475 name="This Day in Phish History",
476 translation_key="this_day_in_history",
477 ),
478 BrowseFolder(
479 item_id="recent",
480 provider=self.domain,
481 path=path + "recent",
482 name="Recent Shows",
483 translation_key="recent_shows",
484 ),
485 BrowseFolder(
486 item_id="venues",
487 provider=self.domain,
488 path=path + "venues",
489 name="Browse by Venue",
490 translation_key="browse_by_venue",
491 ),
492 BrowseFolder(
493 item_id="tags",
494 provider=self.domain,
495 path=path + "tags",
496 name="Browse by Tag",
497 translation_key="browse_by_tag",
498 ),
499 BrowseFolder(
500 item_id="playlists",
501 provider=self.domain,
502 path=path + "playlists",
503 name="User Playlists",
504 translation_key="user_playlists",
505 ),
506 BrowseFolder(
507 item_id="top_shows",
508 provider=self.domain,
509 path=path + "top_shows",
510 name="Top 46 Shows",
511 translation_key="top_shows",
512 ),
513 BrowseFolder(
514 item_id="top_tracks",
515 provider=self.domain,
516 path=path + "top_tracks",
517 name="Top 46 Tracks",
518 translation_key="top_tracks",
519 ),
520 BrowseFolder(
521 item_id="random",
522 provider=self.domain,
523 path=path + "random",
524 name="Random Show",
525 translation_key="random_show",
526 ),
527 ]
528
529 async def _browse_years(self, path: str, subsubpath: str) -> list[BrowseFolder | Album]:
530 """Browse shows by year/period."""
531 if not subsubpath:
532 try:
533 years_data = await self._get_years_data()
534 folders: list[BrowseFolder | Album] = []
535
536 for year_data in years_data:
537 period = year_data.get("period")
538 show_count = year_data.get("shows_count", 0)
539 if period and show_count > 0:
540 folders.append(
541 BrowseFolder(
542 item_id=f"period_{period}",
543 provider=self.domain,
544 path=f"phishin://years/{period}",
545 name=f"{period} ({show_count} shows)",
546 translation_key="year",
547 translation_params=[str(period), str(show_count)],
548 )
549 )
550
551 return sorted(folders, key=lambda x: x.name, reverse=True)
552
553 except MediaNotFoundError, ProviderUnavailableError:
554 raise
555 except Exception as err:
556 self.logger.error("Failed to browse years: %s", err)
557 raise ProviderUnavailableError(f"Browse years error: {err}") from err
558 else:
559 return await self._get_shows_for_period(subsubpath)
560
561 async def _browse_recent(self) -> list[Album]:
562 """Get recent shows."""
563 try:
564 shows_data = await self._get_recent_shows()
565 albums: list[Album] = []
566
567 for show in shows_data.get("shows", []):
568 if show.get("audio_status") in ["complete", "partial"]:
569 album = show_to_album(self, show)
570 albums.append(album)
571
572 return albums
573
574 except MediaNotFoundError, ProviderUnavailableError:
575 raise
576 except Exception as err:
577 self.logger.error("Failed to browse recent shows: %s", err)
578 raise ProviderUnavailableError(f"Browse recent error: {err}") from err
579
580 async def _browse_random(self) -> list[Album]:
581 """Get a random show."""
582 try:
583 show_data = await api_request(self, ENDPOINTS["random_show"])
584 if show_data and show_data.get("audio_status") in ["complete", "partial"]:
585 album = show_to_album(self, show_data)
586 return [album]
587 return []
588
589 except MediaNotFoundError, ProviderUnavailableError:
590 raise
591 except Exception as err:
592 self.logger.error("Failed to get random show: %s", err)
593 raise ProviderUnavailableError(f"Random show error: {err}") from err
594
595 @use_cache(expiration=21600) # 6 hours - today's shows are historical but queried daily
596 async def _browse_today(self) -> list[Album]:
597 """Get shows that happened on this day in history."""
598 try:
599 today = now()
600 target_date = today.strftime("%Y-%m-%d")
601
602 shows_data = await api_request(
603 self,
604 ENDPOINTS["shows_day_of_year"].format(date=target_date),
605 params={"audio_status": "complete_or_partial", "sort": "date:desc"},
606 )
607
608 albums: list[Album] = []
609 shows = shows_data.get("shows", [])
610
611 for show in shows:
612 if show and show.get("audio_status") in ["complete", "partial"]:
613 album = show_to_album(self, show)
614 albums.append(album)
615
616 return albums
617
618 except MediaNotFoundError:
619 self.logger.info("No shows found for %s", today.strftime("%B %d"))
620 return []
621 except ProviderUnavailableError:
622 raise
623 except Exception as err:
624 self.logger.error("Failed to get today's shows: %s", err)
625 raise ProviderUnavailableError(f"Today's shows error: {err}") from err
626
627 @use_cache(expiration=604800) # 7 days - venue list changes rarely
628 async def _browse_venues(self, path: str, subsubpath: str) -> list[BrowseFolder | Album]:
629 """Browse shows by venue."""
630 if not subsubpath:
631 try:
632 venues_data = await api_request(
633 self, ENDPOINTS["venues"], params={"per_page": 100, "sort": "shows_count:desc"}
634 )
635
636 folders: list[BrowseFolder | Album] = []
637 for venue in venues_data.get("venues", []):
638 audio_count = venue.get("shows_with_audio_count", 0)
639 if audio_count > 0:
640 folders.append(
641 BrowseFolder(
642 item_id=f"venue_{venue.get('slug')}",
643 provider=self.domain,
644 path=f"phishin://venues/{venue.get('slug')}",
645 name=f"{venue.get('name')} ({audio_count} shows)",
646 translation_key="venue",
647 translation_params=[str(venue.get("name")), str(audio_count)],
648 )
649 )
650
651 return folders[:50]
652
653 except MediaNotFoundError, ProviderUnavailableError:
654 raise
655 except Exception as err:
656 self.logger.error("Failed to browse venues: %s", err)
657 raise ProviderUnavailableError(f"Browse venues error: {err}") from err
658 else:
659 return await self._get_shows_for_venue(subsubpath)
660
661 @use_cache(expiration=604800) # 7 days - tags list changes rarely
662 async def _browse_tags(self, path: str, subsubpath: str) -> list[BrowseFolder | Album | Track]:
663 """Browse shows and tracks by tag."""
664 if not subsubpath:
665 try:
666 tags_data = await api_request(self, ENDPOINTS["tags"])
667
668 folders: list[BrowseFolder | Album | Track] = []
669 for tag in tags_data:
670 track_count = tag.get("tracks_count", 0)
671 show_count = tag.get("shows_count", 0)
672 if track_count > 0 or show_count > 0:
673 if show_count > 0:
674 count_str = f"{show_count} shows, {track_count} tracks"
675 translation_key = "tag_summary_shows_tracks"
676 translation_params = [
677 str(tag.get("name")),
678 str(show_count),
679 str(track_count),
680 ]
681 else:
682 count_str = f"{track_count} tracks"
683 translation_key = "tag_summary_tracks"
684 translation_params = [str(tag.get("name")), str(track_count)]
685 folders.append(
686 BrowseFolder(
687 item_id=f"tag_{tag.get('slug')}",
688 provider=self.domain,
689 path=f"phishin://tags/{tag.get('slug')}",
690 name=f"{tag.get('name')} ({count_str})",
691 translation_key=translation_key,
692 translation_params=translation_params,
693 )
694 )
695
696 return sorted(folders, key=lambda x: x.name)
697
698 except MediaNotFoundError, ProviderUnavailableError:
699 raise
700 except Exception as err:
701 self.logger.error("Failed to browse tags: %s", err)
702 raise ProviderUnavailableError(f"Browse tags error: {err}") from err
703
704 elif "/" not in subsubpath:
705 tag_slug = subsubpath
706 try:
707 tags_data = await api_request(self, ENDPOINTS["tags"])
708 tag_info: dict[str, Any] = next(
709 (tag for tag in tags_data if tag.get("slug") == tag_slug), {}
710 )
711 tag_name = tag_info.get("name", tag_slug)
712 show_count = tag_info.get("shows_count", 0)
713 track_count = tag_info.get("tracks_count", 0)
714
715 subfolders: list[BrowseFolder | Album | Track] = []
716
717 if show_count > 0:
718 subfolders.append(
719 BrowseFolder(
720 item_id=f"tag_shows_{tag_slug}",
721 provider=self.domain,
722 path=f"phishin://tags/{tag_slug}/shows",
723 name=f"Shows with {tag_name} ({show_count})",
724 translation_key="tag_shows",
725 translation_params=[str(tag_name), str(show_count)],
726 )
727 )
728
729 if track_count > 0:
730 subfolders.append(
731 BrowseFolder(
732 item_id=f"tag_tracks_{tag_slug}",
733 provider=self.domain,
734 path=f"phishin://tags/{tag_slug}/tracks",
735 name=f"All {tag_name} Tracks ({track_count})",
736 translation_key="tag_tracks",
737 translation_params=[str(tag_name), str(track_count)],
738 )
739 )
740
741 return subfolders
742
743 except MediaNotFoundError, ProviderUnavailableError:
744 raise
745 except Exception as err:
746 self.logger.error("Failed to get tag subfolders: %s", err)
747 raise ProviderUnavailableError(f"Tag subfolders error: {err}") from err
748 else:
749 tag_slug, content_type = subsubpath.split("/", 1)
750 if content_type == "shows":
751 return await self._get_shows_for_tag(tag_slug)
752 if content_type == "tracks":
753 return await self._get_tracks_for_tag(tag_slug)
754 return []
755
756 @use_cache(expiration=86400) # 24 hours - Tag associations could change as new shows are tagged
757 async def _get_tracks_for_tag(self, tag_slug: str) -> list[BrowseFolder | Album | Track]:
758 """Get tracks for a specific tag."""
759 try:
760 tracks_data = await api_request(
761 self,
762 ENDPOINTS["tracks"],
763 params={
764 "tag_slug": tag_slug,
765 "per_page": 100,
766 "audio_status": "complete_or_partial",
767 "sort": "likes_count:desc",
768 },
769 )
770
771 tracks: list[BrowseFolder | Album | Track] = []
772 for track_data in tracks_data.get("tracks", []):
773 if track_data.get("mp3_url"):
774 track = track_to_ma_track(self, track_data)
775 tracks.append(track)
776
777 return tracks
778
779 except MediaNotFoundError, ProviderUnavailableError:
780 raise
781 except Exception as err:
782 self.logger.error("Failed to get tracks for tag %s: %s", tag_slug, err)
783 raise ProviderUnavailableError(f"Tag tracks error: {err}") from err
784
785 async def _browse_top_shows(self) -> list[Album]:
786 """Get top 46 most liked shows."""
787 try:
788 shows_data = await api_request(
789 self,
790 ENDPOINTS["shows"],
791 params={
792 "per_page": 46,
793 "sort": "likes_count:desc",
794 "audio_status": "complete_or_partial",
795 },
796 )
797
798 albums: list[Album] = []
799 for show in shows_data.get("shows", []):
800 if show.get("audio_status") in ["complete", "partial"]:
801 album = show_to_album(self, show)
802 albums.append(album)
803
804 return albums
805
806 except MediaNotFoundError, ProviderUnavailableError:
807 raise
808 except Exception as err:
809 self.logger.error("Failed to get top shows: %s", err)
810 raise ProviderUnavailableError(f"Top shows error: {err}") from err
811
812 async def _browse_top_tracks(self) -> list[Track]:
813 """Get top 46 most liked tracks."""
814 try:
815 tracks_data = await api_request(
816 self,
817 ENDPOINTS["tracks"],
818 params={
819 "per_page": 46,
820 "sort": "likes_count:desc",
821 "audio_status": "complete_or_partial",
822 },
823 )
824
825 tracks: list[Track] = []
826 for track_data in tracks_data.get("tracks", []):
827 if track_data.get("mp3_url"):
828 track = track_to_ma_track(self, track_data)
829 tracks.append(track)
830
831 return tracks
832
833 except MediaNotFoundError, ProviderUnavailableError:
834 raise
835 except Exception as err:
836 self.logger.error("Failed to get top tracks: %s", err)
837 raise ProviderUnavailableError(f"Top tracks error: {err}") from err
838
839 @use_cache(expiration=86400) # 24 hours - Shows can be added to the current year
840 async def _get_shows_for_period(self, period: str) -> list[BrowseFolder | Album]:
841 """Get shows for a specific year or period."""
842 try:
843 if "-" in period and len(period.split("-")) == 2:
844 params = {
845 "year_range": period,
846 "per_page": 100,
847 "audio_status": "complete_or_partial",
848 }
849 else:
850 params = {
851 "year": period,
852 "per_page": 100,
853 "audio_status": "complete_or_partial",
854 }
855
856 shows_data = await api_request(self, ENDPOINTS["shows"], params=params)
857
858 albums: list[BrowseFolder | Album] = []
859 for show in shows_data.get("shows", []):
860 if show.get("audio_status") in ["complete", "partial"]:
861 album = show_to_album(self, show)
862 albums.append(album)
863
864 return sorted(albums, key=lambda x: x.name)
865
866 except MediaNotFoundError, ProviderUnavailableError:
867 raise
868 except Exception as err:
869 self.logger.error("Failed to browse period %s: %s", period, err)
870 raise ProviderUnavailableError(f"Browse period error: {err}") from err
871
872 @use_cache(expiration=86400) # 24 hours - Venues might get new shows added
873 async def _get_shows_for_venue(self, venue_slug: str) -> list[BrowseFolder | Album]:
874 """Get shows for a specific venue."""
875 try:
876 shows_data = await api_request(
877 self,
878 ENDPOINTS["shows"],
879 params={
880 "venue_slug": venue_slug,
881 "per_page": 100,
882 "audio_status": "complete_or_partial",
883 "sort": "date:desc",
884 },
885 )
886
887 albums: list[BrowseFolder | Album] = []
888 for show in shows_data.get("shows", []):
889 if show.get("audio_status") in ["complete", "partial"]:
890 album = show_to_album(self, show)
891 albums.append(album)
892
893 return albums
894
895 except MediaNotFoundError, ProviderUnavailableError:
896 raise
897 except Exception as err:
898 self.logger.error("Failed to get shows for venue %s: %s", venue_slug, err)
899 raise ProviderUnavailableError(f"Venue shows error: {err}") from err
900
901 @use_cache(expiration=86400) # 24 hours - Tag associations could change as new shows are tagged
902 async def _get_shows_for_tag(self, tag_slug: str) -> list[BrowseFolder | Album | Track]:
903 """Get shows for a specific tag."""
904 try:
905 shows_data = await api_request(
906 self,
907 ENDPOINTS["shows"],
908 params={
909 "tag_slug": tag_slug,
910 "per_page": 100,
911 "audio_status": "complete_or_partial",
912 "sort": "date:desc",
913 },
914 )
915
916 albums: list[BrowseFolder | Album | Track] = []
917 for show in shows_data.get("shows", []):
918 if show.get("audio_status") in ["complete", "partial"]:
919 album = show_to_album(self, show)
920 albums.append(album)
921
922 return albums
923
924 except MediaNotFoundError, ProviderUnavailableError:
925 raise
926 except Exception as err:
927 self.logger.error("Failed to get shows for tag %s: %s", tag_slug, err)
928 raise ProviderUnavailableError(f"Tag shows error: {err}") from err
929
930 @use_cache(expiration=86400) # 24 hours - the playlist list changes rarely
931 async def _get_playlists(self) -> list[dict[str, Any]]:
932 """Fetch the full list of Phish.in playlists (single cached request)."""
933 # cached and shared by all playlist methods to avoid repeated list fetches
934 playlists_data = await api_request(
935 self, ENDPOINTS["playlists"], params={"per_page": 100, "sort": "likes_count:desc"}
936 )
937 playlists: list[dict[str, Any]] = playlists_data.get("playlists", [])
938 return playlists
939
940 async def _resolve_playlist(self, prov_playlist_id: str) -> dict[str, Any] | None:
941 """Return the raw playlist data for the given id from the cached list."""
942 for playlist in await self._get_playlists():
943 if str(playlist.get("id")) == prov_playlist_id:
944 return playlist
945 return None
946