98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
from google.cloud import firestore
|
|
|
|
import datetime
|
|
import logging
|
|
|
|
from spotframework.engine.playlistengine import PlaylistEngine, PlaylistSource, RecommendationSource, LibraryTrackSource
|
|
from spotframework.engine.processor.shuffle import Shuffle
|
|
from spotframework.engine.processor.sort import SortReleaseDate
|
|
from spotframework.engine.processor.deduplicate import DeduplicateByID
|
|
|
|
from spotframework.model.uri import Uri
|
|
|
|
import music.db.database as database
|
|
from music.db.part_generator import PartGenerator
|
|
|
|
db = firestore.Client()
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def run_user_playlist(username, playlist_name):
|
|
|
|
users = database.get_user_query_stream(username)
|
|
|
|
logger.info(f'running {username} / {playlist_name}')
|
|
|
|
if len(users) == 1:
|
|
|
|
playlist_collection = db.collection(u'spotify_users', u'{}'.format(users[0].id), 'playlists')
|
|
|
|
playlists = [i for i in playlist_collection.where(u'name', u'==', playlist_name).stream()]
|
|
|
|
if len(playlists) == 1:
|
|
|
|
playlist_dict = playlists[0].to_dict()
|
|
|
|
if playlist_dict['uri'] is None:
|
|
logger.critical(f'no playlist id to populate ({username}/{playlist_name})')
|
|
return None
|
|
|
|
if len(playlist_dict['parts']) == 0 and len(playlist_dict['playlist_references']) == 0:
|
|
logger.critical(f'no playlists to use for creation ({username}/{playlist_name})')
|
|
return None
|
|
|
|
net = database.get_authed_spotify_network(username)
|
|
|
|
engine = PlaylistEngine(net)
|
|
|
|
processors = [DeduplicateByID()]
|
|
|
|
if playlist_dict['shuffle'] is True:
|
|
processors.append(Shuffle())
|
|
else:
|
|
processors.append(SortReleaseDate(reverse=True))
|
|
|
|
part_generator = PartGenerator(user_id=users[0].id)
|
|
submit_parts = part_generator.get_recursive_parts(playlist_dict['name'])
|
|
|
|
params = [
|
|
PlaylistSource.Params(names=submit_parts)
|
|
]
|
|
|
|
if playlist_dict['include_recommendations']:
|
|
params.append(RecommendationSource.Params(recommendation_limit=playlist_dict['recommendation_sample']))
|
|
|
|
if playlist_dict.get('include_library_tracks', False):
|
|
params.append(LibraryTrackSource.Params())
|
|
|
|
if playlist_dict['type'] == 'recents':
|
|
boundary_date = datetime.datetime.now(datetime.timezone.utc) - \
|
|
datetime.timedelta(days=int(playlist_dict['day_boundary']))
|
|
tracks = engine.get_recent_playlist(params=params,
|
|
processors=processors,
|
|
boundary_date=boundary_date,
|
|
add_this_month=playlist_dict.get('add_this_month', False),
|
|
add_last_month=playlist_dict.get('add_last_month', False))
|
|
else:
|
|
tracks = engine.make_playlist(params=params,
|
|
processors=processors)
|
|
|
|
engine.execute_playlist(tracks, Uri(playlist_dict['uri']))
|
|
|
|
overwrite = playlist_dict.get('description_overwrite', None)
|
|
suffix = playlist_dict.get('description_suffix', None)
|
|
|
|
engine.change_description(sorted(submit_parts),
|
|
uri=Uri(playlist_dict['uri']),
|
|
overwrite=overwrite,
|
|
suffix=suffix)
|
|
|
|
else:
|
|
logger.critical(f'multiple/no playlists found ({username}/{playlist_name})')
|
|
return None
|
|
|
|
else:
|
|
logger.critical(f'multiple/no user(s) found ({username}/{playlist_name})')
|
|
return None
|