relocating logo_bremen_calling.png and default_email_template to a novel folder in the library: resources. Adapted the dynamic path functions in the library. Created tests, which assert, when the files are missing. Relocating functions from stubs/email_template.py to the email_handling.py to avoid confusion.

This commit is contained in:
Max Metz 2024-07-31 11:49:44 +02:00
parent 9efabf6367
commit bc6d391094
8 changed files with 95 additions and 45 deletions

View File

@ -5,10 +5,9 @@ from BreCal.database.sql_queries import SQLQuery
from BreCal.schemas import model from BreCal.schemas import model
from BreCal.brecal_utils.time_handling import difference_to_then 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.services.email_handling import EmailHandler, create_shipcall_evaluation_notification, send_notification, get_default_html_email
from BreCal.database.enums import ParticipantwiseTimeDelta from BreCal.database.enums import ParticipantwiseTimeDelta
from BreCal.stubs.email_template import get_default_html_email
eta_etd_type_dict = { eta_etd_type_dict = {
model.ShipcallType.arrival : "Ankunft", model.ShipcallType.arrival : "Ankunft",
model.ShipcallType.departure : "Abfahrt", model.ShipcallType.departure : "Abfahrt",
@ -58,30 +57,40 @@ class Notifier():
email_handler = EmailHandler(mail_server='w01d5503.kasserver.com', mail_port=465, mail_address="max.metz@scope-sorting.com") 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=' pwd = b'gAAAAABmqJlkXbtJTL1tFiyQNHhF_Y7sgtVI0xEx07ybwbX70Ro1Vp73CLDq49eFSYG-1SswIDQ2JBSORYlWaR-Vh2kIwPHy_lX8SxkySrRvBRzkyZP5x0I='
try: # get candidates: find all eligible shipcalls, where the evaluation state is yellow or red & the notifications are not yet sent
# login in advance, so the email handler uses a shared connection. It disconnects only once at the end of the call. eligible_shipcalls = Notifier.get_eligible_shipcalls()
email_handler.login(interactive=False, pwd=pwd)
# get candidates: find all eligible shipcalls, where the evaluation state is yellow or red & the notifications are not yet sent # find all notifications, which belong to the shipcall ids of the eligible_shipcall list
eligible_shipcalls = Notifier.get_eligible_shipcalls() # 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)
# find all notifications, which belong to the shipcall ids of the eligible_shipcall list if len(eligible_notifications) > 0: # only perform a login when there are eligible notifications
# a time_diff_threshold is used to block those notifications, which are still fairly novel try:
eligible_notifications = Notifier.get_eligible_notifications(eligible_shipcalls, time_diff_threshold) # 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)
for notification in eligible_notifications: for notification in eligible_notifications:
# get all users, which are attached to the shipcall (uses the History dataset) eligible_users = Notifier.get_eligible_users(notification)
users = Notifier.get_users_via_history(notification.shipcall_id)
# 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)]
# 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()
# create an Email and send it to each eligible_user.
# #TODO: this method must be a distributor. It should send emails for those, who want emails, and provide placeholders for other types of notifications
Notifier.create_and_send_email_notification(email_handler, pwd, eligible_users, notification, update_database=update_database, debug=debug)
finally:
email_handler.close()
return return
@staticmethod
def get_eligible_users(notification):
# get all users, which are attached to the shipcall (uses the History dataset)
users = Notifier.get_users_via_history(notification.shipcall_id)
# filter: only consider the users, which have subscribed to the notification type
eligible_users = [user for user in users if Notifier.check_user_is_subscribed_to_notification_type(user,notification_type=notification.type)]
# filter: consider only those users, where an Email is set
# #TODO: this is Email-specific and should not be a filter for other notifications
eligible_users = [user for user in eligible_users if user.user_email is not None]
return eligible_users
@staticmethod @staticmethod
def create(shipcall_id, old_state, new_state, user, update_database:bool=False)->typing.Optional[model.Notification]: def create(shipcall_id, old_state, new_state, user, update_database:bool=False)->typing.Optional[model.Notification]:
@ -372,6 +381,12 @@ class Notifier():
@staticmethod @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): 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):
"""
# #TODO_rename: when there is more than one type of notification, this should be renamed. This method refers to a validation-state notification
this 'naive' method creates a message and simply sends it to all users in a list of users.
Afterwards, the database will be updated, so the shipcall no longer requires a notification.
"""
email_tgts = [user.user_email for user in users if user.user_email is not None] 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) ship_name, evaluation_message, eta_etd, eta_etd_type = Notifier.prepare_notification_body(notification)
@ -385,6 +400,8 @@ class Notifier():
# send the messages via smtlib's SSL functions # send the messages via smtlib's SSL functions
send_notification(email_handler, email_tgts, msg_multipart, pwd, debug=debug) send_notification(email_handler, email_tgts, msg_multipart, pwd, debug=debug)
# #TODO_refactor: when there are multiple notification types, it makes sense to decouple updating the database
# from this method. Hence, an update would be done after *all* notifications are sent
if update_database: if update_database:
Notifier.shipcall_put_update_evaluation_notifications_sent_flag(notification) Notifier.shipcall_put_update_evaluation_notifications_sent_flag(notification)
return return
@ -412,6 +429,3 @@ class Notifier():
raise NotImplementedError(notification_type) raise NotImplementedError(notification_type)
"""# build the list of evaluation times ('now', as isoformat)"""
#evaluation_times = [datetime.datetime.now().isoformat() for _i in range(len(evaluation_states_new))]

