b4cddf41ea
- Fork elmeg-demo codebase for multi-band support - Add data importer infrastructure with base class - Create band-specific importers: - phish.py: Phish.net API v5 - grateful_dead.py: Grateful Stats API - setlistfm.py: Dead & Company, Billy Strings (Setlist.fm) - Add spec-kit configuration for Gemini - Update README with supported bands and architecture
246 lines
8 KiB
Python
246 lines
8 KiB
Python
"""
|
|
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()
|