407 lines
13 KiB
Python
407 lines
13 KiB
Python
"""
|
|
Setlist.fm API Importer
|
|
|
|
Universal fallback importer for bands without dedicated APIs.
|
|
Used for: Dead & Company, Billy Strings, and as backup for others.
|
|
|
|
API Documentation: https://api.setlist.fm/docs/1.0/index.html
|
|
"""
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Dict, Optional, List
|
|
from sqlmodel import Session
|
|
|
|
from .base import ImporterBase
|
|
|
|
|
|
class SetlistFmImporter(ImporterBase):
|
|
"""Import data from Setlist.fm API"""
|
|
|
|
# Override these in subclasses for specific bands
|
|
ARTIST_MBID: str = "" # MusicBrainz ID
|
|
|
|
BASE_URL = "https://api.setlist.fm/rest/1.0"
|
|
|
|
def __init__(self, session: Session, api_key: Optional[str] = None):
|
|
super().__init__(session)
|
|
self.api_key = api_key or os.getenv("SETLISTFM_API_KEY")
|
|
if not self.api_key:
|
|
raise ValueError(
|
|
"Setlist.fm API key required. Set SETLISTFM_API_KEY environment variable "
|
|
"or pass api_key parameter. Get key at https://www.setlist.fm/settings/api"
|
|
)
|
|
|
|
def _api_get(self, endpoint: str, params: Optional[Dict] = None) -> Optional[Dict]:
|
|
"""Make API request with key"""
|
|
url = f"{self.BASE_URL}/{endpoint}"
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"x-api-key": self.api_key
|
|
}
|
|
|
|
return self.fetch_json(url, params, headers=headers)
|
|
|
|
def _fetch_all_setlists(self) -> List[Dict]:
|
|
"""Fetch all setlists for the artist with pagination"""
|
|
all_setlists = []
|
|
page = 1
|
|
|
|
while True:
|
|
print(f" Fetching page {page}...", end="", flush=True)
|
|
data = self._api_get(
|
|
f"artist/{self.ARTIST_MBID}/setlists",
|
|
params={"p": page}
|
|
)
|
|
|
|
if not data or "setlist" not in data:
|
|
print(" Done.")
|
|
break
|
|
|
|
setlists = data.get("setlist", [])
|
|
if not setlists:
|
|
print(" Done.")
|
|
break
|
|
|
|
all_setlists.extend(setlists)
|
|
print(f" ({len(setlists)} setlists)")
|
|
|
|
# Check if we've reached the last page
|
|
items_per_page = data.get("itemsPerPage", 20)
|
|
total = data.get("total", 0)
|
|
if page * items_per_page >= total:
|
|
break
|
|
|
|
page += 1
|
|
|
|
# Safety limit
|
|
if page > 200:
|
|
print(" (Safety limit reached)")
|
|
break
|
|
|
|
return all_setlists
|
|
|
|
def import_all(self):
|
|
"""Run full import"""
|
|
print("=" * 60)
|
|
print(f"SETLIST.FM IMPORTER: {self.VERTICAL_NAME}")
|
|
print("=" * 60)
|
|
|
|
# 1. Create/get vertical
|
|
self.get_or_create_vertical()
|
|
|
|
# 2. Fetch all setlists (includes venue, show, and song data)
|
|
print("\n📋 Fetching all setlists...")
|
|
setlists = self._fetch_all_setlists()
|
|
print(f" Found {len(setlists)} setlists")
|
|
|
|
# 3. Process setlists (imports venues, shows, songs, performances)
|
|
self._process_setlists(setlists)
|
|
|
|
print("\n" + "=" * 60)
|
|
print(f"✓ {self.VERTICAL_NAME} 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 _process_setlists(self, setlists: List[Dict]):
|
|
"""Process setlist data to extract and import all entities"""
|
|
print("\n🎤 Processing setlists...")
|
|
|
|
performance_count = 0
|
|
|
|
for setlist_data in setlists:
|
|
setlist_id = setlist_data.get("id")
|
|
event_date_str = setlist_data.get("eventDate")
|
|
|
|
if not event_date_str:
|
|
continue
|
|
|
|
try:
|
|
# Setlist.fm uses DD-MM-YYYY format
|
|
event_date = datetime.strptime(event_date_str, "%d-%m-%Y")
|
|
except ValueError:
|
|
continue
|
|
|
|
# Extract venue
|
|
venue_data = setlist_data.get("venue", {})
|
|
venue_name = venue_data.get("name", "Unknown Venue")
|
|
city_data = venue_data.get("city", {})
|
|
city = city_data.get("name", "Unknown")
|
|
state = city_data.get("stateCode")
|
|
country_data = city_data.get("country", {})
|
|
country = country_data.get("name", "Unknown")
|
|
|
|
venue_id = self.upsert_venue(
|
|
name=venue_name,
|
|
city=city,
|
|
state=state,
|
|
country=country,
|
|
external_id=venue_data.get("id")
|
|
)
|
|
|
|
# Handle tour if present
|
|
tour_id = None
|
|
tour_data = setlist_data.get("tour")
|
|
if tour_data:
|
|
tour_name = tour_data.get("name")
|
|
if tour_name:
|
|
tour_id = self.upsert_tour(
|
|
name=tour_name,
|
|
external_id=f"setlistfm_tour_{tour_name}"
|
|
)
|
|
|
|
# Create show
|
|
show_id = self.upsert_show(
|
|
date=event_date,
|
|
venue_id=venue_id,
|
|
tour_id=tour_id,
|
|
notes=setlist_data.get("info"),
|
|
external_id=setlist_id
|
|
)
|
|
|
|
# Process sets and songs
|
|
sets = setlist_data.get("sets", {}).get("set", [])
|
|
position = 0
|
|
|
|
for set_idx, set_data in enumerate(sets):
|
|
set_name = self._parse_set_name(set_data.get("name", str(set_idx + 1)), set_data.get("encore"))
|
|
|
|
songs = set_data.get("song", [])
|
|
for song_data in songs:
|
|
song_name = song_data.get("name")
|
|
if not song_name:
|
|
continue
|
|
|
|
position += 1
|
|
|
|
# Handle covers
|
|
cover_data = song_data.get("cover")
|
|
original_artist = cover_data.get("name") if cover_data else None
|
|
|
|
song_id = self.upsert_song(
|
|
title=song_name,
|
|
original_artist=original_artist,
|
|
external_id=f"setlistfm_song_{song_name}"
|
|
)
|
|
|
|
self.upsert_performance(
|
|
show_id=show_id,
|
|
song_id=song_id,
|
|
position=position,
|
|
set_name=set_name,
|
|
notes=song_data.get("info")
|
|
)
|
|
performance_count += 1
|
|
|
|
if len(self.show_map) % 50 == 0:
|
|
print(f" Progress: {len(self.show_map)} shows, {performance_count} performances...")
|
|
|
|
print(f"✓ Processed {len(self.show_map)} shows, {performance_count} performances")
|
|
|
|
def _parse_set_name(self, name: str, encore: Optional[int] = None) -> str:
|
|
"""Convert Setlist.fm set notation to display name"""
|
|
if encore:
|
|
return f"Encore {encore}" if encore > 1 else "Encore"
|
|
|
|
# Try to parse numeric set
|
|
try:
|
|
set_num = int(name)
|
|
return f"Set {set_num}"
|
|
except (ValueError, TypeError):
|
|
return name or "Set 1"
|
|
|
|
def import_venues(self) -> Dict[str, int]:
|
|
return self.venue_map
|
|
|
|
def import_songs(self) -> Dict[str, int]:
|
|
return self.song_map
|
|
|
|
def import_shows(self) -> Dict[str, int]:
|
|
return self.show_map
|
|
|
|
def import_setlists(self):
|
|
pass # Handled in import_all
|
|
|
|
|
|
class DeadAndCompanyImporter(SetlistFmImporter):
|
|
"""Import Dead & Company data from Setlist.fm"""
|
|
|
|
VERTICAL_NAME = "Dead & Company"
|
|
VERTICAL_SLUG = "dead-and-company"
|
|
VERTICAL_DESCRIPTION = "Dead & Company is a rock band formed in 2015 with Grateful Dead's Bob Weir, Mickey Hart, and Bill Kreutzmann joined by John Mayer, Oteil Burbridge, and Jeff Chimenti."
|
|
|
|
# Dead & Company MusicBrainz ID
|
|
ARTIST_MBID = "94f8947c-2d9c-4519-bcf9-6d11a24ad006"
|
|
|
|
|
|
class BillyStringsImporter(SetlistFmImporter):
|
|
"""Import Billy Strings data from Setlist.fm"""
|
|
|
|
VERTICAL_NAME = "Billy Strings"
|
|
VERTICAL_SLUG = "billy-strings"
|
|
VERTICAL_DESCRIPTION = "Billy Strings is an American bluegrass musician from Michigan, blending traditional bluegrass with progressive elements."
|
|
|
|
# Billy Strings MusicBrainz ID
|
|
ARTIST_MBID = "640db492-34c4-47df-be14-96e2cd4b9fe4"
|
|
|
|
|
|
class JoeRussosAlmostDeadImporter(SetlistFmImporter):
|
|
"""Import Joe Russo's Almost Dead data from Setlist.fm"""
|
|
|
|
VERTICAL_NAME = "Joe Russo's Almost Dead"
|
|
VERTICAL_SLUG = "jrad"
|
|
VERTICAL_DESCRIPTION = "Joe Russo's Almost Dead is an American rock band formed in 2013 that interprets the music of the Grateful Dead."
|
|
|
|
# JRAD MusicBrainz ID
|
|
ARTIST_MBID = "84a69823-3d4f-4ede-b43f-17f85513181a"
|
|
|
|
|
|
class EggyImporter(SetlistFmImporter):
|
|
"""Import Eggy data from Setlist.fm"""
|
|
|
|
VERTICAL_NAME = "Eggy"
|
|
VERTICAL_SLUG = "eggy"
|
|
VERTICAL_DESCRIPTION = "Connecticut jam band formed in 2014. Known for improvisational rock and explosive live shows."
|
|
|
|
# Eggy MusicBrainz ID
|
|
ARTIST_MBID = "ba0b9dc6-bd61-42c7-a28f-5179b1c04391"
|
|
|
|
|
|
class DogsInAPileImporter(SetlistFmImporter):
|
|
"""Import Dogs in a Pile data from Setlist.fm"""
|
|
|
|
VERTICAL_NAME = "Dogs in a Pile"
|
|
VERTICAL_SLUG = "dogs-in-a-pile"
|
|
VERTICAL_DESCRIPTION = "New Jersey jam band. Young and energetic."
|
|
|
|
# Dogs in a Pile MusicBrainz ID
|
|
ARTIST_MBID = "a05236ee-3fac-45d7-96f5-b2cd6d03fda9"
|
|
|
|
|
|
class TheDiscoBiscuitsImporter(SetlistFmImporter):
|
|
"""Import The Disco Biscuits data from Setlist.fm"""
|
|
|
|
VERTICAL_NAME = "The Disco Biscuits"
|
|
VERTICAL_SLUG = "disco-biscuits"
|
|
VERTICAL_DESCRIPTION = "Philadelphia trance-fusion jam band. Pioneers of livetronica."
|
|
|
|
# The Disco Biscuits MusicBrainz ID
|
|
ARTIST_MBID = "4e43632a-afef-4b54-a822-26311110d5c5"
|
|
|
|
|
|
class TheStringCheeseIncidentImporter(SetlistFmImporter):
|
|
"""Import The String Cheese Incident data from Setlist.fm"""
|
|
|
|
VERTICAL_NAME = "The String Cheese Incident"
|
|
VERTICAL_SLUG = "sci"
|
|
VERTICAL_DESCRIPTION = "Colorado jam band formed in 1993. Known for bluegrass-infused improvisational rock."
|
|
|
|
# SCI MusicBrainz ID
|
|
ARTIST_MBID = "cff95140-6d57-498a-8834-10eb72865b29"
|
|
|
|
|
|
class MindlessSelfIndulgenceImporter(SetlistFmImporter):
|
|
"""Import Mindless Self Indulgence data from Setlist.fm"""
|
|
|
|
VERTICAL_NAME = "Mindless Self Indulgence"
|
|
VERTICAL_SLUG = "msi"
|
|
VERTICAL_DESCRIPTION = "New York City electronic rock band. Known for their high-energy, chaotic style."
|
|
|
|
# MSI MusicBrainz ID
|
|
ARTIST_MBID = "44f42386-a733-4b51-8298-fe5c807d03aa"
|
|
|
|
|
|
def main_dead_and_company():
|
|
"""Run Dead & Company import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = DeadAndCompanyImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
def main_billy_strings():
|
|
"""Run Billy Strings import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = BillyStringsImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
def main_jrad():
|
|
"""Run JRAD import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = JoeRussosAlmostDeadImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
def main_eggy():
|
|
"""Run Eggy import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = EggyImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
def main_dogs():
|
|
"""Run Dogs in a Pile import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = DogsInAPileImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
def main_biscuits():
|
|
"""Run The Disco Biscuits import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = TheDiscoBiscuitsImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
def main_sci():
|
|
"""Run SCI import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = TheStringCheeseIncidentImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
def main_msi():
|
|
"""Run Mindless Self Indulgence import"""
|
|
from database import engine
|
|
|
|
with Session(engine) as session:
|
|
importer = MindlessSelfIndulgenceImporter(session)
|
|
importer.import_all()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
if len(sys.argv) > 1:
|
|
if sys.argv[1] == "deadco":
|
|
main_dead_and_company()
|
|
elif sys.argv[1] == "bmfs":
|
|
main_billy_strings()
|
|
elif sys.argv[1] == "jrad":
|
|
main_jrad()
|
|
elif sys.argv[1] == "eggy":
|
|
main_eggy()
|
|
elif sys.argv[1] == "dogs":
|
|
main_dogs()
|
|
elif sys.argv[1] == "biscuits":
|
|
main_biscuits()
|
|
elif sys.argv[1] == "sci":
|
|
main_sci()
|
|
elif sys.argv[1] == "msi":
|
|
main_msi()
|
|
else:
|
|
print("Usage: python -m importers.setlistfm [deadco|bmfs|jrad|eggy|dogs|biscuits|sci|msi]")
|