89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
"""
|
|
Selectors for working with albums
|
|
"""
|
|
from datetime import date
|
|
from functools import partial
|
|
from typing import Iterable, Union
|
|
|
|
from spotify_model import Paging, SearchAlbum, SimplifiedAlbum, SimplifiedTrack
|
|
from spotipy import Spotify
|
|
|
|
from .util import chunk, exhaust, parse_release_date
|
|
|
|
|
|
def album_filter_label(albums: Iterable[SimplifiedAlbum], label: str) -> Iterable[SearchAlbum]:
|
|
"Filter albums that match an exact label string"
|
|
for album in albums:
|
|
if album.label == label:
|
|
yield album
|
|
|
|
|
|
def album_filter_release(
|
|
albums: Iterable[SimplifiedAlbum], released_after: date, is_sorted: bool = False, fast_forward: bool = False
|
|
) -> Iterable[SearchAlbum]:
|
|
"""
|
|
Filter albums to those released on or after a provided date.
|
|
|
|
If `is_sorted` is True, iteration will stop once the first album released prior to
|
|
`released_after` is encountered (may be useful to avoid extra API calls when
|
|
iterating over a pre-sorted playlist).
|
|
|
|
See `temporal_convert` for more information on how album release dates are
|
|
resolved, and usage of `fast_forward`.
|
|
"""
|
|
|
|
for album in albums:
|
|
effective_release = parse_release_date(
|
|
album.release_date, album.release_date_precision, fast_forward=fast_forward
|
|
)
|
|
|
|
if effective_release >= released_after:
|
|
yield album
|
|
elif is_sorted:
|
|
return
|
|
|
|
|
|
def album_from_ids(
|
|
client: Spotify, albums: Iterable[Union[str, SearchAlbum]], chunk_size: int = 20
|
|
) -> Iterable[SimplifiedAlbum]:
|
|
"""
|
|
Given a stream of album IDs (or base album objects), retrieve the full album objects
|
|
"""
|
|
|
|
def _to_id() -> Iterable[SimplifiedAlbum]:
|
|
for album in albums:
|
|
yield album if isinstance(album, str) else album.spotify_id
|
|
|
|
for album_id_chunk in chunk(_to_id(), chunk_size):
|
|
album_chunk = client.albums(album_id_chunk)
|
|
|
|
for album in album_chunk["albums"]:
|
|
yield SimplifiedAlbum(**album)
|
|
|
|
|
|
def album_sort_release(
|
|
albums: Iterable[SearchAlbum], descending: bool = False, fast_forward: bool = False
|
|
) -> Iterable[SearchAlbum]:
|
|
"Sort albums by release date"
|
|
all_albums = list(albums)
|
|
|
|
def _sort_key(album: SearchAlbum) -> date:
|
|
return parse_release_date(album.release_date, album.release_date_precision, fast_forward=fast_forward)
|
|
|
|
for album in sorted(all_albums, key=_sort_key, reverse=descending):
|
|
yield album
|
|
|
|
|
|
def album_to_tracks(client: Spotify, albums: Iterable[SimplifiedAlbum]) -> Iterable[SimplifiedTrack]:
|
|
"Convert an album stream to the tracks on that album"
|
|
|
|
def _album_tracks(album_id: str, limit: int, offset: int) -> Paging:
|
|
return Paging(**client.album_tracks(album_id, limit=limit, offset=offset))
|
|
|
|
# Because most album tracklists don't need to use paging, it's expected that API calls are relatively infrequent
|
|
for album in albums:
|
|
tracks_function = partial(_album_tracks, album_id=album.spotify_id)
|
|
|
|
for track in exhaust(tracks_function, initial=album.tracks):
|
|
yield SimplifiedTrack(**track)
|