This commit is contained in:
GladistonXD
2026-01-24 01:55:31 -03:00
committed by GitHub
parent 74799f7706
commit 5160be1c77
+179 -83
View File
@@ -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,