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