from typing import Dict, Iterable, Optional from server.constants.distributors import DISTRIBUTORS from server.constants.playlists.tracks import AUDIO_FEATURES_MAPPING from server.utils.common import prefix_str from server.utils.mappers.core.getters import getter from server.utils.mappers.core.setters import setter def set_distributed_by_upc( item: dict, item_id: Optional[str] = None, data: Dict[str, str] = None, key_prefix: str = None, upc_path: str = None, set_is_sony: bool = True, ): data = data or {} distributor = data.get(getter(item, upc_path)) setter(item, prefix_str("distributed_by", prefix=key_prefix), distributor) if set_is_sony: setter(item, prefix_str("is_sony", prefix=key_prefix), distributor == DISTRIBUTORS.SME.value) def set_is_sony_by_upc( item: dict, item_id: Optional[str] = None, data: Iterable[str] = (), key_prefix: str = None, upc_path: str = None ): """Set is sony by upc We need default value item_id=None as a 2-nd positional argument (this leads to "data" default value =() on pos. 3) because set_is_sony_by_upc is used as a processor in a flow where item and item_id are passed as positional args by apply_processors function Though there are cases when item_id=None inside apply_processors (for example: case when some tracks in a tracklist, for some reason, have missing isrc(s) or isrc=None) which leads to (item,) instead of (item, item_id) args that are passed because None is a valid value for apply_processors function which leads to an error (AG-11003) """ setter(item, prefix_str("is_sony", prefix=key_prefix), getter(item, upc_path) in data) def set_audio_features(item: dict, item_id: Optional[str] = None, data: Dict[str, dict] = None, key_prefix: str = None): audio_features_item = (data or {}).get(item_id, {}) audio_data = {} for field_name in AUDIO_FEATURES_MAPPING.keys(): audio_data[field_name] = audio_features_item.get(field_name) setter(item, prefix_str("audio_features", prefix=key_prefix), audio_data)