refactor 2
This commit is contained in:
Шурупов Илья Викторович
parent
72af96eada
commit
39c88564ba
7 changed files with 426 additions and 503 deletions
|
|
@ -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