fixes
This commit is contained in:
Шурупов Илья Викторович
parent
5f4bbbaa59
commit
20c7b28ed6
7 changed files with 91 additions and 276 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -2,3 +2,6 @@ prefetched*
|
|||
__pycache__
|
||||
.idea
|
||||
secret
|
||||
work_dir
|
||||
old
|
||||
tmp
|
||||
5
Flow.py
5
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()
|
||||
|
|
|
|||
10
main.py
10
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")
|
||||
|
|
|
|||
36
old/env
36
old/env
|
|
@ -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
|
||||
193
old/main.py
193
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 = []
|
||||
|
|
@ -115,120 +39,3 @@ 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")
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue