tmp

DataBase.py

@@ -17,6 +17,16 @@ def get_data(name):
 
 
 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")

Linker.py

@@ -0,0 +1,2 @@
+
+

LinkerPatternGenerator.py

@@ -0,0 +1,98 @@
+from DataBase import *
+from SpotifyWebAPI import find_song, update_access_token
+
+from fuzzywuzzy import fuzz
+
+
+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):
+        from tqdm import tqdm
+        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):
+        from tqdm import tqdm
+
+        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 run_pattern_generators():
+    #fuzzy_pattern_generator()
+    spotify_pattern_generator()
+
+
+if __name__ == "__main__":
+    run_pattern_generators()

LocalLibrary.py

@@ -0,0 +1,70 @@
+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
+                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")

SpotifyWebAPI.py

@@ -1,7 +1,8 @@
 import requests
-from loadArtwork import load
-from spotifyAuth import authenticate
-from common import *
+from PIL import Image
+from io import BytesIO
+from SpotifyAuthenticator import authenticate
+from DataBase import *
 
 FETCH_STEP = 50
 token = ''
@@ -18,6 +19,10 @@ def fetch(endpoint, method='GET', body=None):
     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"
@@ -129,36 +134,47 @@ def get_user_data():
     return user_id
 
 
-def find_song(song_pattern):
-    tracks = fetch(f"v1/search/?type=track&track=1&q={song_pattern}")['tracks']['items']
-    return tracks
+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 find_local_library():
-    with open('prefetched/local_library.json', 'r') as file:
-        local_library = json.load(file)
+def load_playlist_covers():
+    playlists = get_data('playlists')
 
-    from tqdm import tqdm
-
-    pattern_map = {}
-
-    total = len(local_library)
-    with tqdm(total=total, desc='Searching local songs on spotify') as pbar:
-        found_tracks = {}
-        for track, path in local_library.items():
-            found_tracks[track] = find_song(track)
-            pbar.update(1)
-
-    with open('prefetched/local_library_on_spotify.json', 'w') as file:
-        json.dump(found_tracks, file, indent=2)
+    for pl in playlists:
+        if len(pl['images']):
+            url = pl['images'][0]['url']
+            save_image_from_url(url, pl['name'])
 
 
-def fetch_data():
+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()
 
-    find_local_library()
 
+def fetch_data():
     user_id = get_user_data()
 
     fetch_tracks(user_id)
@@ -166,8 +182,9 @@ def fetch_data():
     fetch_tracks_top()
     fetch_artists_top()
 
-    load()
+    load_artworks()
 
 
 if __name__ == "__main__":
+    update_access_token()
     fetch_data()

libResolver.py

@@ -1,97 +0,0 @@
-import common
-import os
-import fnmatch
-import json
-from fuzzywuzzy import fuzz
-
-
-def get_local_library(root_dir):
-    music_extensions = ('*.mp3', '*.flac', '*.wav', '*.aac', '*.ogg', '*.m4a')
-    music_files_map = {}
-    for dir_path, dir_names, filenames in os.walk(root_dir):
-        for pattern in music_extensions:
-            for filename in fnmatch.filter(filenames, pattern):
-                song_name = os.path.splitext(filename)[0]
-                relative_path = os.path.relpath(os.path.join(dir_path, filename), root_dir)
-                music_files_map[song_name] = relative_path
-    return music_files_map
-
-
-def get_matched(pattern, patterns):
-    result = {}
-    for target_pattern in patterns:
-        similarity_score = fuzz.token_set_ratio(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(source_patterns, target_patterns):
-    from tqdm import tqdm
-
-    pattern_map = {}
-
-    total = len(source_patterns)
-    with tqdm(total=total, desc='Progress') as pbar:
-        for pattern in source_patterns:
-            res = get_matched(pattern, target_patterns)
-            matched = {"score": 0, "items": []}
-            if len(res) and res[0][1] > 0.9:
-                matched['items'] = res[0:min(len(res), 2)]
-                matched['score'] = res[0][1]
-            pattern_map[pattern] = matched
-            pbar.update(1)
-
-    return pattern_map
-
-
-def resolver1(local_library):
-    track_descriptions = common.get_data("tracks")
-
-    source_patterns = []
-    for track in track_descriptions:
-        artists = common.get_artists_str(track['track']['artists'])
-        pattern = f"{track['track']['name']} {artists}"
-        source_patterns.append(pattern)
-
-    target_patterns = [local for _, local in local_library.items()]
-
-    pattern_map = resolve_tracks(source_patterns, target_patterns)
-
-    with open('prefetched/resolved.json', 'w') as file:
-        json.dump(pattern_map, file, indent=2)
-
-
-def resolver2(local_library):
-    found_tracks = common.get_data("local_library_on_spotify")
-    track_descriptions = common.get_data("tracks")
-
-    user_tracks_ids = {track['track']['id']: track for track in track_descriptions}
-
-    resolved_map = {}
-
-    for name, found in found_tracks.items():
-        if not len(found):
-            continue
-        track_id = found[0]['id']
-        if track_id in user_tracks_ids:
-            resolved_map[name] = track_id
-
-    with open('prefetched/resolved2.json', 'w') as file:
-        json.dump(resolved_map, file, indent=2)
-
-def resolve(root_directory):
-    local_library = get_local_library(root_directory)
-    with open('prefetched/local_library.json', 'w') as file:
-        json.dump(local_library, file, indent=2)
-
-    resolver2(local_library)
-
-
-if __name__ == "__main__":
-    # root_directory = "/mnt/main/data/music/raw"
-    #root_directory = input("root: ")
-    resolve("/mnt/main/data/music/raw")
-
-

loadArtwork.py

@@ -1,45 +0,0 @@
-import requests
-import os
-from PIL import Image
-from io import BytesIO
-
-from common import *
-
-
-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():
-    load_user_cover()
-    load_playlist_covers()
-
-
-if __name__ == "__main__":
-    load()

main.py

@@ -1,5 +1,5 @@
-from common import *
-from spotifyFetch import fetch_data
+from DataBase import *
+from SpotifyWebAPI import fetch_data
 import cmd
 
 playlists = get_data('playlistTracks')
@@ -8,8 +8,8 @@ top_tracks = get_data('top_tracks')
 user_tracks = get_data('tracks')
 
 
-def get_playlist_names():
-    return [name for name, items in playlists.items()]
+def get_playlist_names(pl):
+    return [name for name, items in pl.items()]
 
 
 def print_playlist_tracks(name):
@@ -57,7 +57,7 @@ class Interpreter(cmd.Cmd):
 
     def do_playlists(self, arg):
         """prints user playlists"""
-        print("\n".join(get_playlist_names()))
+        print("\n".join(get_playlist_names(playlists)))
 
     def do_playlist_tracks(self, arg):
         """prints playlist [name]"""

spotifyJsonCleanUp.py

@@ -1,32 +0,0 @@
-import os
-import json
-
-
-def process_json_files(directory):
-    for filename in os.listdir(directory):
-        if filename.endswith('.json') and ("_mod.json" not in filename):
-            filepath = os.path.join(directory, filename)
-
-            with open(filepath, 'r') as f:
-                try:
-                    data = json.load(f)
-                except json.JSONDecodeError:
-                    continue
-
-            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
-
-            modified_data = remove_available_markets(data)
-
-            modified_filepath = os.path.join(directory, f"{os.path.splitext(filename)[0]}_mod.json")
-            with open(modified_filepath, 'w') as f:
-                json.dump(modified_data, f, indent=4)
-
-
-if __name__ == "__main__":
-    process_json_files("prefetched")