added playlist reordering and sorting by added date
This commit is contained in:
parent
5140892288
commit
ac410dfeb9
58
sort_playlist.py
Normal file
58
sort_playlist.py
Normal file
@ -0,0 +1,58 @@
|
||||
from spotframework.net.network import Network
|
||||
from spotframework.net.user import NetworkUser
|
||||
from spotframework.engine.playlistengine import PlaylistEngine
|
||||
|
||||
import os
|
||||
import logging
|
||||
import sys
|
||||
|
||||
|
||||
logger = logging.getLogger('spotframework')
|
||||
|
||||
log_format = '%(asctime)s %(levelname)s %(name)s - %(funcName)s - %(message)s'
|
||||
|
||||
file_handler = logging.FileHandler(".spot/resort_playlist.log")
|
||||
formatter = logging.Formatter(log_format)
|
||||
file_handler.setFormatter(formatter)
|
||||
|
||||
logger.addHandler(file_handler)
|
||||
|
||||
stream_log_format = '%(levelname)s %(name)s:%(funcName)s - %(message)s'
|
||||
stream_formatter = logging.Formatter(stream_log_format)
|
||||
|
||||
stream_handler = logging.StreamHandler()
|
||||
stream_handler.setFormatter(stream_formatter)
|
||||
|
||||
logger.addHandler(stream_handler)
|
||||
|
||||
|
||||
def go(playlist_name):
|
||||
|
||||
net = Network(NetworkUser(os.environ['SPOTCLIENT'],
|
||||
os.environ['SPOTSECRET'],
|
||||
os.environ['SPOTACCESS'],
|
||||
os.environ['SPOTREFRESH']))
|
||||
|
||||
engine = PlaylistEngine(net)
|
||||
engine.load_user_playlists()
|
||||
playlist = next((j for j in engine.playlists if j.name == playlist_name), None)
|
||||
|
||||
if playlist is not None:
|
||||
engine.get_playlist_tracks(playlist)
|
||||
engine.reorder_playlist_by_added_date(playlist_name)
|
||||
else:
|
||||
logger.error('playlist not found')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if len(sys.argv) > 1:
|
||||
name = sys.argv[1]
|
||||
if len(sys.argv) > 2:
|
||||
for i in sys.argv[2:]:
|
||||
name += ' ' + i
|
||||
go(sys.argv[1])
|
||||
else:
|
||||
name = input('enter playlist name: ')
|
||||
if name == '':
|
||||
exit()
|
||||
go(name)
|
@ -50,6 +50,13 @@ class PlaylistEngine:
|
||||
else:
|
||||
logger.error('error getting tracks')
|
||||
|
||||
def load_playlist_tracks(self, name: str):
|
||||
playlist = next((i for i in self.playlists if i.name == name), None)
|
||||
if playlist is not None:
|
||||
self.get_playlist_tracks(playlist)
|
||||
else:
|
||||
logger.error(f'playlist {name} not found')
|
||||
|
||||
def make_playlist(self,
|
||||
playlist_parts: List[str],
|
||||
processors: List[AbstractProcessor] = None,
|
||||
@ -128,6 +135,39 @@ class PlaylistEngine:
|
||||
include_recommendations=include_recommendations,
|
||||
recommendation_limit=recommendation_limit)
|
||||
|
||||
def reorder_playlist_by_added_date(self,
|
||||
name: str = None,
|
||||
playlistid: str = None,
|
||||
reverse: bool = False):
|
||||
if name is None and playlistid is None:
|
||||
logger.error('no playlist name or id provided')
|
||||
raise ValueError('no playlist name or id provided')
|
||||
|
||||
if name:
|
||||
playlist = next((i for i in self.playlists if i.name == name), None)
|
||||
else:
|
||||
playlist = next((i for i in self.playlists if i.spotify_id == playlistid), None)
|
||||
|
||||
if playlist is None:
|
||||
logger.error('playlist not found')
|
||||
return None
|
||||
|
||||
tracks_to_sort = list(playlist.tracks)
|
||||
for i in range(len(playlist)):
|
||||
counter_track = tracks_to_sort[0]
|
||||
for track in tracks_to_sort:
|
||||
if reverse is False:
|
||||
if counter_track.added_at > track.added_at:
|
||||
counter_track = track
|
||||
else:
|
||||
if counter_track.added_at < track.added_at:
|
||||
counter_track = track
|
||||
|
||||
self.net.reorder_playlist_tracks(playlist.playlist_id,
|
||||
i + tracks_to_sort.index(counter_track),
|
||||
1, i)
|
||||
tracks_to_sort.remove(counter_track)
|
||||
|
||||
def execute_playlist(self,
|
||||
tracks: List[SpotifyTrack],
|
||||
playlist_id: str) -> Optional[Response]:
|
||||
|
@ -1,7 +1,7 @@
|
||||
from abc import ABC
|
||||
from .abstract import AbstractProcessor
|
||||
from .abstract import AbstractProcessor, BatchSingleTypeAwareProcessor
|
||||
from typing import List
|
||||
from spotframework.model.track import Track
|
||||
from spotframework.model.track import Track, PlaylistTrack
|
||||
|
||||
|
||||
class BasicReversibleSort(AbstractProcessor, ABC):
|
||||
@ -24,3 +24,19 @@ class SortArtistName(BasicReversibleSort):
|
||||
def process(self, tracks: List[Track]) -> List[Track]:
|
||||
tracks.sort(key=lambda x: x.artists[0].name, reverse=self.reverse)
|
||||
return tracks
|
||||
|
||||
|
||||
class SortAddedDate(BatchSingleTypeAwareProcessor):
|
||||
|
||||
def __init__(self,
|
||||
names: List[str] = None,
|
||||
reverse: bool = False,
|
||||
append_malformed: bool = True):
|
||||
super().__init__(names=names,
|
||||
instance_check=PlaylistTrack,
|
||||
append_malformed=append_malformed)
|
||||
self.reverse = reverse
|
||||
|
||||
def process_batch(self, tracks: List[PlaylistTrack]) -> List[PlaylistTrack]:
|
||||
tracks.sort(key=lambda x: x.added_at, reverse=self.reverse)
|
||||
return tracks
|
||||
|
@ -46,9 +46,17 @@ class Playlist:
|
||||
self._tracks = tracks
|
||||
|
||||
def __str__(self):
|
||||
headers = ['name', 'album', 'artist', 'added at', 'popularity', 'uri']
|
||||
|
||||
prefix = f'\n==={self.name}===\n\n' if self.name is not None else ''
|
||||
|
||||
table = prefix + self.get_tracks_string() + '\n' + f'total: {len(self)}'
|
||||
|
||||
return table
|
||||
|
||||
def get_tracks_string(self):
|
||||
|
||||
rows = []
|
||||
headers = ['name', 'album', 'artist', 'added at', 'popularity', 'uri']
|
||||
for track in self.tracks:
|
||||
track_row = [track.name,
|
||||
track.album.name,
|
||||
@ -61,10 +69,6 @@ class Playlist:
|
||||
|
||||
table = tabulate(rows, headers=headers, showindex='always', tablefmt="fancy_grid")
|
||||
|
||||
prefix = f'\n==={self.name}===\n\n' if self.name is not None else ''
|
||||
|
||||
table = prefix + table + '\n' + f'total: {len(self)}'
|
||||
|
||||
return table
|
||||
|
||||
|
||||
@ -95,3 +99,13 @@ class SpotifyPlaylist(Playlist):
|
||||
self.collaborative = collaborative
|
||||
self.public = public
|
||||
self.ext_spotify = ext_spotify
|
||||
|
||||
def __str__(self):
|
||||
|
||||
prefix = f'\n==={self.name}===\n\n' if self.name is not None else ''
|
||||
prefix += f'id: {self.playlist_id}\n' if self.playlist_id is not None else ''
|
||||
prefix += f'uri: {self.uri}\n' if self.uri is not None else ''
|
||||
|
||||
table = prefix + self.get_tracks_string() + '\n' + f'total: {len(self)}'
|
||||
|
||||
return table
|
||||
|
@ -456,3 +456,32 @@ class Network:
|
||||
|
||||
else:
|
||||
logger.error('playlist has no id')
|
||||
|
||||
def reorder_playlist_tracks(self,
|
||||
playlistid: str,
|
||||
range_start: int,
|
||||
range_length: int,
|
||||
insert_before: int) -> Optional[Response]:
|
||||
|
||||
logger.info(f'id: {playlistid}')
|
||||
|
||||
if range_start < 0:
|
||||
logger.error('range_start must be positive')
|
||||
raise ValueError('range_start must be positive')
|
||||
if range_length < 0:
|
||||
logger.error('range_length must be positive')
|
||||
raise ValueError('range_length must be positive')
|
||||
if insert_before < 0:
|
||||
logger.error('insert_before must be positive')
|
||||
raise ValueError('insert_before must be positive')
|
||||
|
||||
json = {'range_start': range_start,
|
||||
'range_length': range_length,
|
||||
'insert_before': insert_before}
|
||||
|
||||
resp = self._make_put_request('reorderPlaylistTracks', f'playlists/{playlistid}/tracks', json=json)
|
||||
|
||||
if resp:
|
||||
return resp
|
||||
else:
|
||||
logger.error('error reordering playlist')
|
||||
|
Loading…
Reference in New Issue
Block a user