from dataclasses import dataclass, asdict import flag from charts_feed_messages.constants import ChartName, Emoji, GLOBE, GLOBAL_MARKET @dataclass class FeedMessageChart: name: str class Message: """Full Basic Message with data field got from User Data Api""" def __init__(self, message=None): self._message = message class MessageData(Message): """meta and data fields from Basic Message """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._meta = self._get_meta() self._data = self._get_data() def _get_data(self): return self._message["data"] def _get_meta(self): return self._message["meta"] class MessageDataStructure(MessageData): """content nested data from Basic Message data field""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._content = self._get_content() def _get_content(self): return self._data["content"] class MessageMetaStructure(MessageDataStructure): """Meta nested information""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.dsp = self._get_dsp() self.country_code = self._get_country_code() self.type = self._get_type() def _get_dsp(self): return self._meta["dsp"] def _get_country_code(self): return self._meta["country_code"] def _get_type(self): return self._meta["type"] class MessageContentStructure(MessageDataStructure): """Content nested information""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._track = self._get_track() self.previous_position = self._get_previous_position() self.current_position = self._get_current_position() def _get_track(self): return self._content["track"] def _get_previous_position(self): return self._content["previous_position"] def _get_current_position(self): return self._content["current_position"] class MessageTrackStructure(MessageContentStructure): """Track nested structure""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.track_name = self._get_track_name() self.artists = self._get_artists_list() def _get_track_name(self): return self._track["name"] def _get_artists_list(self): return self._track["artists"] class FeedMessageFieldsFabricator(MessageTrackStructure, MessageMetaStructure): """Fabricator to create Feed Message fields based on Basic Message fields""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.chart_name = self._get_chart_name() self.flag = self._get_flag() self._body = { "starred_track_top_chart_entry": self._get_entry_message, "starred_track_top_chart_exit": self._get_exit_message, "starred_track_top_chart_move": self._get_move_message, } def _get_flag(self): return GLOBE if self.country_code == GLOBAL_MARKET else flag.flag(self.country_code) def get_chart(self): return asdict(FeedMessageChart(name=ChartName[self.dsp])) def _get_chart_name(self): return self.get_chart()["name"] def _get_artists(self): """As we have a list of artists we need to handle situation when there are more than one artist for the track""" artists = " & ".join(artist["name"] for artist in self.artists) if len(self.artists) > 1 else\ self.artists[0]["name"] return artists def _get_entry_message(self): return f"{Emoji.FIRE.value} {self._get_artists()} - {self.track_name} entered {self.chart_name} {self.flag} " \ f"at position {self.current_position}" def _get_exit_message(self): return f"{Emoji.EXCLAMATION_POINT.value} {self._get_artists()} - {self.track_name} exited {self.chart_name} " \ f"{self.flag} at position {self.previous_position}" def _get_diff(self): return self.current_position - self.previous_position @staticmethod def _get_diff_key(diff): return diff < 0 def _get_move_message(self): diff = self._get_diff() diff_key = self._get_diff_key(diff) moves = { True: f"{Emoji.FIRE.value} {self._get_artists()} - {self.track_name} moved up {abs(diff)} spots on " f"{self.chart_name} {self.flag} to position {self.current_position}", False: f"{Emoji.EXCLAMATION_POINT.value} {self._get_artists()} - {self.track_name} moved down " f"{abs(diff)} spots on {self.chart_name} {self.flag} to position {self.current_position}" } return moves[diff_key] def get_body(self): """Get feed message nested body information (nested str message about what exactly happened to a track in a particular chart based on entry/exit/move event) """ return self._body[self.type]()