diff --git a/votify/downloader_song.py b/votify/downloader_song.py index fac7332..d258748 100644 --- a/votify/downloader_song.py +++ b/votify/downloader_song.py @@ -23,86 +23,123 @@ class DownloaderSong(DownloaderAudio): self.no_lrc = no_lrc def get_tags( - self, - track_metadata: dict, - album_metadata: dict, - track_credits: dict, - lyrics_unsynced: str, + self, + track_metadata: dict, + album_metadata: list, + track_credits: dict, + lyrics_unsynced: str, ) -> dict: - external_ids = track_metadata.get("external_ids") - external_urls = (track_metadata.get("linked_from") or track_metadata)[ - "external_urls" - ] - track_id = self.downloader.get_media_id(track_metadata) + track_info = track_metadata.get("data", {}).get("trackUnion", {}) + album_info = track_info.get("albumOfTrack", {}) + + external_url = track_info.get("sharingInfo", {}).get("shareUrl") + + isrc = None + + release_date_iso = album_info.get("date", {}).get("isoString") + precision_raw = album_info.get("date", {}).get("precision") + + precision = precision_raw.lower() if precision_raw else "day" + + release_date_str = release_date_iso.split("T")[0] if release_date_iso else None + release_date_datetime_obj = self.downloader.get_release_date_datetime_obj( - album_metadata["release_date"], - album_metadata["release_date_precision"], + release_date_str, + precision, ) - producers = next( - role - for role in track_credits["roleCredits"] - if role["roleTitle"] == "Producers" - )["artists"] - composers = next( - role - for role in track_credits["roleCredits"] - if role["roleTitle"] == "Writers" - )["artists"] - disc = next( - ( - track["disc_number"] - for track in album_metadata["tracks"]["items"] - if self.downloader.get_media_id(track) == track_id - ), + + track_artists = [] + + raw_artist_items = track_info.get("firstArtist", {}).get("items", []) + + other_artist_items = track_info.get("otherArtists", {}).get("items", []) + + all_artist_items = raw_artist_items + other_artist_items + + for item in all_artist_items: + name = item.get("profile", {}).get("name") + if name: + track_artists.append({"name": name}) + + if not track_artists: + alt_artists = track_info.get("artists", {}).get("items", []) + for item in alt_artists: + name = item.get("profile", {}).get("name") or item.get("name") + if name: + track_artists.append({"name": name}) + + if not track_artists: + track_artists = [{"name": "Unknown Artist"}] + + album_artists = track_artists + + def get_credits_artists(role_title): + try: + role = next( + r for r in track_credits.get("roleCredits", []) + if r.get("roleTitle") == role_title + ) + return role.get("artists", []) + except StopIteration: + return [] + + producers = get_credits_artists("Producers") + composers = get_credits_artists("Writers") + + current_uri = track_info.get("uri") + + current_track_entry = next( + (item.get("track", {}) for item in album_metadata if item.get("track", {}).get("uri") == current_uri), + {} ) + + disc_number = current_track_entry.get("discNumber", 1) + + try: + disc_total = max( + (item.get("track", {}).get("discNumber", 1) for item in album_metadata), + default=1 + ) + except Exception: + disc_total = 1 + + track_total = album_info.get("tracks", {}).get("totalCount") + if not track_total: + track_total = len(album_metadata) + + copyright_text = next( + (c["text"] for c in album_info.get("copyright", {}).get("items", []) if c.get("type") == "P"), + None + ) + label = album_info.get("courtesyLine") or album_info.get("label") + tags = { - "album": album_metadata["name"], - "album_artist": self.downloader.get_artist_string( - album_metadata["artists"] - ), - "artist": self.downloader.get_artist_string(track_metadata["artists"]), - "compilation": ( - True if album_metadata["album_type"] == "compilation" else False - ), + "album": album_info.get("name"), + "album_artist": self.downloader.get_artist_string(album_artists), + "artist": self.downloader.get_artist_string(track_artists), + "compilation": False, "composer": ( self.downloader.get_artist_string(composers) if composers else None ), - "copyright": next( - (i["text"] for i in album_metadata["copyrights"] if i["type"] == "P"), - None, - ), - "disc": int(disc), - "disc_total": int(album_metadata["tracks"]["items"][-1]["disc_number"]), - "isrc": external_ids.get("isrc") if external_ids is not None else None, - "label": album_metadata.get("label"), + "copyright": copyright_text, + "disc": int(disc_number), + "disc_total": int(disc_total), + "isrc": isrc, + "label": label, "lyrics": lyrics_unsynced, "media_type": "Song", "producer": ( self.downloader.get_artist_string(producers) if producers else None ), - "rating": "Explicit" if track_metadata.get("explicit") else "Unknown", + "rating": "Explicit" if track_info.get("contentRating", {}).get("label") == "EXPLICIT" else "Unknown", "release_date": self.downloader.get_release_date_tag( release_date_datetime_obj ), "release_year": str(release_date_datetime_obj.year), - "title": track_metadata["name"], - "track": int( - next( - ( - track["track_number"] - for track in album_metadata["tracks"]["items"] - if self.downloader.get_media_id(track) == track_id - ), - ) - ), - "track_total": int( - max( - track["track_number"] - for track in album_metadata["tracks"]["items"] - if track["disc_number"] == disc - ) - ), - "url": external_urls["spotify"], + "title": track_info.get("name"), + "track": int(track_info.get("trackNumber", 1)), + "track_total": int(track_total), + "url": external_url, } return tags @@ -140,54 +177,107 @@ class DownloaderSong(DownloaderAudio): self.downloader.cleanup_temp_path() def _download( - self, - track_id: str, - track_metadata: dict = None, - album_metadata: dict = None, - gid_metadata: dict = None, - stream_info: StreamInfoAudio = None, - playlist_metadata: dict = None, - playlist_track: int = None, - decryption_key: bytes = None, + self, + track_id: str, + track_metadata: dict = None, + album_metadata: dict = None, + gid_metadata: dict = None, + stream_info: StreamInfoAudio = None, + playlist_metadata: dict = None, + product_name: dict = None, + playlist_track: int = None, + decryption_key: bytes = None, ): - if not track_metadata: + if not album_metadata: logger.debug("Getting track metadata") track_metadata = self.downloader.spotify_api.get_track(track_id) + + try: + current_date = track_metadata['data']['trackUnion']['albumOfTrack'].get('release_date') + + if current_date == '1970-01-01': + fresh_data = self.downloader.spotify_api.get_track(track_id) + + real_date = None + real_precision = 'day' + real_album_name = None + + if 'album' in fresh_data: + real_date = fresh_data['album']['release_date'] + real_precision = fresh_data['album'].get('release_date_precision', 'day') + real_album_name = fresh_data['album']['name'] + + elif 'data' in fresh_data: + src_album = fresh_data['data']['trackUnion']['albumOfTrack'] + if 'date' in src_album and 'isoString' in src_album['date']: + real_date = src_album['date']['isoString'] + real_precision = 'day' + + if real_date and not str(real_date).startswith('1970'): + track_metadata['data']['trackUnion']['albumOfTrack']['release_date'] = real_date + track_metadata['data']['trackUnion']['albumOfTrack']['release_date_precision'] = real_precision + + if not album_metadata: + album_metadata = {} + + album_metadata['release_date'] = real_date + album_metadata['release_date_precision'] = real_precision + if real_album_name: + album_metadata['name'] = real_album_name + + except Exception as e: + pass + if not album_metadata: logger.debug("Getting album metadata") - album_metadata = self.downloader.spotify_api.get_album( - track_metadata["album"]["id"] - ) + try: + alb_node = track_metadata['data']['trackUnion']['albumOfTrack'] + album_id = alb_node['id'] + album_metadata = self.downloader.spotify_api.get_album(album_id) + except (KeyError, TypeError): + pass + if not gid_metadata: logger.debug("Getting GID metadata") - gid_metadata = self.downloader.get_gid_metadata(track_id, "track") + try: + gid_metadata = self.downloader.get_gid_metadata(track_id, "track", product_name) + except TypeError: + gid_metadata = self.downloader.get_gid_metadata(track_id, "track") + if gid_metadata.get("original_video"): logger.warning("Track is a music video, skipping.") return + if not stream_info: logger.debug("Getting stream info") - stream_info = self.get_stream_info(gid_metadata, "track") + stream_info = self.get_stream_info(gid_metadata, product_name, "track") + if not stream_info.file_id: logger.warning( "Track is not available on Spotify's " "servers and no alternative found, skipping" ) return + if stream_info.quality != self.audio_quality: - logger.warning(f"Quality has been changed to {stream_info.quality.value}") - if gid_metadata.get("has_lyrics"): + logger.warning(f"Quality has been changed to {stream_info.quality}") + + if gid_metadata: logger.debug("Getting lyrics") lyrics = self.get_lyrics(track_id) else: lyrics = Lyrics() + logger.debug("Getting track credits") track_credits = self.downloader.spotify_api.get_track_credits(track_id) + tags = self.get_tags( track_metadata, album_metadata, track_credits, lyrics.unsynced, ) + if playlist_metadata: tags = { **tags, @@ -196,6 +286,7 @@ class DownloaderSong(DownloaderAudio): playlist_track, ), } + file_extension = self.get_file_extension() final_path = self.downloader.get_final_path( "track", @@ -205,11 +296,13 @@ class DownloaderSong(DownloaderAudio): lrc_path = self.downloader.get_lrc_path(final_path) cover_path = self.get_cover_path(final_path) cover_url = self.downloader.get_cover_url( - album_metadata, + track_metadata, COVER_SIZE_X_KEY_MAPPING_SONG, ) + decrypted_path = None remuxed_path = None + if self.lrc_only: pass elif final_path.exists() and not self.downloader.overwrite: @@ -244,18 +337,21 @@ class DownloaderSong(DownloaderAudio): decrypted_path, remuxed_path, ) - if self.no_lrc or not lyrics.synced: + + if self.no_lrc: pass elif lrc_path.exists() and not self.downloader.overwrite: logger.debug(f'Synced lyrics already exists at "{lrc_path}", skipping') else: logger.debug(f'Saving synced lyrics to "{lrc_path}"') - self.downloader.save_lrc(lrc_path, lyrics.synced) + lyrics = None + media_temp_path = ( remuxed_path if remuxed_path is not None and remuxed_path.exists() else decrypted_path ) + self.downloader._final_processing( cover_path, cover_url,