import asyncio from http import HTTPStatus from server.client import services from server.constants import DSP from server.constants.playlists.meta import PlaylistMetaApolloInclude from server.utils.glom import glom from server.utils.processor import process_by_item from server.utils.processor.playlists.playlists import playlist_basic_processors async def get_meta_playlists_info( playlist_list_id: list, dsp: DSP, country_code: str = None, apollo_include: list = tuple() ): # it can happen for example if requested playlist is deleted fm spotify possible_error_status_codes = (HTTPStatus.NOT_FOUND,) meta_tasks = [] if dsp == DSP.APPLE: meta_tasks = [ services.dsp.get_public_playlists_info( playlist_id=playlist_id, apple_country_code=country_code, allowed_status_codes=possible_error_status_codes, ) for playlist_id in playlist_list_id ] elif dsp == DSP.SPOTIFY: meta_tasks = [ services.dsp.get_public_current_playlists_info( playlist_id=playlist_id, allowed_status_codes=possible_error_status_codes ) for playlist_id in playlist_list_id ] need_dates = PlaylistMetaApolloInclude.PLAYLIST_DATES_FULL.value in apollo_include if need_dates: dates_tasks = [ services.dsp.get_public_track_positions_playlists_dates( playlist_id=playlist_id, country_code=country_code, ) for playlist_id in playlist_list_id ] meta_tasks += dates_tasks meta_list = await asyncio.gather(*meta_tasks) if need_dates: playlist_length = len(playlist_list_id) date_list = meta_list[playlist_length:] meta_list = [ item for item in meta_list[:playlist_length] if not item.get("original", {}).get("status_code") in possible_error_status_codes ] for num, meta in enumerate(meta_list): meta_list[num]["playlist_dates"] = glom(date_list, f"{num}.items.0", default={}) process_by_item(meta_list, playlist_basic_processors(dsp=dsp)) return meta_list