# /// script
# dependencies = [
# "graphinate[server]",
# "diskcache",
# "musicbrainzngs"
# ]
# ///
import logging
import operator
import pathlib
from functools import reduce
from time import sleep
import diskcache
import musicbrainzngs
import graphinate
from graphinate.tools import VariableNameManager
# logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def initialize_musicbrainz():
musicbrainzngs.set_useragent(
"MusicArtistGraph",
"0.1.0",
"https://github.com/erivlis/graphinate"
)
initialize_musicbrainz()
def cache_dir():
current_script_path = pathlib.Path(__file__).resolve()
parent_dir = current_script_path.parent
return (parent_dir / 'cache').as_posix()
def music_graph_model(name: str, max_depth: int = 0):
variable_name_manager = VariableNameManager()
graph_model = graphinate.model(f"{name.capitalize()} Graph")
artists_cache = diskcache.Cache(directory=cache_dir(), eviction_policy='none')
result = musicbrainzngs.search_artists(query=name, strict=True, artist=name)
sleep(1)
root_artist = result.get('artist-list', [])[0] if result else None
def artists(parent_artist, artist, relationship, depth):
logger.info(f"Current depth: {depth}")
artist_id = artist.get('id')
if artist_id not in artists_cache:
artists_cache[artist_id] = musicbrainzngs.get_artist_by_id(id=artist_id, includes=['artist-rels']).get(
'artist')
sleep(0.1)
artist = artists_cache.get(artist_id)
yield parent_artist, artist, relationship
if depth < max_depth:
related_artist_ids = set()
for item in artist.get('artist-relation-list', []):
related_artist = item.get('artist')
related_artist_id = related_artist.get('id')
related_artist_relationship = variable_name_manager.make_valid(item.get('type', '_unknown_'))
if related_artist_id not in related_artist_ids:
related_artist_ids.add(related_artist_id)
yield from artists(artist, related_artist, related_artist_relationship, depth + 1)
def get_type(value):
return value.get('type', '_UNKNOWN_')
@graph_model.node(get_type,
key=operator.itemgetter('id'),
label=operator.itemgetter('name'),
multiplicity=graphinate.Multiplicity.FIRST)
def node():
yielded = set()
for a, b, t in artists(None, root_artist, None, 0):
if a and ((a_id := a.get('id')) not in yielded):
yielded.add(a_id)
yield a
if b and ((b_id := b.get('id')) not in yielded):
yielded.add(b_id)
yield b
@graph_model.edge(get_type)
def edge():
for a, b, t in artists(None, root_artist, None, 0):
if a:
yield {'source': a.get('id'), 'target': b.get('id'), 'type': t}
return graph_model
if __name__ == '__main__':
from gui import ListboxChooser
artist_names = [
'Angine de Poitrine',
'Beatles',
'Brand X',
'Caravan',
'Charles Mingus',
'Chick Corea',
'Dave Brubeck',
'Dave Douglas',
'David Bowie',
'Deep Purple',
'Dire Straits',
'Emerson, Lake & Palmer',
'Foo Fighters',
'Frank Zappa',
'Genesis',
'Gentle Giant',
'Herbie Hancock',
'Jethro Tull',
'John Coltrane',
'John Scofield',
'John Zorn',
'Ken Vandermark',
'King Crimson',
'Led Zeppelin',
'Mahavishnu Orchestra',
'Miles Davis',
'Nirvana',
'Ornette Coleman',
'Paul McCartney',
'Pearl Jam',
'Pink Floyd',
'Police',
'Porcupine Tree',
'Radiohead',
'Red Hot Chili Peppers',
'Return to Forever',
'Rush',
'Smashing Pumpkins',
'Soft Machine',
'Soundgarden',
'Stone Temple Pilots',
'System of a Down',
'Thelonious Monk',
'Vienna Art Orchestra',
'Weather Report',
'Wings',
'Yes',
]
# artists_cache = diskcache.Cache(directory=cache_dir(), eviction_policy='none')
# artist_names = (artists_cache.get(i).get('name') for i in artists_cache)
listbox_chooser = ListboxChooser('Choose Artist/s', {name: name for name in artist_names})
models = (music_graph_model(a, 2) for _, a in listbox_chooser.get_choices())
model = reduce(operator.add, models)
schema = graphinate.builders.GraphQLBuilder(model).build()
graphinate.graphql.server(schema)