refactor 2
This commit is contained in:
Шурупов Илья Викторович
parent
a620e7a4f5
commit
5f4bbbaa59
7 changed files with 426 additions and 503 deletions
65
Flow.py
Normal file
65
Flow.py
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
|
||||
from Env import client_id, client_secret, local_library_path
|
||||
from Env import work_dir
|
||||
|
||||
from src.LibrarySerializer import LibrarySaver, LibraryLoader
|
||||
|
||||
from src.spotify.SpotifyWebAPI import fetch_all
|
||||
from src.spotify.ParseSpotify import Parser
|
||||
|
||||
from src.local.LocalLibraryIndexer import update_local_library
|
||||
from src.local.ParseLocal import LocalParser
|
||||
|
||||
from src.MapLocalToSpotify import map_local_to_spotify
|
||||
|
||||
|
||||
class Flow:
|
||||
spotify_library = None
|
||||
local_library = None
|
||||
|
||||
@staticmethod
|
||||
def fetch_spotify(self):
|
||||
fetch_all(client_id, client_secret, work_dir)
|
||||
|
||||
@staticmethod
|
||||
def fetch_local():
|
||||
update_local_library(local_library_path, work_dir)
|
||||
|
||||
@staticmethod
|
||||
def parse_spotify_library():
|
||||
parser = Parser()
|
||||
|
||||
library_parsed = parser.parse(
|
||||
work_dir / "spotify" / "playlistTracks.json",
|
||||
work_dir / "spotify" / "tracks.json",
|
||||
work_dir / "spotify" / "top_tracks.json",
|
||||
work_dir / "spotify" / "top_artists.json",
|
||||
)
|
||||
|
||||
LibrarySaver(library_parsed).save(work_dir / "spotify_library.json")
|
||||
|
||||
@staticmethod
|
||||
def parse_local_library():
|
||||
parser = LocalParser()
|
||||
library_parsed = parser.parse(work_dir / "local" / "local_library.json")
|
||||
LibrarySaver(library_parsed).save(work_dir / "local_library.json")
|
||||
|
||||
def load_libraries(self):
|
||||
self.spotify_library = LibraryLoader().load(work_dir / "spotify_library.json")
|
||||
self.local_library = LibraryLoader().load(work_dir / "local_library.json")
|
||||
|
||||
def map_local_to_spotify(self):
|
||||
map_local_to_spotify(client_id, client_secret, self.local_library, self.spotify_library, work_dir)
|
||||
|
||||
def run(self, arg):
|
||||
self.fetch_local()
|
||||
# self.do_fetch_spotify()
|
||||
|
||||
self.parse_local_library()
|
||||
self.parse_spotify_library()
|
||||
|
||||
self.load_libraries()
|
||||
|
||||
# self.map_local_to_spotify()
|
||||
|
||||
return True
|
||||
86
main.py
86
main.py
|
|
@ -1,77 +1,17 @@
|
|||
|
||||
import cmd
|
||||
|
||||
from Env import client_id, client_secret, local_library_path
|
||||
from Env import work_dir
|
||||
|
||||
from src.LibrarySerializer import LibrarySaver, LibraryLoader
|
||||
|
||||
from src.spotify.SpotifyWebAPI import fetch_all
|
||||
from src.spotify.ParseSpotify import Parser
|
||||
|
||||
from src.local.LocalLibraryIndexer import update_local_library
|
||||
from src.local.ParseLocal import LocalParser
|
||||
|
||||
from src.MapLocalToSpotify import map_local_to_spotify
|
||||
|
||||
|
||||
class Interpreter(cmd.Cmd):
|
||||
intro = "Welcome to the interactive command loop. Type help or ? to list commands.\n"
|
||||
prompt = "(kuw) "
|
||||
|
||||
spotify_library = None
|
||||
local_library = None
|
||||
|
||||
@staticmethod
|
||||
def do_fetch_spotify(self):
|
||||
fetch_all(client_id, client_secret, work_dir)
|
||||
|
||||
|
||||
@staticmethod
|
||||
def do_fetch_local():
|
||||
update_local_library(local_library_path, work_dir)
|
||||
|
||||
|
||||
def do_parse_spotify_library(self):
|
||||
parser = Parser()
|
||||
library_parsed = parser.parse(work_dir / "spotify" / "playlistTracks.json", work_dir / "spotify" / "tracks.json")
|
||||
LibrarySaver(library_parsed).save(work_dir / "spotify_library.json")
|
||||
|
||||
def do_parse_local_library(self):
|
||||
parser = LocalParser()
|
||||
library_parsed = parser.parse(work_dir / "local" / "local_library.json")
|
||||
LibrarySaver(library_parsed).save(work_dir / "local_library.json")
|
||||
|
||||
def do_load_libraries(self):
|
||||
self.spotify_library = LibraryLoader().load(work_dir / "spotify_library.json")
|
||||
self.local_library = LibraryLoader().load(work_dir / "local_library.json")
|
||||
|
||||
def do_map_local_to_spotify(self):
|
||||
map_local_to_spotify(client_id, client_secret, self.local_library, self.spotify_library, work_dir)
|
||||
|
||||
|
||||
def do_run(self, arg):
|
||||
# self.do_fetch_local()
|
||||
# self.do_fetch_spotify()
|
||||
|
||||
# self.do_parse_local_library()
|
||||
# self.do_parse_spotify_library()
|
||||
|
||||
self.do_load_libraries()
|
||||
|
||||
self.do_map_local_to_spotify()
|
||||
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def do_exit(self, arg):
|
||||
return True
|
||||
|
||||
import Flow
|
||||
|
||||
if __name__ == '__main__':
|
||||
try:
|
||||
Interpreter().do_run("")
|
||||
# Interpreter().cmdloop()
|
||||
flow = Flow.Flow()
|
||||
|
||||
except KeyboardInterrupt as kb:
|
||||
print("process terminated")
|
||||
# flow.fetch_spotify()
|
||||
flow.fetch_local()
|
||||
|
||||
flow.parse_spotify_library()
|
||||
flow.parse_local_library()
|
||||
|
||||
flow.load_libraries()
|
||||
|
||||
flow.map_local_to_spotify()
|
||||
|
||||
print("asd")
|
||||
|
|
|
|||
176
src/Library.py
176
src/Library.py
|
|
@ -1,159 +1,51 @@
|
|||
from __future__ import annotations
|
||||
import json
|
||||
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import List, Optional, Any
|
||||
|
||||
class Artist:
|
||||
def __init__(self):
|
||||
self.id : str = None
|
||||
self.title : str = None
|
||||
self.tracks : List[Track] = []
|
||||
|
||||
# =========================
|
||||
# Core models
|
||||
# =========================
|
||||
|
||||
class Track:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
id: str,
|
||||
title: str,
|
||||
artists: List[str],
|
||||
album: Optional[str] = None,
|
||||
duration_ms: Optional[int] = None,
|
||||
added_at: Optional[datetime] = None,
|
||||
spotify: Optional["SpotifyData"] = None,
|
||||
local: Optional["LocalData"] = None,
|
||||
):
|
||||
self.id = id
|
||||
self.title = title
|
||||
self.artists = artists
|
||||
self.album = album
|
||||
self.duration_ms = duration_ms
|
||||
self.added_at = added_at
|
||||
self.spotify = spotify
|
||||
self.local = local
|
||||
|
||||
|
||||
class Playlist:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
name: str,
|
||||
tracks: Optional[List[Track]] = None,
|
||||
created_at: Optional[datetime] = None,
|
||||
description: Optional[str] = None,
|
||||
spotify: Optional["SpotifyData"] = None,
|
||||
):
|
||||
self.name = name
|
||||
self.tracks = tracks or []
|
||||
self.created_at = created_at
|
||||
self.description = description
|
||||
self.spotify = spotify
|
||||
def __init__(self):
|
||||
self.id : str = None
|
||||
self.title : str = None
|
||||
self.artists : List[Artist] = []
|
||||
self.album : Album = None
|
||||
self.duration_ms : int = None
|
||||
self.added_at : datetime = None
|
||||
self.local_path : Path = None
|
||||
|
||||
|
||||
class Album:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
title: str,
|
||||
release_date: Optional[str] = None,
|
||||
tracks: Optional[List[Track]] = None,
|
||||
spotify: Optional["SpotifyData"] = None,
|
||||
):
|
||||
self.title = title
|
||||
self.release_date = release_date
|
||||
self.tracks = tracks or []
|
||||
self.spotify = spotify
|
||||
def __init__(self):
|
||||
self.id : str = None
|
||||
self.title : str = None
|
||||
self.release_date : datetime = None
|
||||
self.tracks : List[Track] = []
|
||||
self.artists : List[Artist] = []
|
||||
|
||||
|
||||
class TrackList:
|
||||
def __init__(self, *, tracks: Optional[List[Track]] = None):
|
||||
self.tracks = tracks or []
|
||||
class Playlist:
|
||||
def __init__(self):
|
||||
self.title : str = None
|
||||
self.tracks : List[Track] = []
|
||||
self.created_at : datetime = None
|
||||
self.description : str = None
|
||||
|
||||
def total_duration_ms(self) -> int:
|
||||
return sum(t.duration_ms or 0 for t in self.tracks)
|
||||
|
||||
|
||||
# =========================
|
||||
# Spotify models
|
||||
# =========================
|
||||
|
||||
class SpotifyArtist:
|
||||
def __init__(self, *, id: str, name: str, uri: str, url: str):
|
||||
self.id = id
|
||||
self.name = name
|
||||
self.uri = uri
|
||||
self.url = url
|
||||
|
||||
|
||||
class SpotifyAlbum:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
id: str,
|
||||
name: str,
|
||||
release_date: str,
|
||||
total_tracks: int,
|
||||
uri: str,
|
||||
url: str,
|
||||
):
|
||||
self.id = id
|
||||
self.name = name
|
||||
self.release_date = release_date
|
||||
self.total_tracks = total_tracks
|
||||
self.uri = uri
|
||||
self.url = url
|
||||
|
||||
|
||||
class SpotifyData:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
id: str,
|
||||
uri: str,
|
||||
url: str,
|
||||
href: str,
|
||||
file_path: Optional[Path] = None,
|
||||
popularity: Optional[int] = None,
|
||||
is_playable: Optional[bool] = None,
|
||||
is_local: Optional[bool] = None,
|
||||
artists: Optional[List[SpotifyArtist]] = None,
|
||||
album: Optional[SpotifyAlbum] = None,
|
||||
external_ids: Optional[dict[str, str]] = None,
|
||||
):
|
||||
self.id = id
|
||||
self.uri = uri
|
||||
self.url = url
|
||||
self.href = href
|
||||
self.file_path = file_path
|
||||
self.popularity = popularity
|
||||
self.is_playable = is_playable
|
||||
self.is_local = is_local
|
||||
self.artists = artists or []
|
||||
self.album = album
|
||||
self.external_ids = external_ids or {}
|
||||
|
||||
class LocalData:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
path: str,
|
||||
):
|
||||
self.path = path
|
||||
|
||||
# =========================
|
||||
# Library container
|
||||
# =========================
|
||||
|
||||
class Library:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
playlists: Optional[List[Playlist]] = None,
|
||||
tracks: Optional[List[Track]] = None,
|
||||
tracklists: Optional[List[TrackList]] = None,
|
||||
albums: Optional[List[Album]] = None,
|
||||
):
|
||||
self.playlists = playlists or []
|
||||
self.tracks = tracks or []
|
||||
self.tracklists = tracklists or []
|
||||
self.albums = albums or []
|
||||
def __init__(self):
|
||||
self.artists : List[Artist] = []
|
||||
self.playlists : List[Playlist] = []
|
||||
self.tracks : List[Track] = []
|
||||
self.albums : List[Album] = []
|
||||
self.liked_tracks : List[Track] = []
|
||||
self.top_tracks : List[Track]= []
|
||||
self.top_artists : List[Artist] = []
|
||||
|
||||
|
|
|
|||
|
|
@ -26,174 +26,112 @@ class LibrarySaver:
|
|||
def __init__(self, library: Library):
|
||||
self.library = library
|
||||
|
||||
@staticmethod
|
||||
def _artist_to_dict(artist: Artist) -> dict[str, Any]:
|
||||
return {
|
||||
"id": artist.id,
|
||||
"title": artist.title,
|
||||
"tracks": [track.id for track in artist.tracks],
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _track_to_dict(track: Track) -> dict[str, Any]:
|
||||
return {
|
||||
"id": track.id,
|
||||
"title": track.title,
|
||||
"artists": [artist.id for artist in track.artists],
|
||||
"album": track.album.id,
|
||||
"duration_ms": track.duration_ms,
|
||||
"added_at": _dt_to_str(track.added_at),
|
||||
"local_path": track.local_path,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _playlist_to_dict(playlist: Playlist) -> dict[str, Any]:
|
||||
return {
|
||||
"title": playlist.title,
|
||||
"tracks": [track.id for track in playlist.tracks],
|
||||
"created_at": _dt_to_str(playlist.created_at),
|
||||
"description": playlist.description,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _album_to_dict(album: Album) -> dict[str, Any]:
|
||||
return {
|
||||
"id": album.id,
|
||||
"title": album.title,
|
||||
"release_date": album.release_date,
|
||||
"tracks": [track.id for track in album.tracks],
|
||||
"artists": [artist.id for artist in album.artists]
|
||||
}
|
||||
|
||||
def save(self, path: Path) -> None:
|
||||
|
||||
data = {
|
||||
"tracks": [self._track_to_dict(track) for track in self.library.tracks],
|
||||
"albums": [self._album_to_dict(album) for album in self.library.albums],
|
||||
"artists": [self._artist_to_dict(artist) for artist in self.library.artists],
|
||||
"playlists": [self._playlist_to_dict(playlist) for playlist in self.library.playlists],
|
||||
"top_tracks": [track.id for track in self.library.top_tracks],
|
||||
"top_artists": [artist.id for artist in self.library.top_artists],
|
||||
"liked_tracks": [track.id for track in self.library.liked_tracks],
|
||||
}
|
||||
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with path.open("w", encoding="utf-8") as f:
|
||||
json.dump(self.to_dict(), f, indent=2, ensure_ascii=False)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"tracks": [self._track_to_dict(t) for t in self.library.tracks],
|
||||
"playlists": [self._playlist_to_dict(p) for p in self.library.playlists],
|
||||
"tracklists": [self._tracklist_to_dict(tl) for tl in self.library.tracklists],
|
||||
"albums": [self._album_to_dict(a) for a in self.library.albums],
|
||||
}
|
||||
|
||||
def _track_to_dict(self, t: Track) -> dict[str, Any]:
|
||||
return {
|
||||
"id": t.id,
|
||||
"title": t.title,
|
||||
"artists": t.artists,
|
||||
"album": t.album,
|
||||
"duration_ms": t.duration_ms,
|
||||
"added_at": _dt_to_str(t.added_at),
|
||||
"spotify": self._spotify_to_dict(t.spotify),
|
||||
"local": self._local_to_dict(t.local),
|
||||
}
|
||||
|
||||
def _playlist_to_dict(self, p: Playlist) -> dict[str, Any]:
|
||||
return {
|
||||
"name": p.name,
|
||||
"tracks": [t.id for t in p.tracks],
|
||||
"created_at": _dt_to_str(p.created_at),
|
||||
"description": p.description,
|
||||
"spotify": self._spotify_to_dict(p.spotify),
|
||||
}
|
||||
|
||||
def _tracklist_to_dict(self, tl: TrackList) -> dict[str, Any]:
|
||||
return {"tracks": [t.id for t in tl.tracks]}
|
||||
|
||||
def _album_to_dict(self, a: Album) -> dict[str, Any]:
|
||||
return {
|
||||
"title": a.title,
|
||||
"release_date": a.release_date,
|
||||
"tracks": [t.id for t in a.tracks],
|
||||
"spotify": self._spotify_to_dict(a.spotify),
|
||||
}
|
||||
|
||||
def _local_to_dict(self, local):
|
||||
if not local:
|
||||
return None
|
||||
return {"path": local.path}
|
||||
|
||||
def _spotify_to_dict(self, s: Optional[SpotifyData]) -> Optional[dict[str, Any]]:
|
||||
if not s:
|
||||
return None
|
||||
return {
|
||||
"id": s.id,
|
||||
"uri": s.uri,
|
||||
"url": s.url,
|
||||
"href": s.href,
|
||||
"file_path": _path_to_str(s.file_path),
|
||||
"popularity": s.popularity,
|
||||
"is_playable": s.is_playable,
|
||||
"is_local": s.is_local,
|
||||
"external_ids": s.external_ids,
|
||||
"artists": [{"id": a.id, "name": a.name, "uri": a.uri, "url": a.url} for a in s.artists],
|
||||
"album": {
|
||||
"id": s.album.id,
|
||||
"name": s.album.name,
|
||||
"release_date": s.album.release_date,
|
||||
"total_tracks": s.album.total_tracks,
|
||||
"uri": s.album.uri,
|
||||
"url": s.album.url,
|
||||
} if s.album else None,
|
||||
}
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
|
||||
|
||||
class LibraryLoader:
|
||||
def __init__(self):
|
||||
# Map of track id -> Track object
|
||||
self.track_map: dict[str, Track] = {}
|
||||
self.album_map: dict[str, Album] = {}
|
||||
self.artist_map: dict[str, Artist] = {}
|
||||
|
||||
def load_track(self, data: dict[str, Any]) -> Track:
|
||||
track = self.track_map.get(data.get("id"), Track())
|
||||
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())
|
||||
|
||||
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.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")]
|
||||
|
||||
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.id = data.get("id")
|
||||
album.tracks = [self.track_map.get(track_id, Track()) 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")]
|
||||
|
||||
self.album_map[album.id] = album
|
||||
return album
|
||||
|
||||
def load(self, path: Path) -> Library:
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
# Load tracks first
|
||||
tracks = []
|
||||
for t in data.get("tracks", []):
|
||||
track = self._track_from_dict(t)
|
||||
tracks.append(track)
|
||||
self.track_map[track.id] = track
|
||||
library = Library()
|
||||
|
||||
# Load playlists using track_map
|
||||
playlists = []
|
||||
for p in data.get("playlists", []):
|
||||
pl_tracks = []
|
||||
for tid in p.get("tracks", []):
|
||||
if tid not in self.track_map:
|
||||
raise KeyError(f"Track ID '{tid}' not found in track map")
|
||||
pl_tracks.append(self.track_map[tid])
|
||||
library.tracks = [self.load_track(track) for track in data.get("tracks")]
|
||||
library.albums = [self.load_album(album) for album in data.get("albums")]
|
||||
library.artists = [self.load_artist(artist) for artist in data.get("artists")]
|
||||
|
||||
playlists.append(Playlist(
|
||||
name=p["name"],
|
||||
tracks=pl_tracks,
|
||||
created_at=_str_to_dt(p.get("created_at")),
|
||||
description=p.get("description"),
|
||||
spotify=self._spotify_from_dict(p.get("spotify")),
|
||||
))
|
||||
library.top_artists = [self.artist_map[artist] for artist in data.get("top_artists")]
|
||||
library.top_tracks = [self.track_map[track] for track in data.get("top_tracks")]
|
||||
library.liked_tracks = [self.track_map[track] for track in data.get("liked_tracks")]
|
||||
|
||||
|
||||
# Load tracklists
|
||||
tracklists = []
|
||||
for tl in data.get("tracklists", []):
|
||||
tl_tracks = []
|
||||
for tid in tl.get("tracks", []):
|
||||
if tid not in self.track_map:
|
||||
raise KeyError(f"Track ID '{tid}' not found in track map (tracklist)")
|
||||
tl_tracks.append(self.track_map[tid])
|
||||
tracklists.append(TrackList(tracks=tl_tracks))
|
||||
|
||||
# Load albums
|
||||
albums = []
|
||||
for a in data.get("albums", []):
|
||||
alb_tracks = []
|
||||
for tid in a.get("tracks", []):
|
||||
if tid not in self.track_map:
|
||||
raise KeyError(f"Track ID '{tid}' not found in track map (album '{a.get('title')}')")
|
||||
alb_tracks.append(self.track_map[tid])
|
||||
albums.append(Album(
|
||||
title=a["title"],
|
||||
release_date=a.get("release_date"),
|
||||
tracks=alb_tracks,
|
||||
spotify=self._spotify_from_dict(a.get("spotify")),
|
||||
))
|
||||
|
||||
return Library(playlists=playlists, tracks=tracks, tracklists=tracklists, albums=albums)
|
||||
|
||||
def _track_from_dict(self, d: dict[str, Any]) -> Track:
|
||||
return Track(
|
||||
id=d["id"],
|
||||
title=d["title"],
|
||||
artists=d["artists"],
|
||||
album=d.get("album"),
|
||||
duration_ms=d.get("duration_ms"),
|
||||
added_at=_str_to_dt(d.get("added_at")),
|
||||
spotify=self._spotify_from_dict(d.get("spotify")),
|
||||
local=self._local_from_dict(d.get("local")),
|
||||
)
|
||||
|
||||
def _local_from_dict(self, d):
|
||||
if not d:
|
||||
return None
|
||||
return LocalData(
|
||||
path=d["path"],
|
||||
)
|
||||
|
||||
def _spotify_from_dict(self, d: Optional[dict[str, Any]]) -> Optional[SpotifyData]:
|
||||
if not d:
|
||||
return None
|
||||
return SpotifyData(
|
||||
id=d["id"],
|
||||
uri=d["uri"],
|
||||
url=d["url"],
|
||||
href=d["href"],
|
||||
file_path=_str_to_path(d.get("file_path")),
|
||||
popularity=d.get("popularity"),
|
||||
is_playable=d.get("is_playable"),
|
||||
is_local=d.get("is_local"),
|
||||
external_ids=d.get("external_ids"),
|
||||
artists=[SpotifyArtist(**a) for a in d.get("artists", [])],
|
||||
album=SpotifyAlbum(**d["album"]) if d.get("album") else None,
|
||||
)
|
||||
return library
|
||||
|
|
|
|||
|
|
@ -22,17 +22,17 @@ def map_local_to_spotify(client_id, client_secret, local_lib : Library, spotify_
|
|||
with tqdm(total=total, desc='Searching local songs on spotify') as pbar:
|
||||
for track in local_lib.tracks:
|
||||
|
||||
parts = [track.title, track.artists[0] if len(track.artists) else "", track.album]
|
||||
parts = [track.title, track.artists[0].title if len(track.artists) else "", track.album.title]
|
||||
search_pattern = " ".join(p for p in parts if p)
|
||||
|
||||
if search_pattern == "":
|
||||
print(f"cannot generate search pattern for the track - {track.local.path}")
|
||||
print(f"cannot generate search pattern for the track - {track.local_path}")
|
||||
pbar.update(1)
|
||||
continue
|
||||
|
||||
found_tracks = find_song(search_pattern)[0:10]
|
||||
found_tracks_map[track.id] = found_tracks
|
||||
search_log.append({"path" : track.local.path, "search_pattern": search_pattern, "result" : found_tracks_map[track.id]})
|
||||
search_log.append({"path" : track.local_path, "search_pattern": search_pattern, "result" : found_tracks_map[track.id]})
|
||||
pbar.update(1)
|
||||
|
||||
save_data(search_log, "spotify_search_log")
|
||||
|
|
|
|||
|
|
@ -1,40 +1,139 @@
|
|||
from datetime import datetime
|
||||
from typing import Any, Iterable
|
||||
from pathlib import Path
|
||||
import json
|
||||
import re
|
||||
|
||||
from src.Library import *
|
||||
|
||||
import re
|
||||
|
||||
PROTECTED_ARTISTS = {
|
||||
"Tyler, The Creator",
|
||||
"The Good, The Bad & The Queen",
|
||||
}
|
||||
|
||||
def split_artists(artist_tag: str) -> list[str]:
|
||||
if not artist_tag:
|
||||
return []
|
||||
|
||||
placeholder_comma = "§COMMA§"
|
||||
placeholder_amp = "§AMP§"
|
||||
|
||||
protected_map = {}
|
||||
|
||||
# Protect known artist names
|
||||
for artist in PROTECTED_ARTISTS:
|
||||
protected = (
|
||||
artist
|
||||
.replace(",", placeholder_comma)
|
||||
.replace("&", placeholder_amp)
|
||||
)
|
||||
protected_map[protected] = artist
|
||||
artist_tag = artist_tag.replace(artist, protected)
|
||||
|
||||
# Split remaining separators
|
||||
parts = re.split(r"[,&]", artist_tag)
|
||||
|
||||
# Restore protected names and trim spaces
|
||||
result = []
|
||||
for part in parts:
|
||||
part = part.strip()
|
||||
if not part:
|
||||
continue
|
||||
|
||||
part = (
|
||||
part
|
||||
.replace(placeholder_comma, ",")
|
||||
.replace(placeholder_amp, "&")
|
||||
)
|
||||
|
||||
result.append(part)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class LocalParser:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def parse_track_item(self, id, track: dict[str, Any]) -> Track:
|
||||
track = Track(
|
||||
title=track["name"],
|
||||
artists=[track.get("artist", [])],
|
||||
album=track["album"] if track.get("album") else None,
|
||||
duration_ms=None,
|
||||
added_at=None,
|
||||
spotify=None,
|
||||
id=id,
|
||||
local=LocalData(path=track["path"])
|
||||
)
|
||||
return track
|
||||
self.track_map: dict[str, Track] = {}
|
||||
self.artist_map: dict[str, Artist] = {}
|
||||
self.album_map: dict[str, Album] = {}
|
||||
self.library = Library()
|
||||
|
||||
|
||||
def parse_all_tracks(self, items: dict[str, list[dict[str, Any]]]) -> list[Track]:
|
||||
return [self.parse_track_item(id, track) for id, track in items.items()]
|
||||
def parse_artists(self, data: str) -> List[Artist]:
|
||||
if not data:
|
||||
data = "unknown"
|
||||
|
||||
def build_library_from_tracks(self, tracks_content_json: dict[str, list[dict[str, Any]]]) -> Library:
|
||||
library = Library()
|
||||
data = split_artists(data)
|
||||
|
||||
tracks = self.parse_all_tracks(tracks_content_json)
|
||||
library.tracklists.append(TrackList(tracks=list(tracks)))
|
||||
library.tracks.extend(tracks)
|
||||
artists = []
|
||||
|
||||
for artist_data in data:
|
||||
name = artist_data
|
||||
artist_id = name
|
||||
|
||||
if artist_id in self.artist_map:
|
||||
artists.append(self.artist_map[artist_id])
|
||||
continue
|
||||
|
||||
artist = Artist()
|
||||
artist.id = artist_id
|
||||
artist.title = name
|
||||
|
||||
self.artist_map[artist_id] = artist
|
||||
self.library.artists.append(artist)
|
||||
|
||||
artists.append(artist)
|
||||
|
||||
return artists
|
||||
|
||||
def parse_album(self, data: str) -> Album:
|
||||
if not data:
|
||||
data = "unknown"
|
||||
|
||||
name = data
|
||||
album_id = name
|
||||
|
||||
if album_id in self.album_map:
|
||||
return self.album_map[album_id]
|
||||
|
||||
album = Album()
|
||||
album.id = album_id
|
||||
album.title = name
|
||||
|
||||
self.album_map[album_id] = album
|
||||
self.library.albums.append(album)
|
||||
|
||||
return album
|
||||
|
||||
def add_tracks_to_albums_and_artists(self):
|
||||
for track_id, track in self.track_map.items():
|
||||
for artist in track.artists:
|
||||
if track not in artist.tracks:
|
||||
artist.tracks.append(track)
|
||||
|
||||
if track not in track.album.tracks:
|
||||
track.album.tracks.append(track)
|
||||
|
||||
return library
|
||||
|
||||
def parse(self, tracks_file: Path) -> Library:
|
||||
tracks_content = json.loads(tracks_file.read_text())
|
||||
return self.build_library_from_tracks(tracks_content)
|
||||
|
||||
for track_id, data in tracks_content.items():
|
||||
track = Track()
|
||||
|
||||
track.id = track_id
|
||||
track.title = data.get("name")
|
||||
track.artists = self.parse_artists(data.get("artist"))
|
||||
track.album = self.parse_album(data.get("album"))
|
||||
track.local_path = data.get("path")
|
||||
|
||||
self.track_map[track.id] = track
|
||||
self.library.tracks.append(track)
|
||||
|
||||
self.add_tracks_to_albums_and_artists()
|
||||
|
||||
return self.library
|
||||
|
||||
|
||||
def parse_track_item(self, id, track: dict[str, Any]) -> Track:
|
||||
|
||||
return track
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
from datetime import datetime
|
||||
from src.Library import *
|
||||
|
||||
from typing import Any, Iterable
|
||||
from pathlib import Path
|
||||
import json
|
||||
|
||||
from src.Library import *
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def get_artists_str(artists_pack):
|
||||
|
|
@ -11,129 +12,117 @@ def get_artists_str(artists_pack):
|
|||
artists = " ".join(track_artists)
|
||||
return artists
|
||||
|
||||
|
||||
class Parser:
|
||||
def __init__(self):
|
||||
# state: Spotify track ID -> Track object
|
||||
self.track_map: dict[str, Track] = {}
|
||||
self.artist_map: dict[str, Artist] = {}
|
||||
self.album_map: dict[str, Album] = {}
|
||||
self.library = Library()
|
||||
|
||||
# -------------------------
|
||||
# Low-level helpers
|
||||
# -------------------------
|
||||
@staticmethod
|
||||
def _parse_spotify_datetime(value: str | None) -> datetime | None:
|
||||
if not value:
|
||||
return None
|
||||
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
def parse(self, playlists: Path, tracks: Path, top_tracks: Path, top_artists: Path) -> Library:
|
||||
self.library = Library()
|
||||
|
||||
@staticmethod
|
||||
def _parse_spotify_artist(a: dict[str, Any]) -> SpotifyArtist:
|
||||
return SpotifyArtist(
|
||||
id=a["id"],
|
||||
name=a["name"],
|
||||
uri=a["uri"],
|
||||
url=a["external_urls"]["spotify"],
|
||||
)
|
||||
self.parse_tracks(json.loads(tracks.read_text()))
|
||||
self.parse_playlists(json.loads(playlists.read_text()))
|
||||
|
||||
@staticmethod
|
||||
def _parse_spotify_album(a: dict[str, Any]) -> SpotifyAlbum:
|
||||
return SpotifyAlbum(
|
||||
id=a["id"],
|
||||
name=a["name"],
|
||||
release_date=a.get("release_date"),
|
||||
total_tracks=a.get("total_tracks", 0),
|
||||
uri=a["uri"],
|
||||
url=a["external_urls"]["spotify"],
|
||||
)
|
||||
self.parse_top_artists(json.loads(top_artists.read_text()))
|
||||
self.parse_top_tracks(json.loads(top_tracks.read_text()))
|
||||
|
||||
def _parse_spotify_data(self, track_json: dict[str, Any]) -> SpotifyData:
|
||||
return SpotifyData(
|
||||
id=track_json["id"],
|
||||
uri=track_json["uri"],
|
||||
url=track_json["external_urls"]["spotify"],
|
||||
href=track_json["href"],
|
||||
popularity=track_json.get("popularity"),
|
||||
is_playable=track_json.get("is_playable"),
|
||||
is_local=track_json.get("is_local"),
|
||||
external_ids=track_json.get("external_ids", {}),
|
||||
artists=[self._parse_spotify_artist(a) for a in track_json.get("artists", [])],
|
||||
album=self._parse_spotify_album(track_json["album"])
|
||||
if track_json.get("album")
|
||||
else None,
|
||||
)
|
||||
self.add_tracks_to_albums_and_artists()
|
||||
|
||||
# -------------------------
|
||||
# Track parsers
|
||||
# -------------------------
|
||||
def parse_track_item(self, item: dict[str, Any]) -> Track:
|
||||
"""
|
||||
Parses a single track item and returns Track object.
|
||||
Reuses existing Track from track_map if already parsed.
|
||||
"""
|
||||
t = item["track"]
|
||||
track_id = t["id"]
|
||||
return self.library
|
||||
|
||||
def add_tracks_to_albums_and_artists(self):
|
||||
for track_id, track in self.track_map.items():
|
||||
for artist in track.artists:
|
||||
if track not in artist.tracks:
|
||||
artist.tracks.append(track)
|
||||
|
||||
if track not in track.album.tracks:
|
||||
track.album.tracks.append(track)
|
||||
|
||||
|
||||
def parse_playlists(self, data):
|
||||
for name, tracks in data.items():
|
||||
playlist = Playlist()
|
||||
playlist.title = name
|
||||
for track in tracks:
|
||||
track_data = track.get("track")
|
||||
if track_data:
|
||||
track_id = track_data.get("id")
|
||||
playlist.tracks.append(self.parse_track(track_data, track.get("added_at")))
|
||||
|
||||
self.library.playlists.append(playlist)
|
||||
|
||||
def parse_top_tracks(self, data):
|
||||
for track in data:
|
||||
if track.get("id") in self.track_map:
|
||||
self.library.top_tracks.append(self.track_map.get(track.get("id")))
|
||||
|
||||
def parse_top_artists(self, data):
|
||||
for artist in data:
|
||||
if artist.get("id") in self.artist_map:
|
||||
self.library.top_artists.append(self.artist_map.get(artist.get("id")))
|
||||
|
||||
def parse_tracks(self, data):
|
||||
for track in data:
|
||||
self.parse_track(track.get("track"), track.get("added_at"))
|
||||
|
||||
for track in data:
|
||||
self.library.liked_tracks.append(self.track_map.get(track.get("track").get("id")))
|
||||
|
||||
def parse_track(self, data: dict[str, Any], added_time : datetime) -> Track:
|
||||
track_id = data.get("id")
|
||||
|
||||
# return existing track if already parsed
|
||||
if track_id in self.track_map:
|
||||
return self.track_map[track_id]
|
||||
|
||||
track = Track(
|
||||
title=t["name"],
|
||||
artists=[a["name"] for a in t.get("artists", [])],
|
||||
album=t["album"]["name"] if t.get("album") else None,
|
||||
duration_ms=t.get("duration_ms"),
|
||||
added_at=self._parse_spotify_datetime(item.get("added_at")),
|
||||
spotify=self._parse_spotify_data(t),
|
||||
id=self._parse_spotify_data(t).id
|
||||
)
|
||||
track = Track()
|
||||
|
||||
self.track_map[track_id] = track
|
||||
|
||||
track.id = track_id
|
||||
track.title = data.get("name")
|
||||
track.artists = [self.parse_artist(artist) for artist in data.get("artists")]
|
||||
track.album = self.parse_album(data.get("album"))
|
||||
track.duration_ms = data.get("duration_ms")
|
||||
|
||||
self.library.tracks.append(track)
|
||||
|
||||
return track
|
||||
|
||||
def parse_track_items(self, items: Iterable[dict[str, Any]]) -> list[Track]:
|
||||
return [self.parse_track_item(i) for i in items if i.get("track")]
|
||||
def parse_album(self, data):
|
||||
album_id = data.get("id")
|
||||
|
||||
# -------------------------
|
||||
# Playlist parsers
|
||||
# -------------------------
|
||||
def parse_playlist_content(self, playlist_name: str, items: list[dict[str, Any]]) -> Playlist:
|
||||
tracks = self.parse_track_items(items)
|
||||
return Playlist(name=playlist_name, tracks=tracks)
|
||||
if album_id in self.album_map:
|
||||
return self.album_map[album_id]
|
||||
|
||||
def parse_all_playlist_content(self, data: dict[str, list[dict[str, Any]]]) -> list[Playlist]:
|
||||
playlists: list[Playlist] = []
|
||||
for name, items in data.items():
|
||||
playlists.append(self.parse_playlist_content(name, items))
|
||||
return playlists
|
||||
album = Album()
|
||||
|
||||
def parse_all_tracks(self, items: dict[str, list[dict[str, Any]]]) -> list[Track]:
|
||||
return [self.parse_track_item(i) for i in items]
|
||||
self.album_map[album_id] = album
|
||||
|
||||
# -------------------------
|
||||
# Library builder
|
||||
# -------------------------
|
||||
def build_library_from_playlist_content(
|
||||
self, playlist_content_json: dict[str, list[dict[str, Any]]],
|
||||
tracks_content_json: dict[str, list[dict[str, Any]]]
|
||||
) -> Library:
|
||||
library = Library()
|
||||
album.id = data.get("id")
|
||||
album.title = data.get("name")
|
||||
album.artists = [self.parse_artist(artist) for artist in data.get("artists")]
|
||||
|
||||
self.parse_all_tracks(tracks_content_json)
|
||||
self.library.albums.append(album)
|
||||
|
||||
library.tracklists.append(TrackList(tracks=list((self.track_map.values()))))
|
||||
return album
|
||||
|
||||
library.playlists = self.parse_all_playlist_content(playlist_content_json)
|
||||
library.tracks.extend(self.track_map.values())
|
||||
def parse_artist(self, data: dict[str, Any]) -> Artist:
|
||||
artist_id = data.get("id")
|
||||
|
||||
return library
|
||||
if artist_id in self.artist_map:
|
||||
return self.artist_map[artist_id]
|
||||
|
||||
# -------------------------
|
||||
# Convenience method
|
||||
# -------------------------
|
||||
def parse(
|
||||
self,
|
||||
playlist_content_file: Path,
|
||||
tracks_file: Path,
|
||||
) -> Library:
|
||||
playlist_content = json.loads(playlist_content_file.read_text())
|
||||
tracks_content = json.loads(tracks_file.read_text())
|
||||
return self.build_library_from_playlist_content(playlist_content, tracks_content)
|
||||
artist = Artist()
|
||||
|
||||
self.artist_map[artist_id] = artist
|
||||
|
||||
artist.id = data.get("id")
|
||||
artist.title = data.get("name")
|
||||
|
||||
self.library.artists.append(artist)
|
||||
|
||||
return artist
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue