added get library tracks with support in engine and decorators

This commit is contained in:
aj 2019-09-16 02:23:50 +01:00
parent ef92b49ee6
commit 91ff0e1bce
4 changed files with 109 additions and 5 deletions

View File

@ -1,6 +1,7 @@
import requests
import os
import logging
import copy
import spotframework.util.monthstrings as monthstrings
from spotframework.engine.processor.added import AddedSince
@ -21,6 +22,7 @@ class PlaylistEngine:
def __init__(self, net: Network):
self.playlists = []
self.library_tracks = []
self.net = net
def load_user_playlists(self) -> None:
@ -41,6 +43,16 @@ class PlaylistEngine:
else:
logger.error('error getting playlists')
def load_library_tracks(self, track_limit: int = None) -> None:
logger.info('loading')
if track_limit:
tracks = self.net.get_library_tracks(response_limit=track_limit)
else:
tracks = self.net.get_library_tracks()
if tracks and len(tracks) > 0:
self.library_tracks = tracks
def get_playlist_tracks(self,
playlist: SpotifyPlaylist) -> None:
logger.info(f"pulling tracks for {playlist.name}")
@ -62,11 +74,17 @@ class PlaylistEngine:
playlist_parts: List[str],
processors: List[AbstractProcessor] = None,
include_recommendations: bool = False,
recommendation_limit: int = 10) -> List[SpotifyTrack]:
recommendation_limit: int = 10,
include_library_tracks: bool = False,
library_processors: List[AbstractProcessor] = None) -> List[SpotifyTrack]:
if processors is None:
processors = []
if library_processors is None:
library_processors = []
tracks = []
for part in playlist_parts:
@ -78,7 +96,7 @@ class PlaylistEngine:
if play.has_tracks() is False:
self.get_playlist_tracks(play)
playlist_tracks = list(play.tracks)
playlist_tracks = copy.deepcopy(play.tracks)
for processor in [i for i in processors if i.has_targets()]:
if play.name in [i for i in processor.playlist_names]:
@ -94,6 +112,13 @@ class PlaylistEngine:
for processor in [i for i in processors if i.has_targets() is False]:
tracks = processor.process(tracks)
if include_library_tracks:
library_tracks = copy.deepcopy(self.library_tracks)
for processor in library_processors:
library_tracks = processor.process(library_tracks)
tracks += library_tracks
if include_recommendations:
recommendations = self.net.get_recommendations(tracks=[i.uri.object_id for i in tracks],
response_limit=recommendation_limit)

View File

@ -87,6 +87,41 @@ class SpotifyTrack(Track):
f': {self.name}, ({self.album}), {self.artists}, {self.uri}'
class LibraryTrack(SpotifyTrack):
def __init__(self,
name: str,
album: Album,
artists: List[Artist],
href: str = None,
uri: Union[str, Uri] = None,
disc_number: int = None,
duration_ms: int = None,
explicit: bool = None,
is_playable: bool = None,
popularity: int = None,
added_at: datetime = None
):
super().__init__(name=name, album=album, artists=artists,
href=href,
uri=uri,
disc_number=disc_number,
duration_ms=duration_ms,
explicit=explicit,
is_playable=is_playable,
popularity=popularity)
self.added_at = added_at
def __repr__(self):
return Color.BOLD + Color.YELLOW + 'LibraryTrack' + Color.END + \
f': {self.name}, ({self.album}), {self.artists}, {self.uri}, {self.added_at}'
class PlaylistTrack(SpotifyTrack):
def __init__(self,
name: str,

View File

@ -10,7 +10,7 @@ from spotframework.model.user import User
from . import const
from spotframework.net.user import NetworkUser
from spotframework.model.playlist import SpotifyPlaylist
from spotframework.model.track import Track, SpotifyTrack, PlaylistTrack, PlayedTrack
from spotframework.model.track import Track, SpotifyTrack, PlaylistTrack, PlayedTrack, LibraryTrack
from spotframework.model.album import LibraryAlbum, SpotifyAlbum
from spotframework.model.service import CurrentlyPlaying, Device, Context
from spotframework.model.uri import Uri
@ -191,7 +191,7 @@ class Network:
logger.info(f"loading")
pager = PageCollection(net=self, url='me/albums', name='getLibrary')
pager = PageCollection(net=self, url='me/albums', name='getLibraryAlbums')
if response_limit:
pager.total_limit = response_limit
pager.iterate()
@ -200,6 +200,19 @@ class Network:
return return_items
def get_library_tracks(self, response_limit: int = None) -> Optional[List[LibraryAlbum]]:
logger.info(f"loading")
pager = PageCollection(net=self, url='me/tracks', name='getLibraryTracks')
if response_limit:
pager.total_limit = response_limit
pager.iterate()
return_items = [self.parse_track(i) for i in pager.items]
return return_items
def get_user_playlists(self) -> Optional[List[SpotifyPlaylist]]:
logger.info('retrieved')
@ -711,7 +724,7 @@ class Network:
if context:
context = self.parse_context(context)
if added_at or added_by or is_local:
if added_by or is_local:
return PlaylistTrack(name=name,
album=album,
artists=artists,
@ -729,6 +742,21 @@ class Network:
is_playable=is_playable,
popularity=popularity)
elif added_at:
return LibraryTrack(name=name,
album=album,
artists=artists,
href=href,
uri=uri,
disc_number=disc_number,
duration_ms=duration_ms,
explicit=explicit,
is_playable=is_playable,
popularity=popularity,
added_at=added_at)
elif played_at or context:
return PlayedTrack(name=name,
album=album,

View File

@ -0,0 +1,16 @@
import functools
import logging
logger = logging.getLogger(__name__)
def debug(func):
@functools.wraps(func)
def wrapper_debug(*args, **kwargs):
args_repr = [repr(a) for a in args]
kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
signature = ", ".join(args_repr + kwargs_repr)
print(f"{func.__name__}({signature})")
value = func(*args, **kwargs)
print(f"{func.__name__!r} -> {value!r}")
return value
return wrapper_debug