/
recording_lookup.py
75 lines (58 loc) · 2.1 KB
/
recording_lookup.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import sys
import uuid
from urllib.parse import quote
import requests
import ujson
from troi import Element, Artist, Recording
class RecordingLookupElement(Element):
'''
Look up a musicbrainz data for a list of recordings, based on MBID.
'''
SERVER_URL = "https://labs.api.listenbrainz.org/recording-mbid-lookup/json?count=%d"
@staticmethod
def inputs():
return [ Recording ]
@staticmethod
def outputs():
return [ Recording ]
def read(self, inputs, debug=False):
recordings = inputs[0]
if not recordings:
return []
data = []
r_mbids = ",".join([ r.mbid for r in recordings ])
for r in recordings:
data.append({ '[recording_mbid]': r.mbid })
if debug:
print("- debug %d recordings" % len(recordings))
r = requests.post(self.SERVER_URL % len(recordings), json=data)
if r.status_code != 200:
r.raise_for_status()
try:
rows = ujson.loads(r.text)
if debug:
print("- debug %d rows in response" % len(rows))
except Exception as err:
raise RuntimeError(str(err))
mbid_index = {}
for row in rows:
mbid_index[row['original_recording_mbid']] = row
for r in recordings:
try:
row = mbid_index[r.mbid]
except KeyError:
if debug:
print("- debug recording MBID %s not found, skipping." % r.mbid)
continue
if not r.artist:
a = Artist(name=row['artist_credit_name'],
mbids=row.get('[artist_credit_mbids]', []),
artist_credit_id=row['artist_credit_id'])
r.artist = a
else:
r.artist.name = row['artist_credit_name']
r.artist.mbids = row.get('[artist_credit_mbids]', []),
r.artist.artist_credit_id = row['artist_credit_id']
r.name = row['recording_name']
r.length = row['length']
return recordings