from apollo_utils.core.constants.dsp import DSP_SPOTIFY_APPLE from apollo_utils.service.schemas.fields.enum import EnumField from marshmallow import fields, post_dump from apollo_messages_views.config import MESSAGE_PREFIX from apollo_messages_views.constants.base import Emoji, UNKNOWN_FLAG from apollo_messages_views.constants.charts import IS_NEGATIVE_DIFF_TO_MOVE_BODY, ChartName from apollo_messages_views.schemas.base import BaseMessage, MetaExcludeUnknown, TrackItemWithArtistNames, data_mixin from apollo_messages_views.utils.misc import get_artists_str, get_flag from apollo_messages_views.utils.charts import get_chart_update_body class ChartItem(MetaExcludeUnknown): name = fields.Function(lambda x: ChartName[x["dsp"]]) dsp = EnumField(DSP_SPOTIFY_APPLE, load_as="value", load_only=True, required=True) class ContentItem(MetaExcludeUnknown): track = fields.Nested(TrackItemWithArtistNames, required=True) chart = fields.Nested(ChartItem, required=True) current_position = fields.Integer(missing=None) previous_position = fields.Integer(missing=None) title = fields.Str(missing="") class ChartsFullMessage(data_mixin(ContentItem), BaseMessage): pass class ChartUpdateContentItem(MetaExcludeUnknown): chart = fields.Nested(ChartItem, required=True) title = fields.Str(missing="") class ChartsEntryMessage(ChartsFullMessage): @post_dump def set_body(self, in_data, **kwargs): content = in_data["data"]["content"] body = ( f"{Emoji.FIRE.value} {get_artists_str(content['track']['artists'])} - {content['track']['name']} entered " f"{content['chart']['name']} {get_flag(in_data['meta']['country_code'])} at position " f"{content['current_position']}" ) content["body"] = MESSAGE_PREFIX + body return in_data class ChartsExitMessage(ChartsFullMessage): @post_dump def set_body(self, in_data, **kwargs): content = in_data["data"]["content"] body = ( f"{Emoji.EXCLAMATION_POINT.value} {get_artists_str(content['track']['artists'])} - " f"{content['track']['name']} exited {content['chart']['name']} " f"{get_flag(in_data['meta']['country_code'])} at position {content['previous_position']}" ) content["body"] = MESSAGE_PREFIX + body return in_data class ChartsMoveMessage(ChartsFullMessage): @post_dump def set_body(self, in_data, **kwargs): content = in_data["data"]["content"] diff = content["current_position"] - content["previous_position"] body = IS_NEGATIVE_DIFF_TO_MOVE_BODY[diff < 0].format( artists=get_artists_str(content["track"]["artists"]), track_name=content["track"]["name"], abs_diff=abs(diff), chart_name=content["chart"]["name"], flag=get_flag(in_data["meta"]["country_code"]), position=content["current_position"], ) content["body"] = MESSAGE_PREFIX + body return in_data class ChartsUpdateMessage(data_mixin(ChartUpdateContentItem), BaseMessage): @post_dump def set_body(self, in_data, **kwargs): in_data["data"]["content"]["body"] = get_chart_update_body( country_code=in_data["meta"]["country_code"], content=in_data["data"]["content"], unknown_flag_as=UNKNOWN_FLAG, ) return in_data