diff --git a/src/server/BreCal/notifications/notification_content_email.py b/src/server/BreCal/notifications/notification_content_email.py new file mode 100644 index 0000000..ce8d653 --- /dev/null +++ b/src/server/BreCal/notifications/notification_content_email.py @@ -0,0 +1,53 @@ +import pandas as pd +from server.BreCal.database.enums import ParticipantType, ShipcallType, StatusFlags + + + +#### Verbosity Functions #### +def get_default_header()->str: + # HEADER (greeting and default message) + header = "Dear Sir or Madam\n\nThank you for participating in the project 'Bremen Calling'. During analysis, our software has identified an event, which may be worth a second look. Here is the summary. \n\n" + return header + + +def get_default_footer()->str: + # FOOTER (signature) + footer = "\n\nWe would kindly ask you to have a look at the shipcall and verify, if any action is required from your side. \n\nKind regards\nThe 'Bremen Calling' Team" + return footer + + +def get_agency_name(sql_handler, times_df): + times_agency = times_df.loc[times_df["participant_type"]==ParticipantType.AGENCY.value,"participant_id"] + if len(times_agency)==0: + agency_name = "" + else: + agency_participant_id = times_agency.iloc[0] + agency_name = sql_handler.df_dict.get("participant").loc[agency_participant_id,"name"] + return agency_name + + +def get_ship_name(sql_handler, shipcall): + ship = sql_handler.df_dict.get("ship").loc[shipcall.ship_id] + ship_name = ship.loc["name"] # when calling ship.name, the ID is returned (pandas syntax) + return ship_name + + +def create_notification_body(sql_handler, times_df, shipcall, result)->str: + # #TODO: add 'Link zum Anlauf' + # URL: https://trello.com/c/qenZyJxR/75-als-bsmd-m%C3%B6chte-ich-%C3%BCber-gelbe-und-rote-ampeln-informiert-werden-um-die-systembeteiligung-zu-st%C3%A4rken + header = get_default_header() + footer = get_default_footer() + + agency_name = get_agency_name(sql_handler, times_df) + ship_name = get_ship_name(sql_handler, shipcall) + + verbosity_introduction = f"Respective Shipcall:\n" + traffic_state_verbosity = f"\tTraffic Light State: {StatusFlags(result[0]).name}\n" + ship_name_verbosity = f"\tShip: {ship_name} (the ship is {ShipcallType(shipcall.type).name.lower()})\n" + agency_name_verbosity = f"\tResponsible Agency: {agency_name}\n" + eta_verbosity = f"\tEstimated Arrival Time: {shipcall.eta.isoformat()}\n" if not pd.isna(shipcall.eta) else "" + etd_verbosity = f"\tEstimated Departure Time: {shipcall.etd.isoformat()}\n" if not pd.isna(shipcall.etd) else "" + error_verbosity = f"\nError Description:\n\t" + "\n\t".join(result[1]) + + message_body = "".join([header, verbosity_introduction, traffic_state_verbosity, ship_name_verbosity, agency_name_verbosity, eta_verbosity, etd_verbosity, error_verbosity, footer]) + return message_body \ No newline at end of file diff --git a/src/server/BreCal/notifications/notification_functions.py b/src/server/BreCal/notifications/notification_functions.py index 86956f0..78bd6bd 100644 --- a/src/server/BreCal/notifications/notification_functions.py +++ b/src/server/BreCal/notifications/notification_functions.py @@ -1,7 +1,6 @@ import datetime -import pandas as pd from BreCal.schemas.model import Notification -from BreCal.database.enums import NotificationType, ParticipantType, ShipcallType, StatusFlags +from BreCal.database.enums import NotificationType, StatusFlags def create_notification(id, times_id, message, level, notification_type:NotificationType, created=None, modified=None): created = (datetime.datetime.now()).isoformat() or created @@ -12,60 +11,12 @@ def create_notification(id, times_id, message, level, notification_type:Notifica ) return notification +class NotifierFunctions(): + """ + # #TODO_refactor: this legacy object needs strong refactoring and shall be simplified in its execution. - - - -#### Verbosity Functions #### - -def get_default_header()->str: - # HEADER (greeting and default message) - header = "Dear Sir or Madam\n\nThank you for participating in the project 'Bremen Calling'. During analysis, our software has identified an event, which may be worth a second look. Here is the summary. \n\n" - return header - -def get_default_footer()->str: - # FOOTER (signature) - footer = "\n\nWe would kindly ask you to have a look at the shipcall and verify, if any action is required from your side. \n\nKind regards\nThe 'Bremen Calling' Team" - return footer - -def get_agency_name(sql_handler, times_df): - times_agency = times_df.loc[times_df["participant_type"]==ParticipantType.AGENCY.value,"participant_id"] - if len(times_agency)==0: - agency_name = "" - else: - agency_participant_id = times_agency.iloc[0] - agency_name = sql_handler.df_dict.get("participant").loc[agency_participant_id,"name"] - return agency_name - -def get_ship_name(sql_handler, shipcall): - ship = sql_handler.df_dict.get("ship").loc[shipcall.ship_id] - ship_name = ship.loc["name"] # when calling ship.name, the ID is returned (pandas syntax) - return ship_name - - -def create_notification_body(sql_handler, times_df, shipcall, result)->str: - # #TODO: add 'Link zum Anlauf' - # URL: https://trello.com/c/qenZyJxR/75-als-bsmd-m%C3%B6chte-ich-%C3%BCber-gelbe-und-rote-ampeln-informiert-werden-um-die-systembeteiligung-zu-st%C3%A4rken - header = get_default_header() - footer = get_default_footer() - - agency_name = get_agency_name(sql_handler, times_df) - ship_name = get_ship_name(sql_handler, shipcall) - - verbosity_introduction = f"Respective Shipcall:\n" - traffic_state_verbosity = f"\tTraffic Light State: {StatusFlags(result[0]).name}\n" - ship_name_verbosity = f"\tShip: {ship_name} (the ship is {ShipcallType(shipcall.type).name.lower()})\n" - agency_name_verbosity = f"\tResponsible Agency: {agency_name}\n" - eta_verbosity = f"\tEstimated Arrival Time: {shipcall.eta.isoformat()}\n" if not pd.isna(shipcall.eta) else "" - etd_verbosity = f"\tEstimated Departure Time: {shipcall.etd.isoformat()}\n" if not pd.isna(shipcall.etd) else "" - error_verbosity = f"\nError Description:\n\t" + "\n\t".join(result[1]) - - message_body = "".join([header, verbosity_introduction, traffic_state_verbosity, ship_name_verbosity, agency_name_verbosity, eta_verbosity, etd_verbosity, error_verbosity, footer]) - return message_body - - -class Notifier(): - """An object that helps with the logic of selecting eligible shipcalls to create the correct notifications for the respective users.""" + An object that helps with the logic of selecting eligible shipcalls to create the correct notifications for the respective users. + """ def __init__(self)->None: pass @@ -103,7 +54,7 @@ class Notifier(): state_old = max(int(state_old), StatusFlags.GREEN.value) return int(state_new) > int(state_old) - def get_notification_times(self, evaluation_states_new)->list[datetime.datetime]: + def get_notification_times(self, evaluation_states_new)->list[str]: """# build the list of evaluation times ('now', as isoformat)""" evaluation_times = [datetime.datetime.now().isoformat() for _i in range(len(evaluation_states_new))] return evaluation_times @@ -113,3 +64,7 @@ class Notifier(): evaluation_notifications_sent = [self.determine_notification_state(state_old=int(state_old), state_new=int(state_new)) for state_old, state_new in zip(evaluation_states_old, evaluation_states_new)] return evaluation_notifications_sent + +class Notifier(NotifierFunctions): + def __init__(self): + super().__init__()