from functools import partial from typing import Dict, Optional from server.domains.playlists.tracks import track_in_playlist_is_entry, track_in_playlist_trend_metric from server.utils.common import add1 from server.utils.mappers.core.getters import getter from server.utils.mappers.core.mappers import mappers_by_keys, mappers_by_map def map_positions(item: dict): mappers_by_keys(item, ("current", "earliest_position"), none_result=None, f=add1) def map_trend_metrics( item: dict, playlist_map: Optional[Dict[str, dict]] = None, playlist_id_field: str = "playlist_id" ): is_personalised = ( getter(playlist_map[item[playlist_id_field]], key="is_personalised", none_result=False) if playlist_map else getter(item, key="playlist.is_personalised", none_result=False) ) is_top_track_for_isrc = getter(item, key="is_top_track_for_isrc", none_result=False) item["is_entry"] = track_in_playlist_is_entry( num_days_on=getter(item, key="num_days_on", none_result=None, f=int), is_personalised=is_personalised, is_top_track_for_isrc=is_top_track_for_isrc, ) get_trend_metric = partial( track_in_playlist_trend_metric, is_personalised=is_personalised, is_top_track_for_isrc=is_top_track_for_isrc ) def get_previous_position(x: Optional[int]) -> Optional[int]: x = get_trend_metric(x) if x is not None: return add1(x) mappers_by_map( item, { "trend_change_14_days": {"key_to": "trend"}, "last_date_change_14_days": {"key_to": "change_position_date_time"}, "previous_position_change_14_days": {"key_to": "previous", "f": get_previous_position}, }, none_result=None, f=get_trend_metric, remove_key=True, )