import uuid from typing import Optional, Set from apollo_notifications.utils import dump_datetime # 'track in playlist' related events def get_track_in_playlist_push_key( date: str, playlist_id: str, user_id: str, topic: str, market: Optional[str], vendor: str, isrc: str, delimiter="/", prefix="playlist") -> uuid.UUID: """Returns UUID from unique set of arguments, which is used as message id.""" return uuid.uuid5(uuid.NAMESPACE_DNS, delimiter.join( [prefix, date, str(playlist_id), user_id, topic, market or '', vendor, isrc])) def build_track_in_playlist_push_message(track: object, message_template: str, country_flag: str) -> str: """Builds push message from tracks data.""" message = message_template.format( artist_name=track.artist_name, track_name=track.track_name, playlist_name=track.playlist_name, position=track.position, country_flag=country_flag ) return message.strip() def filter_track_in_playlist_event( item: object, date: str, topic: str, market: str, vendor: str, existing_messages: Set[str], *args, **kwargs): """Filter tracks to make push messages from.""" return str(get_track_in_playlist_push_key( date, item.playlist_id, item.user_id, topic, market, vendor, item.isrc) ) not in existing_messages # 'playlist' related events def get_playlist_update_push_key( date: str, playlist_id: str, user_id: str, topic: str, market: Optional[str], vendor: str, delimiter="/", prefix="" ) -> uuid.UUID: """Returns UUID from unique set of arguments, which is used as message id.""" return uuid.uuid5(uuid.NAMESPACE_DNS, delimiter.join( [prefix, date, str(playlist_id), user_id, topic, market or '', vendor])) def build_playlist_update_push_message(playlist: object, message_template: str, country_flag: str) -> str: """Builds push message from playlist data.""" message = message_template.format( playlist_name=playlist.playlist_name, country_flag=country_flag ) return message.strip() def filter_playlist_update_event( item: object, topic: str, market: Optional[str], vendor: str, existing_messages: Set[str], *args, **kwargs ) -> bool: """Filter playlists to make push messages from.""" return str( get_playlist_update_push_key( dump_datetime(item.date), item.playlist_id, item.user_id, topic, market, vendor ) ) not in existing_messages