View File

Binary file not shown.

After

Width:  |  Height:  |  Size: 20 KiB

View File

@ -237,6 +237,48 @@ import typing
from email.mime.application import MIMEApplication from email.mime.application import MIMEApplication
import mimetypes import mimetypes
import os
def find_warning_notification_email_template()->str:
"""
dynamically finds the 'default_email_template.txt' file within the module.
"""
# __file__ is BreCal/stubs/email_template.py
# parent of email_template.py is stubs
# parent of stubs is BreCal
brecal_root_folder = os.path.dirname(os.path.dirname(__file__)) # .../BreCal
resource_root_folder = os.path.join(brecal_root_folder, "resources") # .../BreCal/resources
html_filepath = os.path.join(resource_root_folder,"warning_notification_email_template.txt") # .../BreCal/resources/warning_notification_email_template.txt
assert os.path.exists(html_filepath), f"could not find default email template file at path: {html_filepath}"
return html_filepath
def get_default_html_email()->str:
"""
dynamically finds the 'default_email_template.txt' file within the module. It opens the file and returns the content.
__file__ returns to the file, where this function is stored (e.g., within BreCal.stubs.email_template)
using the dirname refers to the directory, where __file__ is stored.
finally, the 'default_email_template.txt' is stored within that folder
"""
html_filepath = find_warning_notification_email_template()
with open(html_filepath,"r", encoding="utf-8") as file: # encoding = "utf-8" allows for German Umlaute
content = file.read()
return content
def find_bremen_calling_logo():
"""
find the path towards the logo file (located at 'brecal\src\BreCalClient\Resources\logo_bremen_calling.png')
"""
# __file__ is services/email_handling.py
# parent of __file__ is services
# parent of services is BreCal
src_root_folder = os.path.dirname(os.path.dirname(__file__)) # .../BreCal
resource_root_folder = os.path.join(src_root_folder, "resources")
path = os.path.join(resource_root_folder, "logo_bremen_calling.png")
assert os.path.exists(path), f"cannot find logo of bremen calling at path: {os.path.abspath(path)}"
return path
def add_bremen_calling_logo(msg_multipart): def add_bremen_calling_logo(msg_multipart):
""" """
The image is not attached automatically when it is embedded to the content. To circumvent this, The image is not attached automatically when it is embedded to the content. To circumvent this,
@ -244,12 +286,7 @@ def add_bremen_calling_logo(msg_multipart):
The content body refers to 'LogoBremenCalling', which the 'Content-ID' of the logo is assigned as. The content body refers to 'LogoBremenCalling', which the 'Content-ID' of the logo is assigned as.
""" """
# find the path towards the logo file (located at 'brecal\src\BreCalClient\Resources\logo_bremen_calling.png') path = find_bremen_calling_logo()
src_root_folder = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
resource_root_folder = os.path.join(src_root_folder, "BreCalClient", "Resources")
path = os.path.join(resource_root_folder, "logo_bremen_calling.png")
assert os.path.exists(path), f"cannot find logo of bremen calling at path: {os.path.abspath(path)}"
with open(path, 'rb') as file: with open(path, 'rb') as file:
attachment = MIMEApplication(file.read(), _subtype=mimetypes.MimeTypes().guess_type(path), Name="bremen_calling.png") attachment = MIMEApplication(file.read(), _subtype=mimetypes.MimeTypes().guess_type(path), Name="bremen_calling.png")

View File

@ -1,15 +0,0 @@
import os
def get_default_html_email():
"""
dynamically finds the 'default_email_template.txt' file within the module. It opens the file and returns the content.
__file__ returns to the file, where this function is stored (e.g., within BreCal.stubs.email_template)
using the dirname refers to the directory, where __file__ is stored.
finally, the 'default_email_template.txt' is stored within that folder
"""
html_filepath = os.path.join(os.path.dirname(__file__),"default_email_template.txt")
assert os.path.exists(html_filepath), f"could not find default email template file at path: {html_filepath}"
with open(html_filepath,"r", encoding="utf-8") as file: # encoding = "utf-8" allows for German Umlaute
content = file.read()
return content

View File

View File

@ -0,0 +1,14 @@
import pytest
import os
def test_find_bremen_calling_logo():
from BreCal.services.email_handling import find_bremen_calling_logo
path = find_bremen_calling_logo()
assert os.path.exists(path), f"cannot find the bremen calling logo file, which is needed for notifications (e.g., Email). Searched at path: \n\t{path}"
return
def test_find_warning_notification_email_template():
from BreCal.services.email_handling import find_warning_notification_email_template
path = find_warning_notification_email_template()
assert os.path.exists(path), f"cannot find the required email template, which is needed for warning notifications. Searched at path: \n\t{path}"
return