From 4734befe984c168abf7b8679b8a542bfa7e6fdc5 Mon Sep 17 00:00:00 2001 From: Ilya Shurupov <163508118+elushaX@users.noreply.github.com> Date: Mon, 8 Jul 2024 23:14:29 +0300 Subject: [PATCH] stable need run main fetch command then run indexer then LinkerPatternGenerator then link_pattern command in the main --- .gitignore | 4 + DataBase.py | 49 ++++++++++ LinkerPatternGenerator.py | 137 ++++++++++++++++++++++++++ LocalLibraryIndexer.py | 80 ++++++++++++++++ ParseItunesToJson.py | 101 +++++++++++++++++++ README.md | 3 +- SpotifyAuthenticator.py | 122 +++++++++++++++++++++++ SpotifyWebAPI.py | 190 ++++++++++++++++++++++++++++++++++++ env | 36 +++++++ main.py | 197 ++++++++++++++++++++++++++++++++++++++ 10 files changed, 918 insertions(+), 1 deletion(-) create mode 100644 .gitignore create mode 100644 DataBase.py create mode 100644 LinkerPatternGenerator.py create mode 100644 LocalLibraryIndexer.py create mode 100755 ParseItunesToJson.py create mode 100644 SpotifyAuthenticator.py create mode 100644 SpotifyWebAPI.py create mode 100755 env create mode 100644 main.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2803b17 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +prefetched* +__pycache__ +.idea +secret diff --git a/DataBase.py b/DataBase.py new file mode 100644 index 0000000..c0f230b --- /dev/null +++ b/DataBase.py @@ -0,0 +1,49 @@ +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: + print(f"File {name}.json not found.") + return {} + except json.JSONDecodeError: + print(f"Error decoding JSON from {name}.json.") + return {} + +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 diff --git a/LinkerPatternGenerator.py b/LinkerPatternGenerator.py new file mode 100644 index 0000000..2f6eba2 --- /dev/null +++ b/LinkerPatternGenerator.py @@ -0,0 +1,137 @@ +from DataBase import * +from SpotifyWebAPI import find_song, update_access_token + +from rapidfuzz 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='fuzzy_tag_pattern_generator') 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() diff --git a/LocalLibraryIndexer.py b/LocalLibraryIndexer.py new file mode 100644 index 0000000..eecdd58 --- /dev/null +++ b/LocalLibraryIndexer.py @@ -0,0 +1,80 @@ +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": + mp3track = eyed3.load(abs_path) + if not mp3track: + print(f"Failed to load mp3 {abs_path} using just name of the file as track name") + track['name'] = track['path'] + else: + tag = mp3track.tag + if not tag: + print(f"Invalid mp3 header {abs_path} using just name of the file as track name") + track['name'] = track['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} using just name of the file as track name") + track['name'] = track['path'] + + save_data(new_library, "local_library") + + +if __name__ == "__main__": + update_local_library("/home/auser/Music/") diff --git a/ParseItunesToJson.py b/ParseItunesToJson.py new file mode 100755 index 0000000..1505a36 --- /dev/null +++ b/ParseItunesToJson.py @@ -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') diff --git a/README.md b/README.md index 508172a..4a45af3 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,3 @@ # spotifyFetcher -loads all data through web api +loads all data through web api
+client secret - 4ced4e4575094b749834a60996680c6c diff --git a/SpotifyAuthenticator.py b/SpotifyAuthenticator.py new file mode 100644 index 0000000..e06d343 --- /dev/null +++ b/SpotifyAuthenticator.py @@ -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 = '3b4db45bb66b45e9a2532989c8034332' +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://127.0.0.1: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'Authorization code received. You can close this window.') + + 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'Authorization code not found.') + + +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://127.0.0.1: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() + diff --git a/SpotifyWebAPI.py b/SpotifyWebAPI.py new file mode 100644 index 0000000..71ec2f7 --- /dev/null +++ b/SpotifyWebAPI.py @@ -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() diff --git a/env b/env new file mode 100755 index 0000000..8955572 --- /dev/null +++ b/env @@ -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 diff --git a/main.py b/main.py new file mode 100644 index 0000000..9709e93 --- /dev/null +++ b/main.py @@ -0,0 +1,197 @@ +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): + if not len(arg.split()): + print("expected name of the pattern") + return + + """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__': + try: + Interpreter().cmdloop() + except KeyboardInterrupt as kb: + print("process terminated") \ No newline at end of file