mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-14 21:38:38 +00:00
gitea
b2aab28460
Fix #1 Test fix 2 Fixing part 2 Fixing ctnd Final fix Fixing spotipy get bug Same plusy kurwa Fixing fucking spotipy test More logging
264 lines
10 KiB
Python
264 lines
10 KiB
Python
import sys
|
|
from spotify_dl.scaffold import log
|
|
from spotify_dl.utils import sanitize
|
|
from rich.progress import Progress
|
|
|
|
|
|
def fetch_tracks(sp, item_type, url):
|
|
"""
|
|
Fetches tracks from the provided URL.
|
|
:param sp: Spotify client
|
|
:param item_type: Type of item being requested for: album/playlist/track
|
|
:param url: URL of the item
|
|
:return Dictionary of song and artist
|
|
"""
|
|
songs_list = []
|
|
offset = 0
|
|
songs_fetched = 0
|
|
|
|
if item_type == "playlist":
|
|
with Progress() as progress:
|
|
songs_task = progress.add_task(description="Fetching songs from playlist..")
|
|
while True:
|
|
items = sp.playlist_items(
|
|
playlist_id=url,
|
|
fields="items.track.name,items.track.artists(name, uri),"
|
|
"items.track.album(name, release_date, total_tracks, images),"
|
|
"items.track.track_number,total, next,offset,"
|
|
"items.track.id",
|
|
additional_types=["track"],
|
|
offset=offset,
|
|
)
|
|
total_songs = items.get("total")
|
|
track_info_task = progress.add_task(
|
|
description="Fetching track info", total=len(items["items"])
|
|
)
|
|
for item in items["items"]:
|
|
track_info = item.get("track")
|
|
# If the user has a podcast in their playlist, there will be no track
|
|
# Without this conditional, the program will fail later on when the metadata is fetched
|
|
if track_info is None:
|
|
offset += 1
|
|
continue
|
|
track_album_info = track_info.get("album")
|
|
track_num = track_info.get("track_number")
|
|
spotify_id = track_info.get("id")
|
|
track_name = track_info.get("name")
|
|
track_artist = ", ".join(
|
|
[artist["name"] for artist in track_info.get("artists")]
|
|
)
|
|
if track_album_info:
|
|
track_album = track_album_info.get("name")
|
|
track_year = (
|
|
track_album_info.get("release_date")[:4]
|
|
if track_album_info.get("release_date")
|
|
else ""
|
|
)
|
|
album_total = track_album_info.get("total_tracks")
|
|
if len(item["track"]["album"]["images"]) > 0:
|
|
cover = item["track"]["album"]["images"][0]["url"]
|
|
else:
|
|
cover = None
|
|
|
|
artists = track_info.get("artists")
|
|
main_artist_id = (
|
|
artists[0].get("uri", None) if len(artists) > 0 else None
|
|
)
|
|
genres = (
|
|
sp.artist(artist_id=main_artist_id).get("genres", [])
|
|
if main_artist_id
|
|
else []
|
|
)
|
|
if len(genres) > 0:
|
|
genre = genres[0]
|
|
else:
|
|
genre = ""
|
|
songs_list.append(
|
|
{
|
|
"name": track_name,
|
|
"artist": track_artist,
|
|
"album": track_album,
|
|
"year": track_year,
|
|
"num_tracks": album_total,
|
|
"num": track_num,
|
|
"playlist_num": offset + 1,
|
|
"cover": cover,
|
|
"genre": genre,
|
|
"spotify_id": spotify_id,
|
|
"track_url": None,
|
|
}
|
|
)
|
|
offset += 1
|
|
songs_fetched += 1
|
|
progress.update(
|
|
task_id=track_info_task,
|
|
description=f"Fetching track info for \n{track_name}",
|
|
advance=1,
|
|
)
|
|
|
|
progress.update(
|
|
task_id=songs_task,
|
|
description=f"Fetched {songs_fetched} of {total_songs} songs from the playlist",
|
|
advance=100,
|
|
total=total_songs,
|
|
)
|
|
if total_songs == offset:
|
|
break
|
|
|
|
elif item_type == "album":
|
|
with Progress() as progress:
|
|
album_songs_task = progress.add_task(
|
|
description="Fetching songs from the album.."
|
|
)
|
|
while True:
|
|
album_info = sp.album(album_id=url)
|
|
items = sp.album_tracks(album_id=url, offset=offset)
|
|
total_songs = items.get("total")
|
|
track_album = album_info.get("name")
|
|
track_year = (
|
|
album_info.get("release_date")[:4]
|
|
if album_info.get("release_date")
|
|
else ""
|
|
)
|
|
album_total = album_info.get("total_tracks")
|
|
if len(album_info["images"]) > 0:
|
|
cover = album_info["images"][0]["url"]
|
|
else:
|
|
cover = None
|
|
if (
|
|
len(sp.artist(artist_id=album_info["artists"][0]["uri"])["genres"])
|
|
> 0
|
|
):
|
|
genre = sp.artist(artist_id=album_info["artists"][0]["uri"])[
|
|
"genres"
|
|
][0]
|
|
else:
|
|
genre = ""
|
|
for item in items["items"]:
|
|
track_name = item.get("name")
|
|
track_artist = ", ".join(
|
|
[artist["name"] for artist in item["artists"]]
|
|
)
|
|
track_num = item["track_number"]
|
|
spotify_id = item.get("id")
|
|
songs_list.append(
|
|
{
|
|
"name": track_name,
|
|
"artist": track_artist,
|
|
"album": track_album,
|
|
"year": track_year,
|
|
"num_tracks": album_total,
|
|
"num": track_num,
|
|
"track_url": None,
|
|
"playlist_num": offset + 1,
|
|
"cover": cover,
|
|
"genre": genre,
|
|
"spotify_id": spotify_id,
|
|
}
|
|
)
|
|
offset += 1
|
|
|
|
progress.update(
|
|
task_id=album_songs_task,
|
|
description=f"Fetched {offset} of {album_total} songs from the album {track_album}",
|
|
advance=offset,
|
|
total=album_total,
|
|
)
|
|
if album_total == offset:
|
|
break
|
|
|
|
elif item_type == "track":
|
|
items = sp.track(track_id=url)
|
|
track_name = items.get("name")
|
|
album_info = items.get("album")
|
|
track_artist = ", ".join([artist["name"] for artist in items["artists"]])
|
|
if album_info:
|
|
track_album = album_info.get("name")
|
|
track_year = (
|
|
album_info.get("release_date")[:4]
|
|
if album_info.get("release_date")
|
|
else ""
|
|
)
|
|
album_total = album_info.get("total_tracks")
|
|
track_num = items["track_number"]
|
|
spotify_id = items["id"]
|
|
if len(items["album"]["images"]) > 0:
|
|
cover = items["album"]["images"][0]["url"]
|
|
else:
|
|
cover = None
|
|
if len(sp.artist(artist_id=items["artists"][0]["uri"])["genres"]) > 0:
|
|
genre = sp.artist(artist_id=items["artists"][0]["uri"])["genres"][0]
|
|
else:
|
|
genre = ""
|
|
songs_list.append(
|
|
{
|
|
"name": track_name,
|
|
"artist": track_artist,
|
|
"album": track_album,
|
|
"year": track_year,
|
|
"num_tracks": album_total,
|
|
"num": track_num,
|
|
"playlist_num": offset + 1,
|
|
"cover": cover,
|
|
"genre": genre,
|
|
"track_url": None,
|
|
"spotify_id": spotify_id,
|
|
}
|
|
)
|
|
|
|
return songs_list
|
|
|
|
|
|
def parse_spotify_url(url):
|
|
"""
|
|
Parse the provided Spotify playlist URL and determine if it is a playlist, track or album.
|
|
:param url: URL to be parsed
|
|
|
|
:return tuple indicating the type and id of the item
|
|
"""
|
|
if url.startswith("spotify:"):
|
|
log.error("Spotify URI was provided instead of a playlist/album/track URL.")
|
|
sys.exit(1)
|
|
parsed_url = url.replace("https://open.spotify.com/", "").split("?")[0]
|
|
item_type = parsed_url.split("/")[0]
|
|
item_id = parsed_url.split("/")[1]
|
|
return item_type, item_id
|
|
|
|
|
|
def get_item_name(sp, item_type, item_id):
|
|
"""
|
|
Fetch the name of the item.
|
|
:param sp: Spotify Client
|
|
:param item_type: Type of the item
|
|
:param item_id: id of the item
|
|
:return String indicating the name of the item
|
|
"""
|
|
if item_type == "playlist":
|
|
name = sp.playlist(playlist_id=item_id, fields="name").get("name")
|
|
elif item_type == "album":
|
|
name = sp.album(album_id=item_id).get("name")
|
|
elif item_type == "track":
|
|
name = sp.track(track_id=item_id).get("name")
|
|
elif item_type == "artist_top_tracks":
|
|
name = sp.artist_top_tracks(artist_id=item_id).get("name")
|
|
return sanitize(name)
|
|
|
|
|
|
def validate_spotify_urls(urls):
|
|
"""
|
|
Validate the URLs and determine if the item types they have are supported
|
|
: return a list of valid urls
|
|
"""
|
|
valid_urls = []
|
|
for url in urls:
|
|
item_type, item_id = parse_spotify_url(url)
|
|
log.debug("Got item_type %s and item_id %s", item_type, item_id)
|
|
if item_type not in ["album", "track", "playlist"]:
|
|
log.info("Only albums/tracks/playlists are supported and not %s", item_type)
|
|
continue
|
|
if item_id is None:
|
|
log.info("Couln't get valid item_id for %s", url)
|
|
else:
|
|
valid_urls.append(url)
|
|
return valid_urls
|