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