From 20c7b28ed69e4cafe58c2584e9df9d5cb92009f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A8=D1=83=D1=80=D1=83=D0=BF=D0=BE=D0=B2=20=D0=98=D0=BB?= =?UTF-8?q?=D1=8C=D1=8F=20=D0=92=D0=B8=D0=BA=D1=82=D0=BE=D1=80=D0=BE=D0=B2?= =?UTF-8?q?=D0=B8=D1=87?= Date: Mon, 5 Jan 2026 01:19:00 +0300 Subject: [PATCH] fixes --- .gitignore | 3 + Flow.py | 5 + main.py | 10 +- old/env | 36 -------- old/main.py | 195 +-------------------------------------- src/LibrarySerializer.py | 37 ++++++-- src/StatGenerator.py | 81 +++++++++------- 7 files changed, 91 insertions(+), 276 deletions(-) delete mode 100755 old/env diff --git a/.gitignore b/.gitignore index 2803b17..880f9c3 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,6 @@ prefetched* __pycache__ .idea secret +work_dir +old +tmp \ No newline at end of file diff --git a/Flow.py b/Flow.py index bb35731..e982d39 100644 --- a/Flow.py +++ b/Flow.py @@ -11,6 +11,7 @@ from src.local.LocalLibraryIndexer import update_local_library from src.local.ParseLocal import LocalParser from src.MapLocalToSpotify import map_local_to_spotify +from src.StatGenerator import log_stats class Flow: @@ -51,6 +52,10 @@ class Flow: def map_local_to_spotify(self): map_local_to_spotify(client_id, client_secret, self.local_library, self.spotify_library, work_dir) + def log_stats(self): + log_stats(work_dir / "local_to_spotify_id_map.json", work_dir / "stats.json", self.local_library, self.spotify_library) + + def run(self, arg): self.fetch_local() # self.do_fetch_spotify() diff --git a/main.py b/main.py index 98b4efa..1069171 100644 --- a/main.py +++ b/main.py @@ -5,13 +5,15 @@ if __name__ == '__main__': flow = Flow.Flow() # flow.fetch_spotify() - flow.fetch_local() + # flow.fetch_local() - flow.parse_spotify_library() - flow.parse_local_library() + # flow.parse_spotify_library() + # flow.parse_local_library() flow.load_libraries() - flow.map_local_to_spotify() + # flow.map_local_to_spotify() + + flow.log_stats() print("asd") diff --git a/old/env b/old/env deleted file mode 100755 index 8955572..0000000 --- a/old/env +++ /dev/null @@ -1,36 +0,0 @@ - -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/old/main.py b/old/main.py index 10d378a..0b2a991 100644 --- a/old/main.py +++ b/old/main.py @@ -1,79 +1,3 @@ -from DataBase import * -from LocalLibraryIndexer import update_local_library -from LinkerPatternGenerator import run_pattern_generators -from LocalPLaylistGenerator import generate_playlists - -from SpotifyWebAPI import fetch_data, update_access_token -import cmd - -local_library_path = "/home/auser/Music/" -rel_autogen_dir = "0autogen" - -playlists = get_data('playlistTracks') -top_artists = get_data('top_artists') -top_tracks = get_data('top_tracks') -user_tracks = get_data('tracks') -link_pattern = get_data('link_pattern_spotify') -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 = [] @@ -114,121 +38,4 @@ def print_link_stats(link_pattern): 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 - - -class Interpreter(cmd.Cmd): - intro = "Welcome to the interactive command loop. Type help or ? to list commands.\n" - prompt = "(spotify) " - - def do_list_stat_vs_types(self, arg): - """prints available linking patterns""" - print(f"0 - fuzzy tags") - print(f"1 - spotify") - print(f"2 - fuzzy") - - def do_stat_local_library_vs_spotify(self, arg): - """compares local library and remote with the link pattern generated by the generate_pattern_command""" - print_link_stats(link_pattern) - - 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_generate_local_playlists(self, arg): - """generates local playlists based on the vs comparison type""" - generate_playlists(local_library, local_library_path, playlists, link_pattern, local_library_path + rel_autogen_dir) - - - def do_index_local_library(self, arg): - """find local tracks and fetches the data from them for further analysis""" - global local_library - update_local_library(local_library_path) - local_library = get_data("local_library") - - def do_generate_vs_stat(self, arg): - """compares local and remote library and generates the vs stats""" - - global link_pattern - run_pattern_generators() - link_pattern = get_data('link_pattern_spotify') - - 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_generate_todo(self, arg): - """generates todos for the pirates""" - print("sorry not implemented yet.") - - def do_run_all_default(self, arg): - """runs general flow""" - - local_lib_path = "/home/auser/Music/" - print("Fetching spotify data...") - self.do_fetch([]) - - print("Fetching local data...") - self.do_index_local_library(f"{local_lib_path}/organized") - - print("Generating vs patterns...") - self.do_generate_vs_stat("0") - - print("Generating todos...") - self.do_generate_todo(f"{local_lib_path}") - - print("Generating local playlists...") - self.do_generate_local_playlists(f"{local_lib_path}") - - return True - - 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 + index += 1 \ No newline at end of file diff --git a/src/LibrarySerializer.py b/src/LibrarySerializer.py index bc968ea..cbccd9f 100644 --- a/src/LibrarySerializer.py +++ b/src/LibrarySerializer.py @@ -88,34 +88,55 @@ class LibraryLoader: self.album_map: dict[str, Album] = {} self.artist_map: dict[str, Artist] = {} + def get_artist(self, artist_id) -> Artist: + artist = self.artist_map.get(artist_id) + if artist is None: + artist = Artist() + self.artist_map[artist_id] = artist + return artist + + def get_album(self, id) -> Album: + item = self.album_map.get(id) + if item is None: + item = Album() + self.album_map[id] = item + return item + + def get_track(self, id) -> Track: + item = self.track_map.get(id) + if item is None: + item = Track() + self.track_map[id] = item + return item + def load_track(self, data: dict[str, Any]) -> Track: - track = self.track_map.get(data.get("id"), Track()) + track = self.get_track(data.get("id")) track.id = data.get("id") track.title = data.get("title") track.duration_ms = data.get("duration_ms") track.added_at = _str_to_dt(data.get("added_at")) track.local_path = data.get("local_path") - track.artists = [self.artist_map.get(artist_id, Artist()) for artist_id in data.get("artists")] - track.album = self.album_map.get(data.get("album"), Album()) + track.artists = [self.get_artist(artist_id) for artist_id in data.get("artists")] + track.album = self.get_album(data.get("album")) self.track_map[track.id] = track return track def load_artist(self, data: dict[str, Any]): - artist = self.artist_map.get(data.get("id"), Artist()) + artist = self.get_artist(data.get("id")) artist.id = data.get("id") artist.title = data.get("title") - artist.tracks = [self.track_map.get(track_id, Track()) for track_id in data.get("tracks")] + artist.tracks = [self.get_track(track_id) for track_id in data.get("tracks")] self.artist_map[artist.id] = artist return artist def load_album(self, data: dict[str, Any]): - album = self.album_map.get(data.get("id"), Album()) + album = self.get_album(data.get("id")) album.id = data.get("id") - album.tracks = [self.track_map.get(track_id, Track()) for track_id in data.get("tracks")] + album.tracks = [self.get_track(track_id) for track_id in data.get("tracks")] album.title = data.get("title") - album.artists = [self.artist_map.get(artist_id, Artist()) for artist_id in data.get("artists")] + album.artists = [self.get_artist(artist_id) for artist_id in data.get("artists")] self.album_map[album.id] = album return album diff --git a/src/StatGenerator.py b/src/StatGenerator.py index c40bebb..d6ebc41 100644 --- a/src/StatGenerator.py +++ b/src/StatGenerator.py @@ -1,42 +1,55 @@ +from src.Library import * -def print_link_stats(link_pattern): - resolved_reversed = {} - resolved = [] - unresolved_local = [] - unresolved_remote = [] +from pathlib import Path +import json - 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) +out = { + "local_count": 0, + "resolved_count": 0, + "unresolved_tracks": [], +} - for track in user_tracks: - track_id = track['track']['id'] - if track_id not in resolved_reversed: - unresolved_remote.append(track_id) +def get_mapping(mapping_path) -> dict[str, Any]: + data = {} - print(f"Local Tracks: {len(local_library)}") - print(f"Remote Tracks: {len(user_tracks)}") + with Path(mapping_path).open("r", encoding="utf-8") as f: + data = json.load(f) - unresolved_remote = sort_remote_tracks_by_popularity(unresolved_remote) - unresolved_remote = unresolved_remote[0:min(len(unresolved_remote), 100)] + out = {} - 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 + for local, remotes in data.items(): + if len(remotes["items"]): + out[local] = remotes["items"][0] - 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] + return out - 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 \ No newline at end of file + +def find_add_local_paths(mapping_path: Path, local_lib: Library, remote_lib: Library): + mapping = get_mapping(mapping_path) + + out["resolved_count"] = len(mapping) + + remote_id_map = {track.id: track for track in remote_lib.tracks} + local_id_map = {track.id: track for track in local_lib.tracks} + + for local, remote in mapping.items(): + remote_id_map[remote].local_path = local_id_map[local].local_path + + +def generate_stats(local_lib: Library, remote_lib: Library, output_path : Path): + + out["local_count"] = len(local_lib.tracks) + + for track in remote_lib.top_tracks: + if not track.local_path: + out["unresolved_tracks"].append( + f"{track.title} - {track.artists[0].title if len(track.artists) else ""} - {track.album.title}") + + + with output_path.open("w", encoding="utf-8") as f: + json.dump(out, f, indent=2, ensure_ascii=False) + + +def log_stats(mapping_path : Path, output_path : Path, local_lib: Library, remote_lib: Library): + find_add_local_paths(mapping_path, local_lib, remote_lib) + generate_stats(local_lib, remote_lib, output_path)