Compare commits
8 commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d8929d3064 | ||
|
|
b2e582c566 | ||
|
|
e044b7ffd4 | ||
|
|
3744bd1c56 | ||
|
|
59a3be1108 | ||
|
|
22e19557d8 | ||
|
|
366860a1ad | ||
|
|
bb5453b59e |
10 changed files with 903 additions and 1 deletions
4
.gitignore
vendored
Normal file
4
.gitignore
vendored
Normal file
|
|
@ -0,0 +1,4 @@
|
||||||
|
prefetched*
|
||||||
|
__pycache__
|
||||||
|
.idea
|
||||||
|
secret
|
||||||
48
DataBase.py
Normal file
48
DataBase.py
Normal file
|
|
@ -0,0 +1,48 @@
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
|
||||||
|
data_dir = 'prefetched'
|
||||||
|
|
||||||
|
|
||||||
|
class SongTags:
|
||||||
|
def __init__(self, title='', artist='', album=''):
|
||||||
|
self.artist = artist
|
||||||
|
self.title = title
|
||||||
|
self.album = album
|
||||||
|
|
||||||
|
|
||||||
|
def get_data(name):
|
||||||
|
file_path = os.path.join(data_dir, f"{name}.json")
|
||||||
|
try:
|
||||||
|
with open(file_path, 'r') as file:
|
||||||
|
data = json.load(file)
|
||||||
|
return data
|
||||||
|
except FileNotFoundError:
|
||||||
|
return f"File {name}.json not found."
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
return f"Error decoding JSON from {name}.json."
|
||||||
|
|
||||||
|
|
||||||
|
def save_data(data, name):
|
||||||
|
def remove_available_markets(obj):
|
||||||
|
if isinstance(obj, dict):
|
||||||
|
return {key: remove_available_markets(value) for key, value in obj.items() if key != 'available_markets'}
|
||||||
|
elif isinstance(obj, list):
|
||||||
|
return [remove_available_markets(item) for item in obj]
|
||||||
|
else:
|
||||||
|
return obj
|
||||||
|
|
||||||
|
data = remove_available_markets(data)
|
||||||
|
|
||||||
|
os.makedirs(data_dir, exist_ok=True)
|
||||||
|
|
||||||
|
file_path = os.path.join(data_dir, f"{name}.json")
|
||||||
|
with open(file_path, 'w') as file:
|
||||||
|
json.dump(data, file, indent=4)
|
||||||
|
return f"Data saved to {name}.json."
|
||||||
|
|
||||||
|
|
||||||
|
def get_artists_str(artists_pack):
|
||||||
|
track_artists = [artist['name'] for artist in artists_pack]
|
||||||
|
artists = " ".join(track_artists)
|
||||||
|
return artists
|
||||||
137
LinkerPatternGenerator.py
Normal file
137
LinkerPatternGenerator.py
Normal file
|
|
@ -0,0 +1,137 @@
|
||||||
|
from DataBase import *
|
||||||
|
from SpotifyWebAPI import find_song, update_access_token
|
||||||
|
|
||||||
|
from fuzzywuzzy import fuzz
|
||||||
|
from tqdm import tqdm
|
||||||
|
|
||||||
|
|
||||||
|
def spotify_pattern_generator():
|
||||||
|
update_access_token()
|
||||||
|
|
||||||
|
local_library = get_data("local_library")
|
||||||
|
spotify_library = get_data("tracks")
|
||||||
|
|
||||||
|
def find_local_songs_on_spotify(local_lib):
|
||||||
|
total = len(local_library)
|
||||||
|
found_tracks_map = {}
|
||||||
|
with tqdm(total=total, desc='Searching local songs on spotify') as pbar:
|
||||||
|
for track_id, track in local_lib.items():
|
||||||
|
search_pattern = f"{track['name']} {track['artist']} {track['album']}"
|
||||||
|
found_tracks = find_song(search_pattern)
|
||||||
|
found_tracks_map[track_id] = found_tracks[0:1]
|
||||||
|
pbar.update(1)
|
||||||
|
return found_tracks_map
|
||||||
|
|
||||||
|
spotify_found_tracks = find_local_songs_on_spotify(local_library)
|
||||||
|
spotify_user_track = {track['track']['id']: {"local_mappings": []} for track in spotify_library}
|
||||||
|
|
||||||
|
spotify_pattern = {}
|
||||||
|
for local_id, found_items in spotify_found_tracks.items():
|
||||||
|
spotify_pattern[local_id] = {"score": 0.0, "items": []}
|
||||||
|
|
||||||
|
if not len(found_items):
|
||||||
|
continue
|
||||||
|
|
||||||
|
for found_item in found_items:
|
||||||
|
spotify_track_id = found_item['id']
|
||||||
|
if spotify_track_id in spotify_user_track:
|
||||||
|
spotify_pattern[local_id]['items'].append(spotify_track_id)
|
||||||
|
spotify_user_track[spotify_track_id]['local_mappings'].append(local_id)
|
||||||
|
|
||||||
|
for spotify_id, track_local_mappings in spotify_user_track.items():
|
||||||
|
score = len(track_local_mappings['local_mappings'])
|
||||||
|
for local_id in track_local_mappings['local_mappings']:
|
||||||
|
spotify_pattern[local_id]['score'] = 1 / score
|
||||||
|
|
||||||
|
save_data(spotify_pattern, "link_pattern_spotify")
|
||||||
|
|
||||||
|
|
||||||
|
def fuzzy_pattern_generator():
|
||||||
|
local_library = get_data("local_library")
|
||||||
|
track_descriptions = get_data("tracks")
|
||||||
|
|
||||||
|
def get_matched(src_pattern, patterns):
|
||||||
|
result = {}
|
||||||
|
for target_pattern in patterns:
|
||||||
|
similarity_score = fuzz.token_set_ratio(src_pattern, target_pattern)
|
||||||
|
result[target_pattern] = similarity_score / 100
|
||||||
|
|
||||||
|
sorted_results = sorted(result.items(), key=lambda item: item[1], reverse=True)
|
||||||
|
return sorted_results
|
||||||
|
|
||||||
|
def resolve_tracks(s_patterns, t_patterns):
|
||||||
|
|
||||||
|
pattern_map = {}
|
||||||
|
|
||||||
|
total = len(s_patterns)
|
||||||
|
with tqdm(total=total, desc='Progress') as pbar:
|
||||||
|
for s_pattern in s_patterns:
|
||||||
|
res = get_matched(s_pattern, t_patterns)
|
||||||
|
matched = {"score": 0, "items": []}
|
||||||
|
if len(res):
|
||||||
|
matched['items'] = res[0:min(len(res), 4)]
|
||||||
|
matched['score'] = res[0][1]
|
||||||
|
pattern_map[s_pattern] = matched
|
||||||
|
pbar.update(1)
|
||||||
|
|
||||||
|
return pattern_map
|
||||||
|
|
||||||
|
source_patterns = []
|
||||||
|
for track in track_descriptions:
|
||||||
|
artists = get_artists_str(track['track']['artists'])
|
||||||
|
pattern = f"{track['track']['name']} {artists}"
|
||||||
|
source_patterns.append(pattern)
|
||||||
|
|
||||||
|
target_patterns = [local['path'] for _, local in local_library.items()]
|
||||||
|
|
||||||
|
fuzzy_pattern = resolve_tracks(source_patterns, target_patterns)
|
||||||
|
save_data(fuzzy_pattern, "link_pattern_fuzzy")
|
||||||
|
|
||||||
|
|
||||||
|
def fuzzy_tag_pattern_generator():
|
||||||
|
local_library = get_data("local_library")
|
||||||
|
spotify_library = get_data("tracks")
|
||||||
|
mappings = {}
|
||||||
|
|
||||||
|
def fuzzy_tags_ratio(first: SongTags, second: SongTags):
|
||||||
|
title_ratio = fuzz.token_set_ratio(first.title, second.title) / 100
|
||||||
|
artist_ratio = fuzz.token_set_ratio(first.artist, second.artist) / 100
|
||||||
|
album_ratio = fuzz.token_set_ratio(first.album, second.album) / 100
|
||||||
|
return title_ratio, artist_ratio, album_ratio
|
||||||
|
|
||||||
|
with tqdm(total=len(local_library), desc='Searching local songs on spotify') as pbar:
|
||||||
|
for local_track_id, local_track in local_library.items():
|
||||||
|
mapping = mappings.get(local_track_id, {"score": 1.0, "items": []})
|
||||||
|
ratios = []
|
||||||
|
for remote_track in spotify_library:
|
||||||
|
track = remote_track['track']
|
||||||
|
|
||||||
|
tag1 = SongTags(local_track['name'], local_track['artist'], local_track['album'])
|
||||||
|
tag2 = SongTags(track['name'], get_artists_str(track['artists']), track['album']['name'])
|
||||||
|
|
||||||
|
ratios.append((fuzzy_tags_ratio(tag1, tag2), track['id']))
|
||||||
|
|
||||||
|
filtered_ratios = []
|
||||||
|
threshold = 0.8
|
||||||
|
for ratio in ratios:
|
||||||
|
if all([val > threshold for val in ratio[0]]):
|
||||||
|
filtered_ratios.append((sum(i for i in ratio[0]) / len(ratio[0]), ratio[1]))
|
||||||
|
|
||||||
|
filtered_ratios.sort(reverse=True, key=lambda x: x[0])
|
||||||
|
if len(filtered_ratios):
|
||||||
|
mapping['items'] = [filtered_ratios[0][1]]
|
||||||
|
|
||||||
|
mappings[local_track_id] = mapping
|
||||||
|
pbar.update(1)
|
||||||
|
|
||||||
|
save_data(mappings, "link_pattern_tags")
|
||||||
|
|
||||||
|
|
||||||
|
def run_pattern_generators():
|
||||||
|
#fuzzy_pattern_generator()
|
||||||
|
#spotify_pattern_generator()
|
||||||
|
fuzzy_tag_pattern_generator()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
run_pattern_generators()
|
||||||
73
LocalLibraryIndexer.py
Normal file
73
LocalLibraryIndexer.py
Normal file
|
|
@ -0,0 +1,73 @@
|
||||||
|
import os
|
||||||
|
import fnmatch
|
||||||
|
|
||||||
|
import mutagen
|
||||||
|
|
||||||
|
from DataBase import *
|
||||||
|
import eyed3
|
||||||
|
from mutagen.flac import FLAC
|
||||||
|
|
||||||
|
|
||||||
|
music_extensions = ('*.mp3', '*.flac', '*.wav', '*.aac', '*.ogg', '*.m4a')
|
||||||
|
exclude_directories = ("*mary--*", "*example-word*")
|
||||||
|
|
||||||
|
|
||||||
|
def update_local_library(root_dir):
|
||||||
|
old_library = get_data("local_library")
|
||||||
|
known_paths = {track['path']: track_id for track_id, track in old_library.items()}
|
||||||
|
new_library = {}
|
||||||
|
|
||||||
|
song_id = '0'
|
||||||
|
|
||||||
|
def get_new_song_id(prev_id, song_path):
|
||||||
|
if song_path in known_paths:
|
||||||
|
return int(known_paths[song_path])
|
||||||
|
while (prev_id in old_library) or (prev_id in new_library):
|
||||||
|
prev_id = str(int(prev_id) + 1)
|
||||||
|
return prev_id
|
||||||
|
|
||||||
|
for dir_path, dir_names, filenames in os.walk(root_dir):
|
||||||
|
dir_names[:] = [d for d in dir_names if not any(fnmatch.fnmatch(d, exclude) for exclude in exclude_directories)]
|
||||||
|
filtered_filenames = [filename for filename in filenames if any(fnmatch.fnmatch(filename, ext) for ext in music_extensions)]
|
||||||
|
|
||||||
|
for filename in filtered_filenames:
|
||||||
|
song_name, track_type = os.path.splitext(filename)
|
||||||
|
relative_path = os.path.relpath(os.path.join(dir_path, filename), root_dir)
|
||||||
|
song_id = get_new_song_id(song_id, relative_path)
|
||||||
|
new_library[song_id] = {
|
||||||
|
"name": song_name,
|
||||||
|
"path": relative_path,
|
||||||
|
"type": track_type,
|
||||||
|
"artist": "",
|
||||||
|
"album": ""
|
||||||
|
}
|
||||||
|
|
||||||
|
for track_id, track in new_library.items():
|
||||||
|
abs_path = os.path.join(root_dir, track['path'])
|
||||||
|
match track['type']:
|
||||||
|
case ".mp3":
|
||||||
|
tag = eyed3.load(abs_path).tag
|
||||||
|
if not tag:
|
||||||
|
print(f"Invalid mp3 header {abs_path}")
|
||||||
|
else:
|
||||||
|
track['artist'] = tag.artist
|
||||||
|
track['name'] = tag.title
|
||||||
|
track['album'] = tag.album
|
||||||
|
|
||||||
|
case ".flac":
|
||||||
|
try:
|
||||||
|
metadata = FLAC(abs_path).tags
|
||||||
|
if 'artist' in metadata:
|
||||||
|
track['artist'] = " ".join(metadata['artist'])
|
||||||
|
if 'TITLE' in metadata:
|
||||||
|
track['name'] = " ".join(metadata['TITLE'])
|
||||||
|
if 'album' in metadata:
|
||||||
|
track['album'] = " ".join(metadata['album'])
|
||||||
|
except mutagen.MutagenError:
|
||||||
|
print(f"Invalid flac header {abs_path}")
|
||||||
|
|
||||||
|
save_data(new_library, "local_library")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
update_local_library("/mnt/main/data/music/raw")
|
||||||
101
ParseItunesToJson.py
Executable file
101
ParseItunesToJson.py
Executable file
|
|
@ -0,0 +1,101 @@
|
||||||
|
import xmltodict
|
||||||
|
import json
|
||||||
|
|
||||||
|
|
||||||
|
def flatten_dict(d):
|
||||||
|
|
||||||
|
out = []
|
||||||
|
for song in d:
|
||||||
|
newSong = {}
|
||||||
|
|
||||||
|
strCount = 0
|
||||||
|
intCount = 0
|
||||||
|
dateCount = 0
|
||||||
|
|
||||||
|
def getStr(id):
|
||||||
|
nonlocal strCount
|
||||||
|
if id not in song["key"]:
|
||||||
|
return "undef"
|
||||||
|
if len(song["string"]) <= strCount:
|
||||||
|
return "error"
|
||||||
|
strCount += 1
|
||||||
|
return song["string"][strCount - 1]
|
||||||
|
|
||||||
|
def getInt(id):
|
||||||
|
nonlocal intCount
|
||||||
|
if id not in song["key"]:
|
||||||
|
return -1
|
||||||
|
if len(song["integer"]) <= intCount:
|
||||||
|
return -1
|
||||||
|
intCount += 1
|
||||||
|
return song["integer"][intCount - 1]
|
||||||
|
|
||||||
|
def getDate(id):
|
||||||
|
nonlocal dateCount
|
||||||
|
if id not in song["key"]:
|
||||||
|
return "undef"
|
||||||
|
if len(song["date"]) <= dateCount:
|
||||||
|
return "error"
|
||||||
|
dateCount += 1
|
||||||
|
return song["date"][dateCount - 1]
|
||||||
|
|
||||||
|
newSong["Track ID"] = getInt("Track ID")
|
||||||
|
newSong["Name"] = getStr("Name")
|
||||||
|
newSong["Artist"] = getStr("Artist")
|
||||||
|
newSong["Album Artist"] = getStr("Album Artist")
|
||||||
|
newSong["Composer"] = getStr("Composer")
|
||||||
|
newSong["Album"] = getStr("Album")
|
||||||
|
newSong["Genre"] = getStr("Genre")
|
||||||
|
newSong["Kind"] = getStr("Kind")
|
||||||
|
newSong["Size"] = getInt("Size")
|
||||||
|
newSong["Total Time"] = getInt("Total Time")
|
||||||
|
newSong["Disc Number"] = getInt("Disc Number")
|
||||||
|
newSong["Disc Count"] = getInt("Disc Count")
|
||||||
|
newSong["Track Number"] = getInt("Track Number")
|
||||||
|
newSong["Track Count"] = getInt("Track Count")
|
||||||
|
newSong["Year"] = getInt("Year")
|
||||||
|
newSong["Date Modified"] = getDate("Date Modified")
|
||||||
|
newSong["Date Added"] = getDate("Date Added")
|
||||||
|
newSong["Bit Rate"] = getInt("Bit Rate")
|
||||||
|
newSong["Sample Rate"] = getInt("Sample Rate")
|
||||||
|
newSong["Play Count"] = getInt("Play Count")
|
||||||
|
newSong["Play Date"] = getInt("Play Date")
|
||||||
|
newSong["Play Date UTC"] = getDate("Play Date UTC")
|
||||||
|
newSong["Skip Count"] = getInt("Skip Count")
|
||||||
|
newSong["Skip Date"] = getDate("Skip Date")
|
||||||
|
newSong["Release Date"] = getDate("Release Date")
|
||||||
|
newSong["Album Rating"] = getInt("Album Rating")
|
||||||
|
newSong["Album Rating Computed"] = "Album Rating Computed" in song["key"]
|
||||||
|
newSong["Loved"] = "Loved" in song["key"]
|
||||||
|
newSong["Album Loved"] = "Album Loved" in song["key"]
|
||||||
|
newSong["Explicit"] = "Explicit" in song["key"]
|
||||||
|
newSong["Compilation"] = "Compilation" in song["key"]
|
||||||
|
newSong["Artwork Count"] = getInt("Artwork Count")
|
||||||
|
newSong["Sort Album"] = getStr("Sort Album")
|
||||||
|
newSong["Sort Artist"] = getStr("Sort Artist")
|
||||||
|
newSong["Sort Name"] = getStr("Sort Name")
|
||||||
|
newSong["Persistent ID"] = getStr("Persistent ID")
|
||||||
|
newSong["Track Type"] = getStr("Track Type")
|
||||||
|
|
||||||
|
out.append(newSong)
|
||||||
|
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def convert(filename, out_path):
|
||||||
|
with open(filename, 'r', encoding='utf-8') as xml_file:
|
||||||
|
data_dict = xmltodict.parse(xml_file.read())
|
||||||
|
|
||||||
|
# Extract the "Tracks" dictionary to be flattened
|
||||||
|
tracks_dict = data_dict['plist']['dict']['dict']
|
||||||
|
|
||||||
|
flat_tracks_dict = flatten_dict(tracks_dict["dict"])
|
||||||
|
|
||||||
|
json_data = json.dumps(flat_tracks_dict, indent=2)
|
||||||
|
|
||||||
|
with open(out_path, 'w', encoding='utf-8') as json_file:
|
||||||
|
json_file.write(json_data)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
convert('./prefetched/ItunesLibrary.xml', './prefetched/ItunesLibrary.json')
|
||||||
|
|
@ -1,2 +1,3 @@
|
||||||
# spotifyFetcher
|
# spotifyFetcher
|
||||||
loads all data through web api
|
loads all data through web api<br>
|
||||||
|
client secret - ***REDACTED***
|
||||||
|
|
|
||||||
122
SpotifyAuthenticator.py
Normal file
122
SpotifyAuthenticator.py
Normal file
|
|
@ -0,0 +1,122 @@
|
||||||
|
import base64
|
||||||
|
|
||||||
|
import requests
|
||||||
|
import webbrowser
|
||||||
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||||
|
import urllib.parse
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
import os.path
|
||||||
|
|
||||||
|
client_id = '***REDACTED***'
|
||||||
|
client_secret = None
|
||||||
|
token = None
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_client_secret():
|
||||||
|
global client_secret
|
||||||
|
if os.path.exists('secret'):
|
||||||
|
with open('secret', 'r') as file:
|
||||||
|
client_secret = file.readline().strip()
|
||||||
|
else:
|
||||||
|
client_secret = input("Enter the application secret: ")
|
||||||
|
with open('secret', 'w') as file:
|
||||||
|
file.write(client_secret)
|
||||||
|
|
||||||
|
|
||||||
|
def retrieve_web_api_token(authorization_code):
|
||||||
|
global token
|
||||||
|
|
||||||
|
token_url = 'https://accounts.spotify.com/api/token'
|
||||||
|
|
||||||
|
token_data = {
|
||||||
|
"grant_type": "authorization_code",
|
||||||
|
"client_secret": client_secret,
|
||||||
|
"client_id": client_id,
|
||||||
|
"code": authorization_code,
|
||||||
|
'redirect_uri': 'http://localhost:8888',
|
||||||
|
}
|
||||||
|
|
||||||
|
print("\nAuthenticating WEB API\n")
|
||||||
|
|
||||||
|
auth_string = f'{client_id}:{client_secret}'
|
||||||
|
b64_auth_string = base64.b64encode(auth_string.encode()).decode()
|
||||||
|
|
||||||
|
# Define the headers
|
||||||
|
headers = {
|
||||||
|
'Content-Type': 'application/x-www-form-urlencoded',
|
||||||
|
'Authorization': f'Basic {b64_auth_string}'
|
||||||
|
}
|
||||||
|
|
||||||
|
token_response = requests.post(token_url, data=token_data, headers=headers)
|
||||||
|
|
||||||
|
if token_response.status_code not in range(200, 299):
|
||||||
|
print(f"Failed to retrieve access token: {token_response.status_code}")
|
||||||
|
print(token_response.json())
|
||||||
|
raise ValueError
|
||||||
|
|
||||||
|
token = token_response.json()['access_token']
|
||||||
|
|
||||||
|
|
||||||
|
class RequestHandler(BaseHTTPRequestHandler):
|
||||||
|
def do_GET(self):
|
||||||
|
|
||||||
|
parsed_url = urllib.parse.urlparse(self.path)
|
||||||
|
query_params = urllib.parse.parse_qs(parsed_url.query)
|
||||||
|
|
||||||
|
if 'code' in query_params:
|
||||||
|
authorization_code = query_params['code'][0]
|
||||||
|
|
||||||
|
self.send_response(200)
|
||||||
|
self.send_header('Content-type', 'text/html')
|
||||||
|
self.end_headers()
|
||||||
|
self.wfile.write(b'<html><body>Authorization code received. You can close this window.</body></html>')
|
||||||
|
|
||||||
|
retrieve_web_api_token(authorization_code)
|
||||||
|
|
||||||
|
threading.Thread(target=self.server.shutdown).start()
|
||||||
|
|
||||||
|
else:
|
||||||
|
self.send_response(400)
|
||||||
|
self.send_header('Content-type', 'text/html')
|
||||||
|
self.end_headers()
|
||||||
|
self.wfile.write(b'<html><body>Authorization code not found.</body></html>')
|
||||||
|
|
||||||
|
|
||||||
|
def run_redirect_server():
|
||||||
|
httpd = HTTPServer(('localhost', 8888), RequestHandler)
|
||||||
|
print('Starting HTTP redirection server')
|
||||||
|
httpd.serve_forever()
|
||||||
|
|
||||||
|
|
||||||
|
def open_authenticate_page():
|
||||||
|
authorize_url = 'https://accounts.spotify.com/authorize'
|
||||||
|
|
||||||
|
params = {
|
||||||
|
'client_id': client_id,
|
||||||
|
'response_type': 'code',
|
||||||
|
'redirect_uri': 'http://localhost:8888',
|
||||||
|
"scope": "user-top-read playlist-read-collaborative user-read-email user-library-read user-read-private playlist-read-private",
|
||||||
|
}
|
||||||
|
|
||||||
|
auth_url = f"{authorize_url}?client_id={params['client_id']}&response_type={params['response_type']}&redirect_uri={params['redirect_uri']}&scope={params['scope']}"
|
||||||
|
webbrowser.open(auth_url)
|
||||||
|
|
||||||
|
|
||||||
|
def authenticate():
|
||||||
|
resolve_client_secret()
|
||||||
|
|
||||||
|
redirect_server = threading.Thread(target=run_redirect_server)
|
||||||
|
redirect_server.start()
|
||||||
|
|
||||||
|
time.sleep(1)
|
||||||
|
open_authenticate_page()
|
||||||
|
|
||||||
|
redirect_server.join()
|
||||||
|
|
||||||
|
return token
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
authenticate()
|
||||||
|
|
||||||
190
SpotifyWebAPI.py
Normal file
190
SpotifyWebAPI.py
Normal file
|
|
@ -0,0 +1,190 @@
|
||||||
|
import requests
|
||||||
|
from PIL import Image
|
||||||
|
from io import BytesIO
|
||||||
|
from SpotifyAuthenticator import authenticate
|
||||||
|
from DataBase import *
|
||||||
|
|
||||||
|
FETCH_STEP = 50
|
||||||
|
token = ''
|
||||||
|
|
||||||
|
|
||||||
|
def fetch(endpoint, method='GET', body=None):
|
||||||
|
url = f'https://api.spotify.com/{endpoint}'
|
||||||
|
headers = {
|
||||||
|
'Authorization': f'Bearer {token}',
|
||||||
|
}
|
||||||
|
|
||||||
|
response = requests.request(method, url, headers=headers, json=body)
|
||||||
|
response.raise_for_status()
|
||||||
|
return response.json()
|
||||||
|
|
||||||
|
|
||||||
|
def find_song(song_pattern):
|
||||||
|
return fetch(f"v1/search/?type=track&track=1&q={song_pattern}")['tracks']['items']
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_playlists(user_id):
|
||||||
|
print("Fetching playlists")
|
||||||
|
endpoint = f"v1/me/playlists"
|
||||||
|
pl_total = fetch(f'{endpoint}?limit=1')['total']
|
||||||
|
pl_count = 0
|
||||||
|
playlists = []
|
||||||
|
|
||||||
|
while pl_count < pl_total:
|
||||||
|
res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={pl_count}')
|
||||||
|
playlists += res['items']
|
||||||
|
pl_count += FETCH_STEP
|
||||||
|
|
||||||
|
return playlists
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_playlist_items(playlist_id):
|
||||||
|
endpoint = f"v1/playlists/{playlist_id}/tracks"
|
||||||
|
items = fetch(f'{endpoint}?fields=total')['total']
|
||||||
|
loaded = 0
|
||||||
|
tracks = []
|
||||||
|
|
||||||
|
while loaded < items:
|
||||||
|
res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={loaded}')
|
||||||
|
|
||||||
|
tracks += res['items']
|
||||||
|
loaded += FETCH_STEP
|
||||||
|
|
||||||
|
return tracks
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_tracks(user_id):
|
||||||
|
print("Fetching tracks")
|
||||||
|
endpoint = f"v1/me/tracks"
|
||||||
|
total = fetch(f"{endpoint}?limit=1&market=ES")['total']
|
||||||
|
current = 0
|
||||||
|
tracks = []
|
||||||
|
|
||||||
|
while total > current:
|
||||||
|
res = fetch(f"{endpoint}?limit={FETCH_STEP}&offset={current}&market=ES")
|
||||||
|
tracks += res['items']
|
||||||
|
current += FETCH_STEP
|
||||||
|
|
||||||
|
save_data(tracks, "tracks")
|
||||||
|
return tracks
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_tracks_top():
|
||||||
|
print("Fetching top tracks")
|
||||||
|
total = fetch(f'v1/me/top/tracks?fields=total')['total']
|
||||||
|
current = 0
|
||||||
|
tracks = []
|
||||||
|
|
||||||
|
while current < total:
|
||||||
|
res = fetch(f'v1/me/top/tracks?limit={FETCH_STEP}&offset={current}&time_range=long_term')
|
||||||
|
tracks += res['items']
|
||||||
|
current += FETCH_STEP
|
||||||
|
|
||||||
|
save_data(tracks, "top_tracks")
|
||||||
|
return tracks
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_artists_top():
|
||||||
|
print("Fetching top artists")
|
||||||
|
|
||||||
|
total = fetch(f'v1/me/top/artists?fields=total')['total']
|
||||||
|
current = 0
|
||||||
|
artists = []
|
||||||
|
|
||||||
|
while current < total:
|
||||||
|
res = fetch(f'v1/me/top/artists?limit={FETCH_STEP}&offset={current}&time_range=long_term')
|
||||||
|
artists += res['items']
|
||||||
|
current += FETCH_STEP
|
||||||
|
|
||||||
|
save_data(artists, "top_artists")
|
||||||
|
return artists
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_all_playlist_data(user_id):
|
||||||
|
playlists = fetch_playlists(user_id)
|
||||||
|
|
||||||
|
save_data(playlists, 'playlists')
|
||||||
|
|
||||||
|
print(len(playlists))
|
||||||
|
|
||||||
|
playlist_items = {}
|
||||||
|
|
||||||
|
print("Fetching playlists items")
|
||||||
|
|
||||||
|
index = 0
|
||||||
|
for pl in playlists:
|
||||||
|
pl_id = pl['id']
|
||||||
|
name = pl['name']
|
||||||
|
playlist_items[name] = fetch_playlist_items(pl_id)
|
||||||
|
print(f"{index} - {len(playlist_items[name])}")
|
||||||
|
index += 1
|
||||||
|
|
||||||
|
save_data(playlist_items, "playlistTracks")
|
||||||
|
|
||||||
|
|
||||||
|
def get_user_data():
|
||||||
|
print("Fetching user data")
|
||||||
|
|
||||||
|
user_profile = fetch(f'v1/me')
|
||||||
|
user_id = user_profile['id']
|
||||||
|
user_data = fetch(f'v1/users/{user_id}')
|
||||||
|
|
||||||
|
save_data(user_profile, "user_profile")
|
||||||
|
save_data(user_data, "user_data")
|
||||||
|
return user_id
|
||||||
|
|
||||||
|
|
||||||
|
def save_image_from_url(url, name, image_dir="playlist_covers"):
|
||||||
|
response = requests.get(url)
|
||||||
|
|
||||||
|
if response.status_code != 200:
|
||||||
|
raise "Cannot fetch the playlist cover"
|
||||||
|
|
||||||
|
image = Image.open(BytesIO(response.content))
|
||||||
|
|
||||||
|
directory = os.path.join(data_dir, image_dir)
|
||||||
|
os.makedirs(directory, exist_ok=True)
|
||||||
|
file_path = os.path.join(directory, f"{name}.jpg")
|
||||||
|
print(f"Image saved {file_path}")
|
||||||
|
image.save(file_path)
|
||||||
|
|
||||||
|
|
||||||
|
def load_playlist_covers():
|
||||||
|
playlists = get_data('playlists')
|
||||||
|
|
||||||
|
for pl in playlists:
|
||||||
|
if len(pl['images']):
|
||||||
|
url = pl['images'][0]['url']
|
||||||
|
save_image_from_url(url, pl['name'])
|
||||||
|
|
||||||
|
|
||||||
|
def load_user_cover():
|
||||||
|
user_data = get_data("user_data")
|
||||||
|
url = user_data['images'][1]['url']
|
||||||
|
save_image_from_url(url, "user", ".")
|
||||||
|
|
||||||
|
|
||||||
|
def load_artworks():
|
||||||
|
load_user_cover()
|
||||||
|
load_playlist_covers()
|
||||||
|
|
||||||
|
|
||||||
|
def update_access_token():
|
||||||
|
global token
|
||||||
|
token = authenticate()
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_data():
|
||||||
|
user_id = get_user_data()
|
||||||
|
|
||||||
|
fetch_tracks(user_id)
|
||||||
|
fetch_all_playlist_data(user_id)
|
||||||
|
fetch_tracks_top()
|
||||||
|
fetch_artists_top()
|
||||||
|
|
||||||
|
load_artworks()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
update_access_token()
|
||||||
|
fetch_data()
|
||||||
36
env
Executable file
36
env
Executable file
|
|
@ -0,0 +1,36 @@
|
||||||
|
|
||||||
|
PS1=" > "
|
||||||
|
|
||||||
|
#export PATH="/usr/local/bin/:$PATH"
|
||||||
|
export PATH="$HOME/bin/scripts:$HOME/bin/:$PATH"
|
||||||
|
#export PATH="$HOME/home/auser/.local/bin:$PATH"
|
||||||
|
|
||||||
|
export SIP="185.238.170.251"
|
||||||
|
|
||||||
|
alias v=nvim
|
||||||
|
alias ll="ls -l -a"
|
||||||
|
alias gl="git log --oneline"
|
||||||
|
alias ssh_server="ssh auser@185.238.170.251"
|
||||||
|
|
||||||
|
#eval "$(zoxide init bash --cmd zd)"
|
||||||
|
|
||||||
|
#source $PMAIN/.scripts/bashmarks.sh
|
||||||
|
|
||||||
|
fe() {
|
||||||
|
local result=$(command tere "$@")
|
||||||
|
[ -n "$result" ] && cd -- "$result"
|
||||||
|
}
|
||||||
|
|
||||||
|
vim_configure() {
|
||||||
|
pwd=$(pwd)
|
||||||
|
cd ~/.config/nvim/
|
||||||
|
nvim
|
||||||
|
cd $pwd
|
||||||
|
}
|
||||||
|
|
||||||
|
pyenv() {
|
||||||
|
source ~/bin/python/env311/bin/activate
|
||||||
|
}
|
||||||
|
|
||||||
|
source ~/src/scripts/remotes.sh
|
||||||
|
source /usr/share/fzf/key-bindings.bash
|
||||||
190
main.py
Normal file
190
main.py
Normal file
|
|
@ -0,0 +1,190 @@
|
||||||
|
from DataBase import *
|
||||||
|
from SpotifyWebAPI import fetch_data, update_access_token
|
||||||
|
import cmd
|
||||||
|
|
||||||
|
playlists = get_data('playlistTracks')
|
||||||
|
top_artists = get_data('top_artists')
|
||||||
|
top_tracks = get_data('top_tracks')
|
||||||
|
user_tracks = get_data('tracks')
|
||||||
|
link_patterns = [get_data('link_pattern_tags'), get_data('link_pattern_tags'), get_data('link_pattern_tags')]
|
||||||
|
local_library = get_data("local_library")
|
||||||
|
user_tracks_by_id = {track['track']['id']: track['track'] for track in user_tracks}
|
||||||
|
|
||||||
|
|
||||||
|
def get_playlist_names(pl):
|
||||||
|
return [name for name, items in pl.items()]
|
||||||
|
|
||||||
|
|
||||||
|
def print_playlist_tracks(pl_id: int):
|
||||||
|
pls = list(playlists)
|
||||||
|
name = pls[pl_id]
|
||||||
|
pl = playlists[name]
|
||||||
|
print(name)
|
||||||
|
for track in pl:
|
||||||
|
track_data = track['track']
|
||||||
|
track_name = track_data['name']
|
||||||
|
artists = get_artists_str(track_data['artists'])
|
||||||
|
print(f" '{track_name}' - '{artists}' ")
|
||||||
|
|
||||||
|
|
||||||
|
def print_top_artists():
|
||||||
|
for artist in top_artists:
|
||||||
|
name = artist['name']
|
||||||
|
print(f" '{name}'")
|
||||||
|
|
||||||
|
|
||||||
|
def print_top_tracks():
|
||||||
|
for track in top_tracks:
|
||||||
|
track_name = track['name']
|
||||||
|
artists = get_artists_str(track['artists'])
|
||||||
|
print(f" '{track_name}' - '{artists}' ")
|
||||||
|
|
||||||
|
|
||||||
|
def print_stats():
|
||||||
|
print(f" tracks - {len(user_tracks)}")
|
||||||
|
print(f" playlists - {len(playlists)}")
|
||||||
|
|
||||||
|
for name, tracks in playlists.items():
|
||||||
|
print(f" '{name}' - {len(tracks)}")
|
||||||
|
|
||||||
|
print(f" top tracks - {len(top_tracks)}")
|
||||||
|
print(f" top artists - {len(top_artists)}")
|
||||||
|
|
||||||
|
|
||||||
|
def print_tracks():
|
||||||
|
track_idx = 0
|
||||||
|
for trackItem in user_tracks:
|
||||||
|
track = trackItem['track']
|
||||||
|
print(f" {track_idx}: '{track['name']}' by '{get_artists_str(track['artists'])}'")
|
||||||
|
track_idx += 1
|
||||||
|
|
||||||
|
|
||||||
|
def sort_remote_tracks_by_popularity(track_ids):
|
||||||
|
out = []
|
||||||
|
for top_track in top_tracks:
|
||||||
|
if top_track['id'] in track_ids:
|
||||||
|
out.append(top_track['id'])
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def print_link_stats(link_pattern):
|
||||||
|
resolved_reversed = {}
|
||||||
|
resolved = []
|
||||||
|
unresolved_local = []
|
||||||
|
unresolved_remote = []
|
||||||
|
|
||||||
|
for local_id, links in link_pattern.items():
|
||||||
|
if len(links['items']):
|
||||||
|
resolved.append(local_id)
|
||||||
|
resolved_reversed[links['items'][0]] = local_id
|
||||||
|
else:
|
||||||
|
unresolved_local.append(local_id)
|
||||||
|
|
||||||
|
for track in user_tracks:
|
||||||
|
track_id = track['track']['id']
|
||||||
|
if track_id not in resolved_reversed:
|
||||||
|
unresolved_remote.append(track_id)
|
||||||
|
|
||||||
|
print(f"Local Tracks: {len(local_library)}")
|
||||||
|
print(f"Remote Tracks: {len(user_tracks)}")
|
||||||
|
|
||||||
|
unresolved_remote = sort_remote_tracks_by_popularity(unresolved_remote)
|
||||||
|
unresolved_remote = unresolved_remote[0:min(len(unresolved_remote), 100)]
|
||||||
|
|
||||||
|
print(f"\nUnresolved tracks from the remote: {len(unresolved_remote)}")
|
||||||
|
index = 0
|
||||||
|
for remote_id in unresolved_remote:
|
||||||
|
remote_track = user_tracks_by_id[remote_id]
|
||||||
|
print(f" {index}: '{remote_track['name']}' by '{get_artists_str(remote_track['artists'])}'")
|
||||||
|
index += 1
|
||||||
|
|
||||||
|
print(f"\nResolved tracks: {len(resolved)}")
|
||||||
|
index = 0
|
||||||
|
for link in resolved:
|
||||||
|
remote_id = link_pattern[link]['items'][0]
|
||||||
|
local_track = local_library[link]
|
||||||
|
remote_track = user_tracks_by_id[remote_id]
|
||||||
|
|
||||||
|
print(f" {index}: '{local_track['name']}' by '{local_track['artist']}'", end=' ----> ')
|
||||||
|
print(f"'{remote_track['name']}' by '{get_artists_str(remote_track['artists'])}'")
|
||||||
|
index += 1
|
||||||
|
|
||||||
|
return
|
||||||
|
print(f"\nUnresolved tracks from the local: {len(unresolved_local)}")
|
||||||
|
index = 0
|
||||||
|
for link in unresolved_local:
|
||||||
|
local_track = local_library[link]
|
||||||
|
print(f" {index}: '{local_track['name']}' by '{local_track['artist']}'")
|
||||||
|
index += 1
|
||||||
|
|
||||||
|
|
||||||
|
class Interpreter(cmd.Cmd):
|
||||||
|
intro = "Welcome to the interactive command loop. Type help or ? to list commands.\n"
|
||||||
|
prompt = "(spotify) "
|
||||||
|
|
||||||
|
def do_link_patterns(self, arg):
|
||||||
|
"""prints available linking patterns"""
|
||||||
|
print(f"0 - fuzzy tags")
|
||||||
|
print(f"1 - spotify")
|
||||||
|
print(f"2 - fuzzy")
|
||||||
|
|
||||||
|
def do_link_pattern(self, arg):
|
||||||
|
"""prints available linking patterns"""
|
||||||
|
pattern_id = int(arg.split()[0])
|
||||||
|
print_link_stats(link_patterns[pattern_id])
|
||||||
|
|
||||||
|
def do_playlists(self, arg):
|
||||||
|
"""prints user playlists"""
|
||||||
|
pl_id = 0
|
||||||
|
for pl in get_playlist_names(playlists):
|
||||||
|
print(f"{pl_id} - {pl}")
|
||||||
|
pl_id += 1
|
||||||
|
|
||||||
|
def do_playlist_tracks(self, arg):
|
||||||
|
"""prints playlist [name]"""
|
||||||
|
try:
|
||||||
|
pl_id = int(arg.split()[0])
|
||||||
|
print_playlist_tracks(pl_id)
|
||||||
|
except IndexError:
|
||||||
|
print("Invalid input")
|
||||||
|
|
||||||
|
def do_top_artists(self, arg):
|
||||||
|
"""Top artists"""
|
||||||
|
print_top_artists()
|
||||||
|
|
||||||
|
def do_top_tracks(self, arg):
|
||||||
|
"""Top tracks"""
|
||||||
|
print_top_tracks()
|
||||||
|
|
||||||
|
def do_stat(self, arg):
|
||||||
|
"""Print stats"""
|
||||||
|
print_stats()
|
||||||
|
|
||||||
|
def do_tracks(self, arg):
|
||||||
|
"""Print tracks"""
|
||||||
|
print_tracks()
|
||||||
|
|
||||||
|
def do_fetch(self, arg):
|
||||||
|
"""Fetch all spotify data"""
|
||||||
|
try:
|
||||||
|
update_access_token()
|
||||||
|
fetch_data()
|
||||||
|
|
||||||
|
global playlists, top_artists, top_tracks, user_tracks
|
||||||
|
|
||||||
|
playlists = get_data('playlistTracks')
|
||||||
|
top_artists = get_data('top_artists')
|
||||||
|
top_tracks = get_data('top_tracks')
|
||||||
|
user_tracks = get_data('tracks')
|
||||||
|
|
||||||
|
except ValueError:
|
||||||
|
print("Can not fetch the data.")
|
||||||
|
|
||||||
|
def do_exit(self, arg):
|
||||||
|
"""Exit the command loop: exit"""
|
||||||
|
print("Goodbye!")
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
Interpreter().cmdloop()
|
||||||
Loading…
Add table
Add a link
Reference in a new issue