249 lines
8.6 KiB
Python
249 lines
8.6 KiB
Python
"""
|
|
Grateful Stats API Importer
|
|
|
|
Imports Grateful Dead setlist data from gratefulstats.com API.
|
|
This is the primary source for comprehensive Grateful Dead data (1965-1995).
|
|
|
|
API Documentation: https://gratefulstats.com/api
|
|
"""
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Dict, Optional, List
|
|
from sqlmodel import Session
|
|
|
|
from .base import ImporterBase
|
|
|
|
|
|
class GratefulDeadImporter(ImporterBase):
|
|
"""Import Grateful Dead data from Grateful Stats API"""
|
|
|
|
VERTICAL_NAME = "Grateful Dead"
|
|
VERTICAL_SLUG = "grateful-dead"
|
|
VERTICAL_DESCRIPTION = "The Grateful Dead was an American rock band formed in 1965 in Palo Alto, California. Known for their eclectic style and live performances, they are considered one of the most influential bands of all time."
|
|
|
|
# Grateful Stats API settings
|
|
BASE_URL = "https://gratefulstats.com/api"
|
|
|
|
def __init__(self, session: Session, api_key: Optional[str] = None):
|
|
super().__init__(session)
|
|
self.api_key = api_key or os.getenv("GRATEFULSTATS_API_KEY")
|
|
# Grateful Stats may not require API key for basic access
|
|
|
|
def _api_get(self, endpoint: str, params: Optional[Dict] = None) -> Optional[Dict]:
|
|
"""Make API request"""
|
|
url = f"{self.BASE_URL}/{endpoint}"
|
|
if self.api_key:
|
|
params = params or {}
|
|
params["api_key"] = self.api_key
|
|
|
|
return self.fetch_json(url, params)
|
|
|
|
def import_all(self):
|
|
"""Run full import of all Grateful Dead data"""
|
|
print("=" * 60)
|
|
print("GRATEFUL DEAD DATA IMPORTER (via Grateful Stats)")
|
|
print("=" * 60)
|
|
|
|
# 1. Create/get vertical
|
|
self.get_or_create_vertical()
|
|
|
|
# 2. Import songs first
|
|
self.import_songs()
|
|
|
|
# 3. Import shows by year (1965-1995)
|
|
self.import_shows()
|
|
|
|
# 4. Import setlists for each show
|
|
self.import_setlists()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✓ GRATEFUL DEAD 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 venues (handled during show import for GD)"""
|
|
return self.venue_map
|
|
|
|
def import_songs(self) -> Dict[str, int]:
|
|
"""Import all Grateful Dead songs"""
|
|
print("\n🎵 Importing songs...")
|
|
|
|
data = self._api_get("songs")
|
|
if not data:
|
|
print(" ⚠️ Could not fetch songs from API, will extract from setlists")
|
|
return self.song_map
|
|
|
|
songs = data if isinstance(data, list) else data.get("songs", [])
|
|
|
|
for song_data in songs:
|
|
song_id = str(song_data.get("id", ""))
|
|
title = song_data.get("name") or song_data.get("title", "Unknown")
|
|
original_artist = song_data.get("original_artist")
|
|
|
|
if not title or title == "Unknown":
|
|
continue
|
|
|
|
self.upsert_song(
|
|
title=title,
|
|
original_artist=original_artist,
|
|
external_id=song_id if song_id else None
|
|
)
|
|
|
|
print(f"✓ Imported {len(self.song_map)} songs")
|
|
return self.song_map
|
|
|
|
def import_shows(self) -> Dict[str, int]:
|
|
"""Import all Grateful Dead shows (1965-1995)"""
|
|
print("\n🎤 Importing shows...")
|
|
|
|
# Grateful Dead active years
|
|
years = range(1965, 1996)
|
|
|
|
for year in years:
|
|
print(f" Fetching {year}...", end="", flush=True)
|
|
data = self._api_get(f"shows/{year}")
|
|
|
|
if not data:
|
|
print(" (no data)")
|
|
continue
|
|
|
|
shows = data if isinstance(data, list) else data.get("shows", [])
|
|
year_count = 0
|
|
|
|
for show_data in shows:
|
|
show_id = str(show_data.get("id", ""))
|
|
show_date_str = show_data.get("date") or show_data.get("showdate")
|
|
|
|
if not show_date_str:
|
|
continue
|
|
|
|
try:
|
|
# Try various date formats
|
|
for fmt in ["%Y-%m-%d", "%m/%d/%Y", "%d-%m-%Y"]:
|
|
try:
|
|
show_date = datetime.strptime(show_date_str, fmt)
|
|
break
|
|
except ValueError:
|
|
continue
|
|
else:
|
|
continue
|
|
except Exception:
|
|
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"gd_venue_{venue_name}_{city}"
|
|
)
|
|
|
|
# Handle tour if present
|
|
tour_id = None
|
|
tour_name = show_data.get("tour") or show_data.get("tour_name")
|
|
if tour_name:
|
|
tour_id = self.upsert_tour(
|
|
name=tour_name,
|
|
external_id=f"gd_tour_{tour_name}"
|
|
)
|
|
|
|
self.upsert_show(
|
|
date=show_date,
|
|
venue_id=venue_id,
|
|
tour_id=tour_id,
|
|
notes=show_data.get("notes"),
|
|
external_id=show_id if show_id else None
|
|
)
|
|
year_count += 1
|
|
|
|
print(f" {year_count} 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
|
|
|
|
for external_show_id, internal_show_id in self.show_map.items():
|
|
if not external_show_id:
|
|
continue
|
|
|
|
data = self._api_get(f"shows/{external_show_id}/setlist")
|
|
|
|
if not data:
|
|
continue
|
|
|
|
setlist = data if isinstance(data, list) else data.get("setlist", [])
|
|
|
|
for perf_data in setlist:
|
|
song_name = perf_data.get("song") or perf_data.get("name")
|
|
if not song_name:
|
|
continue
|
|
|
|
# Get or create song
|
|
song_id = self.upsert_song(
|
|
title=song_name,
|
|
original_artist=perf_data.get("original_artist"),
|
|
external_id=f"gd_song_{song_name}"
|
|
)
|
|
|
|
# Parse set information
|
|
set_val = perf_data.get("set", "1")
|
|
set_name = self._parse_set_name(set_val)
|
|
position = perf_data.get("position", performance_count + 1)
|
|
|
|
# Check for segue
|
|
segue = perf_data.get("segue", False) or ">" in str(perf_data.get("transition", ""))
|
|
|
|
self.upsert_performance(
|
|
show_id=internal_show_id,
|
|
song_id=song_id,
|
|
position=position,
|
|
set_name=set_name,
|
|
segue=segue,
|
|
notes=perf_data.get("notes")
|
|
)
|
|
performance_count += 1
|
|
|
|
if performance_count % 1000 == 0:
|
|
print(f" Progress: {performance_count} performances...")
|
|
|
|
print(f"✓ Imported {performance_count} performances")
|
|
|
|
def _parse_set_name(self, set_value) -> str:
|
|
"""Convert set notation to display name"""
|
|
set_str = str(set_value).lower()
|
|
set_map = {
|
|
"1": "Set 1",
|
|
"2": "Set 2",
|
|
"3": "Set 3",
|
|
"e": "Encore",
|
|
"e2": "Encore 2",
|
|
"encore": "Encore",
|
|
}
|
|
return set_map.get(set_str, f"Set {set_value}")
|
|
|
|
|
|
def main():
|
|
"""Run Grateful Dead import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = GratefulDeadImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|