creating an HTML Email template for notifications, which includes a logo file (#TODO: store logo within BreCal git). Built out most of the Notifier. Provided suitable SQLQueries and updated the EmailHandler.

This commit is contained in:
Max Metz 2024-07-30 17:20:51 +02:00
parent 7c546f5ab7
commit f344588f89
5 changed files with 428 additions and 10 deletions

View File

@ -222,7 +222,7 @@ class SQLQuery():
"api_key, notify_email, notify_whatsapp, notify_signal, notify_popup, created, modified FROM user " +\
"WHERE user_name = ?username? OR user_email = ?username?"
return query
@staticmethod
def get_notifications()->str:
query = "SELECT id, shipcall_id, level, type, message, created, modified FROM notification " + \
@ -266,6 +266,11 @@ class SQLQuery():
query = "SELECT id, name, imo, callsign, participant_id, length, width, is_tug, bollard_pull, eni, created, modified, deleted FROM ship ORDER BY name"
return query
@staticmethod
def get_ship_by_id()->str:
query = "SELECT id, name, imo, callsign, participant_id, length, width, is_tug, bollard_pull, eni, created, modified, deleted FROM ship WHERE id = ?id?"
return query
@staticmethod
def get_times()->str:
query = "SELECT id, eta_berth, eta_berth_fixed, etd_berth, etd_berth_fixed, lock_time, lock_time_fixed, " + \
@ -316,6 +321,22 @@ class SQLQuery():
query = prefix+stage1+bridge+stage2+suffix
return query
@staticmethod
def get_notifications_post(schemaModel:dict)->str:
param_keys = {key:key for key in schemaModel.keys()}
prefix = "INSERT INTO notification ("
bridge = ") VALUES ("
suffix = ")"
non_dynamic_keys = ["id", "created", "modified"]
stage1 = ",".join([key for key in schemaModel.keys() if not key in non_dynamic_keys])
stage2 = ",".join([f"?{param_keys.get(key)}?" for key in schemaModel.keys() if not key in non_dynamic_keys])
query = prefix+stage1+bridge+stage2+suffix
return query
@staticmethod
def get_last_insert_id()->str:

View File

@ -1,7 +1,18 @@
import typing
import datetime
from BreCal.database.sql_handler import execute_sql_query_standalone
from BreCal.database.sql_queries import SQLQuery
from BreCal.schemas import model
from BreCal.brecal_utils.time_handling import difference_to_then
from BreCal.schemas.model import ShipcallType
eta_etd_type_dict = {
ShipcallType.arrival : "Ankunft",
ShipcallType.departure : "Abfahrt",
ShipcallType.shifting : "Wechselnd"
}
class Notifier():
"""
@ -141,6 +152,8 @@ class Notifier():
@staticmethod
def create(shipcall_id, old_state, new_state, user, update_database:bool=False)->typing.Optional[model.Notification]:
"""
# #TODO_refactor: drastically change this method. It should only generate notifications, but not send them.
Standalone function, which creates a Notification for a specific user.
Steps:
@ -173,6 +186,7 @@ class Notifier():
# get a list of all subscribed notification types and track the state (success or failure)
raise NotImplementedError("skeleton")
successes = {}
notification_type_list = Notifier.build_notification_type_list(user)
for notification_type in notification_type_list:
@ -183,7 +197,6 @@ class Notifier():
success_state = Notifier.send_notification_by_type(notification_type, message)
successes[notification_type] = success_state
raise NotImplementedError("skeleton")
notification = ...
return notification
@ -197,11 +210,10 @@ class Notifier():
def get_users_via_history(shipcall_id:int)->list[model.User]:
"""using the History objects, one can infer the user_id, which allows querying the Users"""
histories = execute_sql_query_standalone(query=SQLQuery.get_history(), param={"shipcallid" : shipcall_id}, model=model.History, command_type="query")
user_ids = [
history.user_id
for history in histories
]
users = [Notifier.get_user(user_id) for user_id in user_ids]
assert isinstance(histories,list)
assert all([isinstance(history,model.History) for history in histories])
users = [Notifier.get_user(history.user_id) for history in histories]
return users
@staticmethod
@ -306,8 +318,147 @@ class Notifier():
else:
raise ValueError(notification_type)
return
@staticmethod
def get_eligible_shipcalls():
"""
get all eligible shipcalls, which do not have a sent notification yet
criterion a) notification shall not be sent yet (evaluation_notifications_sent = 0)
criterion b) evaluation state is yellow or red (type 2 or 3)
"""
query = 'SELECT * FROM shipcall WHERE (evaluation_notifications_sent = ?evaluation_notifications_sent?) AND (evaluation = 2 OR evaluation = 3)'
evaluation_notifications_sent = 0
eligible_shipcalls = execute_sql_query_standalone(query=query, model=model.Shipcall, param={"evaluation_notifications_sent" : evaluation_notifications_sent})
return eligible_shipcalls
@staticmethod
def get_eligible_notifications_of_shipcall(shipcall:model.Shipcall, time_diff_threshold:float)->list[model.Notification]:
"""obtain all notifications, which belong to the shipcall id"""
query = SQLQuery.get_notifications()
eligible_notifications = execute_sql_query_standalone(query=query, model=model.Notification, param={"scid" : shipcall.id})
eligible_notifications = [notification for notification in eligible_notifications if Notifier.check_notification_exceeds_minimum_time_difference(notification, time_diff_threshold)]
eligible_notifications = [notification for notification in eligible_notifications if Notifier.check_notification_level_matches_shipcall_entry(notification, shipcall)]
return eligible_notifications
@staticmethod
def check_notification_exceeds_minimum_time_difference(notification:model.Notification, time_diff_threshold:float):
"""
a notification may only be sent, when the created notification has been created or modified {time_diff_threshold} seconds ago.
"""
assert (notification.created is not None) or (notification.modified is not None), f"must provide either 'created' or 'modified'"
if notification.modified is not None:
return difference_to_then(notification.modified)>time_diff_threshold
else:
return difference_to_then(notification.created)>time_diff_threshold
@staticmethod
def check_notification_level_matches_shipcall_entry(notification, shipcall):
"""
a notification may only be sent, when the shipcall entry matches the notification level.
otherwise, a user may have adapted the shipcall in the mean-time, so a notification would no longer be useful.
"""
return int(shipcall.evaluation) == int(notification.level)
@staticmethod
def get_eligible_notifications(shipcalls:list[model.Shipcall], time_diff_threshold:float):
"""obtain a list of all notifications of each element of the shipcall list."""
eligible_notifications = []
for shipcall in shipcalls:
eligible_notification = Notifier.get_eligible_notifications_of_shipcall(shipcall, time_diff_threshold)
eligible_notifications.extend(eligible_notification)
return eligible_notifications
@staticmethod
def create_notifications_for_user_list(shipcall, users:list[model.User]):
for user in users:
notification_type_list = Notifier.build_notification_type_list(user)
for notification_type in notification_type_list:
schemaModel = dict(shipcall_id = shipcall.id, level = int(shipcall.evaluation), type = notification_type, message = "", created = datetime.datetime.now(), modified=None)
query = SQLQuery.get_notifications_post(schemaModel)
schemas = execute_sql_query_standalone(query=query, param=schemaModel, command_type="execute")
return
@staticmethod
def generate_notifications(shipcall_id):
"""
This one-line method creates all notifications for the provided shipcall id. It does so by obtaining the shipcall,
looking up its history, and finding all attached users.
For each user, a notification will be created for each subscribed notification type (e.g., Email)
"""
# get the respective shipcall
shipcall = Notifier.get_shipcall(shipcall_id)
# find all attached users of the shipcall (checks the history, then reads out the user ids and builds the users)
users = Notifier.get_users_via_history(shipcall_id=shipcall.id)
# for each user, create one notification for each subscribed notification type (e.g., Email)
Notifier.create_notifications_for_user_list(shipcall, users)
return
@staticmethod
def create_etaetd_string(eta, etd): # #TODO_rename: function name is improvable
eta = eta.strftime("%d.%m.%Y %H:%M") if eta is not None else None
etd = etd.strftime("%d.%m.%Y %H:%M") if etd is not None else None
eta_etd = ""
if eta is not None and etd is not None:
eta_etd = f"{eta} - {etd}"
if eta is not None and etd is None:
eta_etd = f"{eta}"
if etd is None and etd is not None:
eta_etd = f"{etd}"
return eta_etd
@staticmethod
def prepare_notification_body(notification):
# obtain the respective shipcall and ship
shipcall = execute_sql_query_standalone(query=SQLQuery.get_shipcall_by_id(), model=model.Shipcall, param={"id" : notification.shipcall_id}, command_type="single")
ship = execute_sql_query_standalone(query=SQLQuery.get_ship_by_id(), model=model.Ship, param={"id" : shipcall.ship_id}, command_type="single")
# 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)]
evaluation_message = shipcall.evaluation_message
return (ship_name, evaluation_message, eta_etd, eta_etd_type)
@staticmethod
def shipcall_put_update_evaluation_notifications_sent_flag(notification):
# change the 'evaluation_notifications_sent' flag
evaluation_notifications_sent = 1
schemaModel = {"id":notification.shipcall_id, "evaluation_notifications_sent":evaluation_notifications_sent}
query = SQLQuery.get_shipcall_put(schemaModel)
schemas = execute_sql_query_standalone(query=query, param=schemaModel, command_type="execute")
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"""
if int(notification_type) == int(model.NotificationType.email):
return user.notify_email
elif int(notification_type) == int(model.NotificationType.push):
return user.notify_popup
elif int(notification_type) == int(model.NotificationType.undefined):
pass
### placeholders:
#elif int(notification_type) == int(model.NotificationType.whatsapp):
#return user.notify_whatsapp
#elif int(notification_type) == int(model.NotificationType.signal):
#return user.notify_signal
else: # placeholder: whatsapp/signal
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))]
#evaluation_times = [datetime.datetime.now().isoformat() for _i in range(len(evaluation_states_new))]

View File

@ -1,6 +1,8 @@
import os
import typing
import datetime
import smtplib
from socket import gaierror
from getpass import getpass
from email.message import EmailMessage
import mimetypes
@ -42,7 +44,10 @@ class EmailHandler():
self.mail_port = mail_port
self.mail_address = mail_address
self.server = smtplib.SMTP_SSL(self.mail_server, self.mail_port) # alternatively, use smtplib.SMTP
try:
self.server = smtplib.SMTP_SSL(self.mail_server, self.mail_port) # alternatively, use smtplib.SMTP
except gaierror:
raise Exception(f"'socket.gaierror' raised. This commonly happens, when there is no access to the server (e.g., by not having an internet connection)")
def check_state(self):
"""check, whether the server login took place and is open."""
@ -226,4 +231,71 @@ class EmailHandler():
time.sleep(delete_after_s_seconds)
if os.path.exists(temp_filename):
os.remove(temp_filename)
return
return
import typing
from email.mime.application import MIMEApplication
import mimetypes
def add_bremen_calling_logo(msg_multipart, path):
"""
The image is not attached automatically when it is embedded to the content. To circumvent this,
one commonly creates attachments, which are referred to in the email content.
The content body refers to 'LogoBremenCalling', which the 'Content-ID' of the logo is assigned as.
"""
with open(path, 'rb') as file:
attachment = MIMEApplication(file.read(), _subtype=mimetypes.MimeTypes().guess_type(path), Name="bremen_calling.png")
attachment.add_header('Content-Disposition','attachment',filename=str(os.path.basename(path)))
attachment.add_header('Content-ID', '<LogoBremenCalling>')
msg_multipart.attach(attachment)
return msg_multipart
def create_shipcall_evaluation_notification(email_handler, ship_name:str, evaluation_message:str, eta_etd_str:str, eta_etd_type:str, content:str, files:typing.Optional[list[str]]):
"""
email_handler : EmailHandler. Contains meta-level information about the mail server and sender's Email.
ship_name : str. Name of the referenced ship, so the user knows the context.
evaluation_message : str. Brief description of the current evaluation state
eta_etd_str : str. Readable format of a datetime.datetime object, which is either ETA, ETD or both. Informs the user about when the shipcall is due.
eta_etd_type : str. Reference to the time, whether it arrives/leaves/shifts.
content : str (or filepath). Should refer to the template, which defines the content. This file contains HTML-structured text.
files: (optional). List of file paths, which are included as attachments.
"""
subject = f"{ship_name} (vorauss. {eta_etd_type}: {eta_etd_str})"
# create message_body
message_body = content # "Hello World."
evaluation_message_reformatted = evaluation_message.replace("\n", "<br>")
adaptive_content = f'<br>Betrifft: {ship_name} ({eta_etd_str})<font size="1"><br>{evaluation_message_reformatted}</font>'
message_body = message_body.replace("#ADAPTIVECONTENT", adaptive_content)
msg = email_handler.create_email(subject=subject, message_body=message_body, subtype="html")
msg_multipart = email_handler.translate_mail_to_multipart(msg=msg)
if files is not None:
for path in files:
assert os.path.exists(path), f"cannot find attachment at path: {path}"
email_handler.attach_file(path, msg=msg_multipart)
# add the bremen calling logo, which is referred to in the email body
msg_multipart = add_bremen_calling_logo(msg_multipart, path=os.path.join("C:/Users/User/brecal/misc/logo_bremen_calling.png"))
return (msg_multipart,content)
def send_notification(email_handler, email_tgts, msg, pwd, debug=False):
email_handler.login(interactive=False, pwd=pwd)
try:
assert email_handler.check_login()
if not debug:
email_handler.send_email(msg, email_tgts)
else:
print(f"(send_notification INFO): debugging state. Would have sent an Email to: {email_tgts}")
finally:
email_handler.close()
return

View File

@ -0,0 +1,159 @@
<!doctype html>
<html lang="en">
<head>
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
<title>Simple Transactional Email</title>
<style media="all" type="text/css">
@media all {
.btn-primary table td:hover {
background-color: #ec0867 !important;
}
.btn-primary a:hover {
background-color: #ec0867 !important;
border-color: #ec0867 !important;
}
}
@media only screen and (max-width: 640px) {
.main p,
.main td,
.main span {
font-size: 16px !important;
}
.wrapper {
padding: 8px !important;
}
.content {
padding: 0 !important;
}
.container {
padding: 0 !important;
padding-top: 8px !important;
width: 100% !important;
}
.main {
border-left-width: 0 !important;
border-radius: 0 !important;
border-right-width: 0 !important;
}
.btn table {
max-width: 100% !important;
width: 100% !important;
}
.btn a {
font-size: 16px !important;
max-width: 100% !important;
width: 100% !important;
}
}
@media all {
.ExternalClass {
width: 100%;
}
.ExternalClass,
.ExternalClass p,
.ExternalClass span,
.ExternalClass font,
.ExternalClass td,
.ExternalClass div {
line-height: 100%;
}
.apple-link a {
color: inherit !important;
font-family: inherit !important;
font-size: inherit !important;
font-weight: inherit !important;
line-height: inherit !important;
text-decoration: none !important;
}
#MessageViewBody a {
color: inherit;
text-decoration: none;
font-size: inherit;
font-family: inherit;
font-weight: inherit;
line-height: inherit;
}
}
</style>
</head>
<body style="font-family: Helvetica, sans-serif; -webkit-font-smoothing: antialiased; font-size: 16px; line-height: 1.3; -ms-text-size-adjust: 100%; -webkit-text-size-adjust: 100%; background-color: #f4f5f6; margin: 0; padding: 0;">
<table role="presentation" border="0" cellpadding="0" cellspacing="0" class="body" style="border-collapse: separate; mso-table-lspace: 0pt; mso-table-rspace: 0pt; background-color: #f4f5f6; width: 100%;" width="100%" bgcolor="#f4f5f6">
<tr>
<td style="font-family: Helvetica, sans-serif; font-size: 16px; vertical-align: top;" valign="top">&nbsp;</td>
<td class="container" style="font-family: Helvetica, sans-serif; font-size: 16px; vertical-align: top; max-width: 600px; padding: 0; padding-top: 24px; width: 600px; margin: 0 auto;" width="600" valign="top">
<div class="content" style="box-sizing: border-box; display: block; margin: 0 auto; max-width: 600px; padding: 0;">
<!-- START CENTERED WHITE CONTAINER -->
<span class="preheader" style="color: transparent; display: none; height: 0; max-height: 0; max-width: 0; opacity: 0; overflow: hidden; mso-hide: all; visibility: hidden; width: 0;">Ein Schiffsanlauf benötigt Ihre Aufmerksamkeit.</span>
<table role="presentation" border="0" cellpadding="0" cellspacing="0" class="main" style="border-collapse: separate; mso-table-lspace: 0pt; mso-table-rspace: 0pt; background: #ffffff; border: 1px solid #eaebed; border-radius: 16px; width: 100%;" width="100%">
<!-- START MAIN CONTENT AREA -->
<div style="text-align: center;">
<img src="cid:LogoBremenCalling" height="100" width="100" alt="Bild kann nicht geladen werden." border="0" align="center">
</div>
<tr>
<td class="wrapper" style="font-family: Helvetica, sans-serif; font-size: 16px; vertical-align: top; box-sizing: border-box; padding: 24px;" valign="top">
<p style="font-family: Helvetica, sans-serif; font-size: 16px; font-weight: normal; margin: 0; margin-bottom: 16px;">Ahoi,</p>
<p style="font-family: Helvetica, sans-serif; font-size: 16px; font-weight: normal; margin: 0; margin-bottom: 16px;">ein Schiffsanlauf benötigt Ihre Aufmerksamkeit. Bei der Prüfung der Daten haben wir wahrgenommen, dass ein Problem aufgetreten sein könnte.</p>
<table role="presentation" border="0" cellpadding="0" cellspacing="0" class="btn btn-primary" style="border-collapse: separate; mso-table-lspace: 0pt; mso-table-rspace: 0pt; box-sizing: border-box; width: 100%; min-width: 100%;" width="100%">
<tbody>
<tr>
<td align="left" style="font-family: Helvetica, sans-serif; font-size: 16px; vertical-align: top; padding-bottom: 16px;" valign="top">
<table role="presentation" border="0" cellpadding="0" cellspacing="0" style="border-collapse: separate; mso-table-lspace: 0pt; mso-table-rspace: 0pt; width: auto;">
<tbody>
<tr>
<td style="font-family: Helvetica, sans-serif; font-size: 16px; vertical-align: top; border-radius: 4px; text-align: center; background-color: #0867ec;" valign="top" align="center" bgcolor="#0867ec"> <a href="https://bsmd.de/" target="_blank" style="border: solid 2px #0867ec; border-radius: 4px; box-sizing: border-box; cursor: pointer; display: inline-block; font-size: 16px; font-weight: bold; margin: 0; padding: 12px 24px; text-decoration: none; text-transform: capitalize; background-color: #0867ec; border-color: #0867ec; color: #ffffff;">Zu Bremen Calling</a> </td>
</tr>
</tbody>
</table>
</td>
</tr>
</tbody>
</table>
<p style="font-family: Helvetica, sans-serif; font-size: 16px; font-weight: normal; margin: 0; margin-bottom: 16px;">#ADAPTIVECONTENT</p>
<p style="font-family: Helvetica, sans-serif; font-size: 16px; font-weight: normal; margin: 0; margin-bottom: 16px;"><br>Falls es sich hierbei um eine Fehlmeldung handelt, Verzeihung. Wir sind stets interessiert daran, die Software zu verbessern. Senden Sie uns gerne eine <a href="mailto:bsmd@bsmd.de">Email</a>.</p>
</td>
</tr>
<!-- END MAIN CONTENT AREA -->
</table>
<!-- START FOOTER -->
<div class="footer" style="clear: both; padding-top: 24px; text-align: center; width: 100%;">
<table role="presentation" border="0" cellpadding="0" cellspacing="0" style="border-collapse: separate; mso-table-lspace: 0pt; mso-table-rspace: 0pt; width: 100%;" width="100%">
<tr>
<td class="content-block" style="font-family: Helvetica, sans-serif; vertical-align: top; color: #9a9ea6; font-size: 12px; text-align: center;" valign="top" align="center">
<span class="apple-link" style="color: #9a9ea6; font-size: 12px; text-align: center;">Bremer Schiffsmeldedienst, Kapt. P. Langbein e.K., Hafenkopf II / Überseetor 20, 28217 Bremen / Germany</span>
<br> Sie möchten keine Benachrichtigungen mehr erhalten? <a href="mailto:bsmd@bsmd.de" style="text-decoration: underline; color: #9a9ea6; font-size: 12px; text-align: center;">Hier abmelden</a>.
</td>
</tr>
<tr>
<td class="content-block powered-by" style="font-family: Helvetica, sans-serif; vertical-align: top; color: #9a9ea6; font-size: 12px; text-align: center;" valign="top" align="center">
Mail-Design by <a href="http://htmlemail.io" style="color: #9a9ea6; font-size:12px; text-align: center; text-decoration: none;">HTMLemail.io</a>
</td>
</tr>
</table>
</div>
<!-- END FOOTER -->
<!-- END CENTERED WHITE CONTAINER --></div>
</td>
<td style="font-family: Helvetica, sans-serif; font-size: 16px; vertical-align: top;" valign="top">&nbsp;</td>
</tr>
</table>
</body>
</html>

View File

@ -0,0 +1,15 @@
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