86 lines
3.1 KiB
Python
Executable File
86 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
from time import sleep
|
|
import json
|
|
import pylistenbrainz
|
|
import sys
|
|
|
|
|
|
def connect_client(auth_token):
|
|
try:
|
|
client = pylistenbrainz.ListenBrainz()
|
|
client.set_auth_token(auth_token)
|
|
except BaseException:
|
|
print("Unable to connect to ListenBrainz.")
|
|
return client
|
|
|
|
|
|
def open_export(export_file):
|
|
try:
|
|
with open(export_file, 'r') as read_file:
|
|
export_json = json.load(read_file)
|
|
except FileNotFoundError:
|
|
print("File does not exist.")
|
|
return export_json
|
|
|
|
|
|
def parse_listen(raw):
|
|
listen = pylistenbrainz.Listen(
|
|
track_name=raw['track_metadata']['track_name'],
|
|
artist_name=raw['track_metadata']['artist_name']
|
|
)
|
|
if 'listened_at' in raw:
|
|
listen.listened_at = raw['listened_at']
|
|
if 'user_name' in raw:
|
|
listen.username = raw['user_name']
|
|
if 'release_name' in raw['track_metadata']:
|
|
listen.release_name = raw['track_metadata']['release_name']
|
|
if 'additional_info' in raw['track_metadata']:
|
|
listen.additional_info = raw['track_metadata']['additional_info']
|
|
if 'tracknumber' in raw['track_metadata']['additional_info']:
|
|
listen.tracknumber = raw['track_metadata']['additional_info']['tracknumber']
|
|
if 'spotify_id' in raw['track_metadata']['additional_info']:
|
|
listen.spotify_id = raw['track_metadata']['additional_info']['spotify_id']
|
|
if 'isrc' in raw['track_metadata']['additional_info']:
|
|
listen.spotify_id = raw['track_metadata']['additional_info']['isrc']
|
|
if 'listening_from' in raw['track_metadata']['additional_info']:
|
|
listen.listening_from = raw['track_metadata']['additional_info']['listening_from']
|
|
if 'mbid_mapping' in raw['track_metadata']:
|
|
if 'recording_mbid' in raw['track_metadata']['mbid_mapping']:
|
|
listen.recording_mbid = raw['track_metadata']['mbid_mapping']['recording_mbid']
|
|
if 'artist_mbids' in raw['track_metadata']['mbid_mapping']:
|
|
listen.artist_mbids = raw['track_metadata']['mbid_mapping']['artist_mbids']
|
|
if 'release_mbid' in raw['track_metadata']['mbid_mapping']:
|
|
listen.release_mbid = raw['track_metadata']['mbid_mapping']['release_mbid']
|
|
return listen
|
|
|
|
|
|
def main():
|
|
client = connect_client(sys.argv[1])
|
|
exported = open_export(sys.argv[2])
|
|
print("Opened file with {} listens, starting import...".format(len(exported)))
|
|
imported = 0
|
|
for raw_listen in exported:
|
|
listen = parse_listen(raw_listen)
|
|
for x in range(0, 5):
|
|
try:
|
|
client.submit_single_listen(listen)
|
|
imported = imported + 1
|
|
print("Imported {} listens.".format(imported), end='\r')
|
|
submit_error = None
|
|
except BaseException:
|
|
submit_error = True
|
|
|
|
if submit_error:
|
|
sleep_time = x * 5
|
|
print(
|
|
"\nError occurred, retrying in {} seconds...".format(sleep_time))
|
|
sleep(sleep_time)
|
|
else:
|
|
break
|
|
print("\nSuccesfully imported {} listens.".format(imported))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|