""" Phish.net API v5 Importer Imports Phish setlist data from the official Phish.net API. Requires API key from https://phish.net/api API Documentation: https://phish.net/api/docs """ import os from datetime import datetime from typing import Dict, Optional, List from sqlmodel import Session from .base import ImporterBase class PhishImporter(ImporterBase): """Import Phish data from Phish.net API v5""" VERTICAL_NAME = "Phish" VERTICAL_SLUG = "phish" VERTICAL_DESCRIPTION = "Phish is an American rock band from Burlington, Vermont, formed in 1983." # Phish.net API settings BASE_URL = "https://api.phish.net/v5" def __init__(self, session: Session, api_key: Optional[str] = None): super().__init__(session) self.api_key = api_key or os.getenv("PHISHNET_API_KEY") if not self.api_key: raise ValueError( "Phish.net API key required. Set PHISHNET_API_KEY environment variable " "or pass api_key parameter. Get key at https://phish.net/api" ) def _api_get(self, endpoint: str, params: Optional[Dict] = None) -> Optional[Dict]: """Make API request with key""" url = f"{self.BASE_URL}/{endpoint}.json" params = params or {} params["apikey"] = self.api_key data = self.fetch_json(url, params) if data and data.get("error") == 0: return data.get("data", []) elif data: error_msg = data.get("error_message", "Unknown error") print(f"❌ API Error: {error_msg}") return None def import_all(self): """Run full import of all Phish data""" print("=" * 60) print("PHISH.NET DATA IMPORTER") print("=" * 60) # 1. Create/get vertical self.get_or_create_vertical() # 2. Import venues self.import_venues() # 3. Import songs self.import_songs() # 4. Import shows self.import_shows() # 5. Import setlists self.import_setlists() print("\n" + "=" * 60) print("✓ PHISH IMPORT COMPLETE!") print("=" * 60) print(f" • {len(self.venue_map)} venues") print(f" • {len(self.song_map)} songs") print(f" • {len(self.show_map)} shows") def import_venues(self) -> Dict[str, int]: """Import all Phish venues""" print("\n🏛️ Importing venues...") # Phish.net doesn't have a dedicated venues endpoint # Venues are included with show data, so we'll extract them during show import print(" (Venues will be extracted from show data)") return self.venue_map def import_songs(self) -> Dict[str, int]: """Import all Phish songs""" print("\n🎵 Importing songs...") data = self._api_get("songs") if not data: print("❌ Failed to fetch songs") return self.song_map for song_data in data: song_id = str(song_data.get("songid")) title = song_data.get("song", "Unknown") original_artist = song_data.get("original_artist") # Skip if not a real song title if not title or title == "Unknown": continue self.upsert_song( title=title, original_artist=original_artist if original_artist != "Phish" else None, external_id=song_id ) print(f"✓ Imported {len(self.song_map)} songs") return self.song_map def import_shows(self) -> Dict[str, int]: """Import all Phish shows""" print("\n🎤 Importing shows...") data = self._api_get("shows") if not data: print("❌ Failed to fetch shows") return self.show_map for show_data in data: show_id = str(show_data.get("showid")) show_date_str = show_data.get("showdate") if not show_date_str: continue try: show_date = datetime.strptime(show_date_str, "%Y-%m-%d") except ValueError: continue # Extract venue info venue_name = show_data.get("venue", "Unknown Venue") city = show_data.get("city", "Unknown") state = show_data.get("state") country = show_data.get("country", "USA") venue_id = self.upsert_venue( name=venue_name, city=city, state=state, country=country, external_id=f"phish_venue_{venue_name}_{city}" ) # Handle tour if present tour_id = None tour_name = show_data.get("tour_name") if tour_name: tour_id = self.upsert_tour( name=tour_name, external_id=f"phish_tour_{tour_name}" ) self.upsert_show( date=show_date, venue_id=venue_id, tour_id=tour_id, notes=show_data.get("setlistnotes"), external_id=show_id ) if len(self.show_map) % 100 == 0: print(f" Progress: {len(self.show_map)} shows...") print(f"✓ Imported {len(self.show_map)} shows") return self.show_map def import_setlists(self): """Import setlists for all shows""" print("\n📋 Importing setlists...") performance_count = 0 # Process each show for external_show_id, internal_show_id in self.show_map.items(): # Fetch setlist for this show by showid data = self._api_get(f"setlists/showid/{external_show_id}") if not data: continue for perf_data in data: song_id_ext = str(perf_data.get("songid")) song_id = self.song_map.get(song_id_ext) if not song_id: # Song not in our map, try to add it song_title = perf_data.get("song") if song_title: song_id = self.upsert_song(title=song_title, external_id=song_id_ext) if not song_id: continue # Parse set information set_name = self._parse_set_name(perf_data.get("set", "1")) position = perf_data.get("position", 0) # Check for segue (> symbol in transition) trans = perf_data.get("trans", "") segue = ">" in trans if trans else False self.upsert_performance( show_id=internal_show_id, song_id=song_id, position=position, set_name=set_name, segue=segue, notes=perf_data.get("footnote") ) performance_count += 1 if len(self.show_map) > 0 and performance_count % 500 == 0: print(f" Progress: {performance_count} performances...") print(f"✓ Imported {performance_count} performances") def _parse_set_name(self, set_value: str) -> str: """Convert Phish.net set notation to display name""" set_map = { "1": "Set 1", "2": "Set 2", "3": "Set 3", "e": "Encore", "e2": "Encore 2", "s": "Soundcheck", } return set_map.get(str(set_value).lower(), f"Set {set_value}") def main(): """Run Phish import""" from database import engine with Session(engine) as session: importer = PhishImporter(session) importer.import_all() if __name__ == "__main__": main()