from apollo_utils.core.constants.dsp import DSP from typing import Any, Callable, Dict, List, Optional, Tuple, Union from server.utils.mappers.core.getters import DEFAULT_NONE_VALUES, getter from server.utils.mappers.core.mappers import mapper from server.utils.mappers.misc import map_id, map_id_not_none from server.utils.mappers.playlists.tracks.base import map_positions, map_trend_metrics from server.utils.processor.core import Processor, processor def collector( item: dict, key: str, storage: List[Any], f: Callable = None, none_values: Tuple[Any] = DEFAULT_NONE_VALUES, none_result: Any = None, skip_none: bool = False, ): """ Function to be used as processor for collecting any data from items in a collection without extra traversal. @param item: item to get data from @param key: key to get data by @param storage: list to collect tha data taken from 'item' by 'key' @param f: function to apply to value before saving to storage @param none_values: values considered as None @param none_result: result to be returned for nullable values @param skip_none: should nullable values be saved to storage or not """ v = getter(item, key=key, none_values=none_values, none_result=none_result, f=f) if skip_none and v == none_result: return storage.append(v) def track_in_playlist_basic_processors( dsp: DSP, id_to_int: bool = False, # used for APPLE playlist_map: Optional[Dict[str, dict]] = None, ) -> List[Union[Callable, Processor]]: """Get basic processors for track-in-playlist items.""" return [ map_positions, processor(map_trend_metrics, playlist_map=playlist_map), processor(map_id_not_none, keys=("dsp_track_id", "track_id"), key_to="track_id", to_int=id_to_int), processor(map_id, key="playlist_id", remove_prefix=True), processor(mapper, key=f"{dsp.value.lower()}_track", key_to="track", remove_key=True), ]