From 5f4bbbaa592a92a9b94455793421aada76108ab4 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: Sun, 4 Jan 2026 23:40:41 +0300 Subject: [PATCH] refactor 2 --- Flow.py | 65 ++++++++++ main.py | 86 ++----------- src/Library.py | 176 +++++-------------------- src/LibrarySerializer.py | 248 ++++++++++++++---------------------- src/MapLocalToSpotify.py | 6 +- src/local/ParseLocal.py | 151 ++++++++++++++++++---- src/spotify/ParseSpotify.py | 197 ++++++++++++++-------------- 7 files changed, 426 insertions(+), 503 deletions(-) create mode 100644 Flow.py diff --git a/Flow.py b/Flow.py new file mode 100644 index 0000000..bb35731 --- /dev/null +++ b/Flow.py @@ -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 \ No newline at end of file diff --git a/main.py b/main.py index 32682e3..98b4efa 100644 --- a/main.py +++ b/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") diff --git a/src/Library.py b/src/Library.py index dc2efa3..c3287b8 100644 --- a/src/Library.py +++ b/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] = [] diff --git a/src/LibrarySerializer.py b/src/LibrarySerializer.py index 06ea57b..bc968ea 100644 --- a/src/LibrarySerializer.py +++ b/src/LibrarySerializer.py @@ -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 diff --git a/src/MapLocalToSpotify.py b/src/MapLocalToSpotify.py index dae5252..84e37c5 100644 --- a/src/MapLocalToSpotify.py +++ b/src/MapLocalToSpotify.py @@ -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") diff --git a/src/local/ParseLocal.py b/src/local/ParseLocal.py index 4991ede..602e89e 100644 --- a/src/local/ParseLocal.py +++ b/src/local/ParseLocal.py @@ -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 diff --git a/src/spotify/ParseSpotify.py b/src/spotify/ParseSpotify.py index 535caa0..49882cb 100644 --- a/src/spotify/ParseSpotify.py +++ b/src/spotify/ParseSpotify.py @@ -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