from typing import Any, Dict, Set from apollo_playlists_messages.constants import EventCode from apollo_playlists_messages.utils.playlist_id import (PlaylistId, starred_id, top_country_codes) SINGLE_POSITION_EVENTS = [EventCode.ENTRY, EventCode.EXIT] EVENT_CODE_TO_POSITIONS = { EventCode.ENTRY: "current_position", EventCode.EXIT: "previous_position", EventCode.MAJOR_MOVE: ["current_position", "previous_position"] } def _get_event_position(event: Dict[str, Any], event_code: EventCode) -> Dict[str, int]: if event_code in SINGLE_POSITION_EVENTS: key = EVENT_CODE_TO_POSITIONS[event_code] return {key: event[key]} return {key: event[key] for key in EVENT_CODE_TO_POSITIONS[event_code]} def parse_event( event_code: EventCode, event: Dict[str, Any], dsp_to_isrc_to_track_id_to_playlist_id_to_meta: Dict[str, Dict[str, Dict[str, Dict[PlaylistId, dict]]]], playlist_id_to_top_codes: Dict[PlaylistId, Set[str]], playlist_starred_id_to_data: Dict[str, dict], track_id_to_data: Dict[str, dict], isrc_entered: Set[str], ignored_playlist_self_ids: Set[str] = None, ): self_id = event["playlist"]["id"] if ignored_playlist_self_ids and self_id in ignored_playlist_self_ids: return dsp, isrc, track_id, playlist_data = event["dsp"], event["track"]["isrc"], event["track"]["id"], event["playlist"] isrc_entered.add(isrc) track_id_to_data[track_id] = event["track"] top_codes = event["top_country_code"] for code in event["country_code"]: playlist_id = PlaylistId(dsp, self_id, code) playlist_starred_id = starred_id(playlist_id) dsp_to_isrc_to_track_id_to_playlist_id_to_meta[dsp][isrc][track_id][playlist_id] = { "event_id": event["id"], **_get_event_position(event, event_code), } playlist_id_to_top_codes[playlist_id] = top_country_codes(playlist_id, top_codes) playlist_starred_id_to_data[playlist_starred_id] = playlist_data