spotify_actions/examples/label_playlist.py

# pylint: disable=missing-module-docstring, missing-function-docstring

from argparse import ArgumentParser
from datetime import date, datetime, timedelta
from typing import Iterable

from spotipy import Spotify

from spotify_actions.album import (
    album_filter_label,
    album_filter_release,
    album_from_ids,
    album_sort_release,
    album_to_tracks,
)
from spotify_actions.combinator import combinator_join
from spotify_actions.playlist import (
    playlist_current_user_all,
    playlist_current_user_assure,
    playlist_replace,
    playlist_tracks,
)
from spotify_actions.search import Query, search_albums
from spotify_actions.track import track_unique_albums
from spotify_actions.util import read_credentials_oauth


def label_playlist(client: Spotify, label_name: str, playlist_id: str) -> None:
    # Given a label name, replace all songs in the provided `playlist_id` with the
    # label's songs, ordered by descending release date.

    albums_search = search_albums(client, Query(label=label_name))
    albums_unfiltered = album_from_ids(client, albums_search)
    albums_unsorted = album_filter_label(albums_unfiltered, label_name)
    albums = album_sort_release(albums_unsorted, descending=True)
    tracks = album_to_tracks(client, albums)

    playlist_replace(client, playlist_id, tracks)


def label_recent(client: Spotify, label_playlist_ids: Iterable[str], playlist_id: str, released_after: date) -> None:

    album_iterables = []
    for label_playlist_id in label_playlist_ids:
        # Get all albums in a playlist released after the provided date
        tracks = playlist_tracks(client, [label_playlist_id])
        album_ids = track_unique_albums(tracks)
        albums = album_from_ids(client, album_ids)

        # Because the playlists were created in descending release date order,
        # `is_sorted=True` is enabled to reduce the number of API queries needed
        album_iterables.append(album_filter_release(albums, released_after, is_sorted=True))

    # Merge all the albums from each label playlist
    recent_albums = combinator_join(*album_iterables)
    recent_tracks = album_to_tracks(client, recent_albums)

    # Create the recent releases playlist
    playlist_replace(client, playlist_id, recent_tracks)


def main() -> None:
    # Intentionally 6 days - if running on a Friday, we don't want to include last Friday's releases
    one_week_ago = (datetime.now().date() - timedelta(days=6)).strftime("%Y-%m-%d")

    parser = ArgumentParser()
    parser.add_argument("-c", "--credentials", required=True)
    parser.add_argument("-r", "--redirect-uri", required=True)
    parser.add_argument("--recent-release", help='Name of the "recent releases" playlist constructed from all labels.')
    parser.add_argument(
        "--released-after", help="YYYY-MM-DD date that albums must be released after", default=one_week_ago
    )
    parser.add_argument("label", nargs="+")

    cmdline = parser.parse_args()

    client = read_credentials_oauth(
        cmdline.credentials,
        redirect_uri=cmdline.redirect_uri,
        scopes=["playlist-read-private", "playlist-modify-private", "playlist-modify-public"],
    )

    # Get all user playlists; we'll be iterating over this a couple times
    user_playlists = list(playlist_current_user_all(client))

    # To simplify, this assumes that the label playlist name is unique for this user
    def _locate_playlist(name: str) -> str:
        assured = playlist_current_user_assure(client, user_playlists, name)
        # The `str()` wrapper is technically unnecessary, but keeps mypy happy
        return str(list(assured)[0].spotify_id)

    label_ids = {name: _locate_playlist(name) for name in cmdline.label}

    for label_name, playlist_id in label_ids.items():
        label_playlist(client, label_name, playlist_id)

    if cmdline.recent_release:
        recent_release = _locate_playlist(cmdline.recent_release)
        released_after = datetime.strptime(cmdline.released_after, "%Y-%m-%d").date()
        label_recent(client, label_ids.values(), recent_release, released_after)


if __name__ == "__main__":
    main()