dschick/BreCal
dschick/BreCal
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

clearing older methods of the Notifier class. There is now a one-line function, which connects to the Email-server, populates candidates for notifications, creates those notifications one-by-one and sends them. Finally, the database is updated, so a notification is sent only once.

Browse Source
This commit is contained in:
Max Metz 2024-07-30 17:43:17 +02:00
parent f344588f89
commit 8cd6afbb69
2 changed files with 59 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

147
src/server/BreCal/notifications/notifier.py
View File

@ -5,12 +5,14 @@ from BreCal.database.sql_queries import SQLQuery
from BreCal.schemas import model
from BreCal.brecal_utils.time_handling import difference_to_then
from BreCal.services.email_handling import EmailHandler, create_shipcall_evaluation_notification, send_notification
from BreCal.database.enums import ParticipantwiseTimeDelta
from BreCal.schemas.model import ShipcallType
from BreCal.stubs.email_template import get_default_html_email
eta_etd_type_dict = {
ShipcallType.arrival : "Ankunft",
ShipcallType.departure : "Abfahrt",
ShipcallType.shifting : "Wechselnd"
model.ShipcallType.arrival : "Ankunft",
model.ShipcallType.departure : "Abfahrt",
model.ShipcallType.shifting : "Wechselnd"
}
@ -46,107 +48,39 @@ class Notifier():
- iterate over each remaining shipcall and apply .send_notification
- those which are unsent, shall be sent by the respective type
"""
raise NotImplementedError("skeleton")
# set a threshold, when alarm event notifications become eligible
time_diff_threshold = float(ParticipantwiseTimeDelta.NOTIFICATION)*60 # m minutes, converted to seconds
# get all shipcalls
all_shipcalls = NotImplementedError
debug = is_test # if is_test, the Emails will not be issued. Only a print message will be created.
update_database = True if not is_test else False # if_test, the database will not be updated.
time_diff_threshold = time_diff_threshold if not is_test else 0.0 # 0.0 delay when is_test is set.
shipcalls = [shipcall for shipcall in all_shipcalls if not shipcall.evaluation_notifications_sent]
for shipcall in shipcalls:
notification_list = Notifier.send_notification(shipcall, is_test=is_test)
email_handler = EmailHandler(mail_server='w01d5503.kasserver.com', mail_port=465, mail_address="max.metz@scope-sorting.com")
pwd = b'gAAAAABmqJlkXbtJTL1tFiyQNHhF_Y7sgtVI0xEx07ybwbX70Ro1Vp73CLDq49eFSYG-1SswIDQ2JBSORYlWaR-Vh2kIwPHy_lX8SxkySrRvBRzkyZP5x0I='
# #TODO: get all notifications
# #TODO: get matching shipcall (based on shipcall_id)
try:
# login in advance, so the email handler uses a shared connection. It disconnects only once at the end of the call.
email_handler.login(interactive=False, pwd=pwd)
# #TODO: filter: consider only those, which are not yet sent
# get candidates: find all eligible shipcalls, where the evaluation state is yellow or red & the notifications are not yet sent
eligible_shipcalls = Notifier.get_eligible_shipcalls()
# identify necessity
# #TODO: get the 'evaluation_notifications_sent' field from all shipcalls (based on shipcall_id)
# if not -> return
# USE shipcall.evaluation_notifications_sent
# find all notifications, which belong to the shipcall ids of the eligible_shipcall list
# a time_diff_threshold is used to block those notifications, which are still fairly novel
eligible_notifications = Notifier.get_eligible_notifications(eligible_shipcalls, time_diff_threshold)
# #TODO: those which are unsent, shall be created&sent by the respective type -- Note: consider the is_test argument
# iterate over the list of Notifier.build_notification_type_list
# one might use Notifier.create(..., update_database=True)
# use the History (GetHistory -- by shipcall_id) to identify all subscribed users
for notification in eligible_notifications:
# get all users, which are attached to the shipcall (uses the History dataset)
users = Notifier.get_users_via_history(notification.shipcall_id)
# #TODO: update the shipcall dataset ('evaluation_notifications_sent') -- Note: consider the is_test argument
# filter: only consider the users, which have subscribed to the notification type
users = [user for user in users if Notifier.check_user_is_subscribed_to_notification_type(user,notification_type=notification.type)]
# #TODO_clarify: how to handle the 'evaluation_notifications_sent', when there is no recipient?
return
# obtain the mail address of each respective user
Notifier.create_and_send_email_notification(email_handler, pwd, users, notification, update_database=update_database, debug=debug)
finally:
email_handler.close()
@staticmethod
def send_notification(shipcall:model.Shipcall, is_test:bool=False)->list[model.Notification]:
"""
Complex-function, which is responsible of creating notification messages, issuing them to users and optionally updating
the database. The requirement is, that the notification is required and passes through an internal set of filters.
Steps:
- get all notifications of shipcall_id
- identify the assigned list of users
- apply all filters. When a filter triggers, exit. If not, create and send a notification.
"""
update_database = False if is_test else True
# #TODO: the concept of old state and new state must be refactored.
# old state: read shipcall_id from notifications and look for the latest finding (if None -> EvaluationType.undefined)
# new state: read shipcall_id from shipcalls and look for the *current* 'evaluation' (-> EvaluationType(value))
# get existing notifications by shipcall_id (list)
existing_notifications = Notifier.get_existing_notifications(shipcall_id=shipcall.id)
old_state = NotImplementedError
new_state = shipcall.evaluation
# get User by querying all History objects of a shipcall_id
users = Notifier.get_users_via_history(shipcall_id=shipcall.id)
# identify necessity
# state-check: Did the 'evaluation' shift to a higher level of severity?
severity_bool = Notifier.check_higher_severity(old_state, new_state)
if not severity_bool:
return None
# #TODO: time-based filter. There shall be 'enough' time between the evaluation time and NOW
evaluation_time = shipcall.evaluation_time
# latency_bool = #TODO_DIFFERENCE_FROM_NOW_TO_EVALUATION_TIME____THIS_METHOD_ALREADY_EXISTS(evaluation_time)
# careful: what is True, what is False?
# if latency_booL:
# return None
notification_list = []
for user in users:
notification = Notifier.create(
shipcall.id,
old_state,
new_state,
user,
update_database=update_database,
is_test=is_test
)
notification_list.append(notification)
return notification_list
@staticmethod
def publish(shipcall_id, old_state, new_state, user, update_database:bool=False)->typing.Optional[model.Notification]:
"""
Complex-function, which creates, sends and documents a notification. It serves as a convenience function.
The method does not apply internal filters to identify, whether a notification should be created in the first place.
options:
update_database: bool.
# #TODO: instead of update_database, one may also use is_test
"""
# 1.) create
# ... = Notifier.create(shipcall_id, old_state, new_state, user) # e.g., might return a dictionary of dict[model.NotificationType, str], where str is the message
# 2.) send
# ... = Notifier.send(...) # should contain internal 'logistics', which user the respective handlers to send notifications
# 3.) document (mysql database)
# if update_database
# ... = Notifier.document(...)
raise NotImplementedError("skeleton")
return
@staticmethod
@ -422,7 +356,7 @@ class Notifier():
# use ship & shipcall data models to prepare the body
ship_name = ship.name
eta_etd = Notifier.create_etaetd_string(shipcall.eta, shipcall.etd)
eta_etd_type = eta_etd_type_dict[ShipcallType(shipcall.type)]
eta_etd_type = eta_etd_type_dict[model.ShipcallType(shipcall.type)]
evaluation_message = shipcall.evaluation_message
return (ship_name, evaluation_message, eta_etd, eta_etd_type)
@ -436,6 +370,25 @@ class Notifier():
schemas = execute_sql_query_standalone(query=query, param=schemaModel, command_type="execute")
return
@staticmethod
def create_and_send_email_notification(email_handler:EmailHandler, pwd:bytes, users:list[model.User], notification:model.Notification, update_database:bool=True, debug:bool=False):
email_tgts = [user.user_email for user in users if user.user_email is not None]
ship_name, evaluation_message, eta_etd, eta_etd_type = Notifier.prepare_notification_body(notification)
content = get_default_html_email()
files = [] # optional attachments
msg_multipart,msg_content = create_shipcall_evaluation_notification(
email_handler, ship_name, evaluation_message, eta_etd, eta_etd_type, content, files=files
)
# send the messages via smtlib's SSL functions
send_notification(email_handler, email_tgts, msg_multipart, pwd, debug=debug)
if update_database:
Notifier.shipcall_put_update_evaluation_notifications_sent_flag(notification)
return
@staticmethod
def check_user_is_subscribed_to_notification_type(user,notification_type):
"""given a notification, one can check, whether the current user has subscribed to the respective notification_type. Returns a boolean"""

7
src/server/BreCal/services/email_handling.py
View File

@ -287,7 +287,9 @@ def create_shipcall_evaluation_notification(email_handler, ship_name:str, evalua
return (msg_multipart,content)
def send_notification(email_handler, email_tgts, msg, pwd, debug=False):
email_handler.login(interactive=False, pwd=pwd)
already_logged_in = email_handler.check_login()
if not already_logged_in:
email_handler.login(interactive=False, pwd=pwd)
try:
assert email_handler.check_login()
@ -297,5 +299,6 @@ def send_notification(email_handler, email_tgts, msg, pwd, debug=False):
print(f"(send_notification INFO): debugging state. Would have sent an Email to: {email_tgts}")
finally:
email_handler.close()
if not already_logged_in:
email_handler.close()
return