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

Reduced log-verbosity on the server

Browse Source
This commit is contained in:
Daniel Schick 2025-02-27 13:48:13 +01:00
parent 3d76acb2f0
commit 189626d61c
3 changed files with 36 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -15,13 +15,13 @@ from email.mime.application import MIMEApplication
class EmailHandler(): class EmailHandler():
""" """
Creates an EmailHandler, which is capable of connecting to a mail server at a respective port, Creates an EmailHandler, which is capable of connecting to a mail server at a respective port,
as well as logging into a specific user's mail address. as well as logging into a specific user's mail address.
Upon creating messages, these can be sent via this handler. Upon creating messages, these can be sent via this handler.
Options: Options:
mail_server: address of the server, such as 'smtp.gmail.com' or 'w01d5503.kasserver.com mail_server: address of the server, such as 'smtp.gmail.com' or 'w01d5503.kasserver.com
mail_port: mail_port:
25 - SMTP Port, to send emails 25 - SMTP Port, to send emails
110 - POP3 Port, to receive emails 110 - POP3 Port, to receive emails
143 - IMAP Port, to receive from IMAP 143 - IMAP Port, to receive from IMAP
@ -38,6 +38,9 @@ class EmailHandler():
self.server = smtplib.SMTP_SSL(self.mail_server, self.mail_port) # alternatively, SMTP self.server = smtplib.SMTP_SSL(self.mail_server, self.mail_port) # alternatively, SMTP
# set the following to 0 to avoid log spamming
self.server.set_debuglevel(1) # 0: no debug, 1: debug
def check_state(self): def check_state(self):
"""check, whether the server login took place and is open.""" """check, whether the server login took place and is open."""
try: try:
@ -45,7 +48,7 @@ class EmailHandler():
return status_code==250 # 250: b'2.0.0 Ok' return status_code==250 # 250: b'2.0.0 Ok'
except smtplib.SMTPServerDisconnected: except smtplib.SMTPServerDisconnected:
return False return False
def check_connection(self): def check_connection(self):
"""check, whether the server object is connected to the server. If not, connect it. """ """check, whether the server object is connected to the server. If not, connect it. """
try: try:
@ -53,7 +56,7 @@ class EmailHandler():
except smtplib.SMTPServerDisconnected: except smtplib.SMTPServerDisconnected:
self.server.connect(self.mail_server, self.mail_port) self.server.connect(self.mail_server, self.mail_port)
return return
def check_login(self)->bool: def check_login(self)->bool:
"""check, whether the server object is logged in as a user""" """check, whether the server object is logged in as a user"""
user = self.server.__dict__.get("user",None) user = self.server.__dict__.get("user",None)
@ -61,8 +64,8 @@ class EmailHandler():
def login(self, interactive:bool=True): def login(self, interactive:bool=True):
""" """
login on the determined mail server's mail address. By default, this function opens an interactive window to login on the determined mail server's mail address. By default, this function opens an interactive window to
type the password without echoing (printing '*******' instead of readable characters). type the password without echoing (printing '*******' instead of readable characters).
returns (status_code, status_msg) returns (status_code, status_msg)
""" """
@ -77,7 +80,7 @@ class EmailHandler():
def create_email(self, subject:str, message_body:str)->EmailMessage: def create_email(self, subject:str, message_body:str)->EmailMessage:
""" """
Create an EmailMessage object, which contains the Email's header ("Subject"), content ("Message Body") and the sender's address ("From"). Create an EmailMessage object, which contains the Email's header ("Subject"), content ("Message Body") and the sender's address ("From").
The EmailMessage object does not contain the recipients yet, as these will be defined upon sending the Email. The EmailMessage object does not contain the recipients yet, as these will be defined upon sending the Email.
""" """
msg = EmailMessage() msg = EmailMessage()
msg["Subject"] = subject msg["Subject"] = subject
@ -85,16 +88,16 @@ class EmailHandler():
#msg["To"] = email_tgts # will be defined in self.send_email #msg["To"] = email_tgts # will be defined in self.send_email
msg.set_content(message_body) msg.set_content(message_body)
return msg return msg
def build_recipients(self, email_tgts:list[str]): def build_recipients(self, email_tgts:list[str]):
""" """
email formatting does not support lists. Instead, items are joined into a comma-space-separated string. email formatting does not support lists. Instead, items are joined into a comma-space-separated string.
Example: Example:
[mail1@mail.com, mail2@mail.com] becomes [mail1@mail.com, mail2@mail.com] becomes
'mail1@mail.com, mail2@mail.com' 'mail1@mail.com, mail2@mail.com'
""" """
return ', '.join(email_tgts) return ', '.join(email_tgts)
def open_mime_application(self, path:str)->MIMEApplication: def open_mime_application(self, path:str)->MIMEApplication:
"""open a local file, read the bytes into a MIMEApplication object, which is built with the proper subtype (based on the file extension)""" """open a local file, read the bytes into a MIMEApplication object, which is built with the proper subtype (based on the file extension)"""
with open(path, 'rb') as file: with open(path, 'rb') as file:
@ -102,24 +105,24 @@ class EmailHandler():
attachment.add_header('Content-Disposition','attachment',filename=str(os.path.basename(path))) attachment.add_header('Content-Disposition','attachment',filename=str(os.path.basename(path)))
return attachment return attachment
def attach_file(self, path:str, msg:email.mime.multipart.MIMEMultipart)->None: def attach_file(self, path:str, msg:email.mime.multipart.MIMEMultipart)->None:
""" """
attach a file to the message. This function opens the file, reads its bytes, defines the mime type by the attach a file to the message. This function opens the file, reads its bytes, defines the mime type by the
path extension. The filename is appended as the header. path extension. The filename is appended as the header.
mimetypes: # https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types/Common_types mimetypes: # https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types/Common_types
""" """
attachment = self.open_mime_application(path) attachment = self.open_mime_application(path)
msg.attach(attachment) msg.attach(attachment)
return return
def send_email(self, msg:EmailMessage, email_tgts:list[str], cc_tgts:typing.Optional[list[str]]=None, bcc_tgts:typing.Optional[list[str]]=None, debug:bool=False)->typing.Union[dict,EmailMessage]: def send_email(self, msg:EmailMessage, email_tgts:list[str], cc_tgts:typing.Optional[list[str]]=None, bcc_tgts:typing.Optional[list[str]]=None, debug:bool=False)->typing.Union[dict,EmailMessage]:
""" """
send a prepared email message to recipients (email_tgts), copy (cc_tgts) and blind copy (bcc_tgts). send a prepared email message to recipients (email_tgts), copy (cc_tgts) and blind copy (bcc_tgts).
Returns a dictionary of feedback, which is commonly empty and the EmailMessage. Returns a dictionary of feedback, which is commonly empty and the EmailMessage.
When failing, this function returns an SMTP error instead of returning the default outputs. When failing, this function returns an SMTP error instead of returning the default outputs.
""" """
# Set the Recipients # Set the Recipients
msg["To"] = self.build_recipients(email_tgts) msg["To"] = self.build_recipients(email_tgts)
@ -130,15 +133,15 @@ class EmailHandler():
if bcc_tgts is not None: if bcc_tgts is not None:
msg["Bcc"] = self.build_recipients(bcc_tgts) msg["Bcc"] = self.build_recipients(bcc_tgts)
# when debugging, do not send the Email, but return the EmailMessage. # when debugging, do not send the Email, but return the EmailMessage.
if debug: if debug:
return {}, msg return {}, msg
assert self.check_login(), f"currently not logged in. Cannot send an Email. Make sure to properly use self.login first. " assert self.check_login(), f"currently not logged in. Cannot send an Email. Make sure to properly use self.login first. "
# send the prepared EmailMessage via the server. # send the prepared EmailMessage via the server.
feedback = self.server.send_message(msg) feedback = self.server.send_message(msg)
return feedback, msg return feedback, msg
def translate_mail_to_multipart(self, msg:EmailMessage): def translate_mail_to_multipart(self, msg:EmailMessage):
"""EmailMessage does not support HTML and attachments. Hence, one can convert an EmailMessage object.""" """EmailMessage does not support HTML and attachments. Hence, one can convert an EmailMessage object."""
if msg.is_multipart(): if msg.is_multipart():
@ -159,11 +162,11 @@ class EmailHandler():
# attach the remainder of the msg, such as the body, to the MIMEMultipart # attach the remainder of the msg, such as the body, to the MIMEMultipart
msg_new.attach(msg) msg_new.attach(msg)
return msg_new return msg_new
def print_email_attachments(self, msg:MIMEMultipart)->list[str]: def print_email_attachments(self, msg:MIMEMultipart)->list[str]:
"""return a list of lines of an Email, which contain 'filename=' as a list. """ """return a list of lines of an Email, which contain 'filename=' as a list. """
return [line_ for line_ in msg.as_string().split("\n") if "filename=" in line_] return [line_ for line_ in msg.as_string().split("\n") if "filename=" in line_]
def close(self): def close(self):
self.server.__dict__.pop("user",None) self.server.__dict__.pop("user",None)
self.server.__dict__.pop("password",None) self.server.__dict__.pop("password",None)

6
src/server/BreCal/validators/validation_error.py
View File

@ -59,7 +59,7 @@ def create_validation_error_response(ex:ValidationError, status_code:int=400, cr
if create_log: if create_log:
logging.warning(ex) if ex is not None else logging.warning(message) logging.warning(ex) if ex is not None else logging.warning(message)
print(ex) if ex is not None else print(message) # print(ex) if ex is not None else print(message)
return (serialized_response, status_code) return (serialized_response, status_code)
def create_werkzeug_error_response(ex:Forbidden, status_code:int=403, create_log:bool=True)->typing.Tuple[str,int]: def create_werkzeug_error_response(ex:Forbidden, status_code:int=403, create_log:bool=True)->typing.Tuple[str,int]:
@ -71,7 +71,7 @@ def create_werkzeug_error_response(ex:Forbidden, status_code:int=403, create_log
if create_log: if create_log:
logging.warning(ex) if ex is not None else logging.warning(message) logging.warning(ex) if ex is not None else logging.warning(message)
print(ex) if ex is not None else print(message) # print(ex) if ex is not None else print(message)
return serialized_response, status_code return serialized_response, status_code
def create_dynamic_exception_response(ex, status_code:int=400, message:typing.Optional[str]=None, create_log:bool=True): def create_dynamic_exception_response(ex, status_code:int=400, message:typing.Optional[str]=None, create_log:bool=True):
@ -83,5 +83,5 @@ def create_dynamic_exception_response(ex, status_code:int=400, message:typing.Op
if create_log: if create_log:
logging.warning(ex) if ex is not None else logging.warning(message) logging.warning(ex) if ex is not None else logging.warning(message)
print(ex) if ex is not None else print(message) # print(ex) if ex is not None else print(message)
return (serialized_response, status_code) return (serialized_response, status_code)

5
src/server/BreCal/validators/validation_rules.py
View File

@ -52,7 +52,8 @@ class ValidationRules(ValidationRuleFunctions):
# 'translate' all error codes into readable, human-understandable format. # 'translate' all error codes into readable, human-understandable format.
evaluation_results = [(state, self.describe_error_message(msg)) for (state, msg) in evaluation_results] evaluation_results = [(state, self.describe_error_message(msg)) for (state, msg) in evaluation_results]
logging.info(f"Validation results for shipcall {shipcall.id}: {evaluation_results}") if evaluation_results:
logging.info(f"Validation results for shipcall {shipcall.id}: {evaluation_results}")
# check, what the maximum state flag is and return it # check, what the maximum state flag is and return it
evaluation_state = np.max(np.array([result[0].value for result in evaluation_results])) if len(evaluation_results)>0 else StatusFlags.GREEN.value evaluation_state = np.max(np.array([result[0].value for result in evaluation_results])) if len(evaluation_results)>0 else StatusFlags.GREEN.value
@ -92,7 +93,7 @@ class ValidationRules(ValidationRuleFunctions):
if evaluation_states_old[0] != evaluation_states_new[0]: if evaluation_states_old[0] != evaluation_states_new[0]:
pooledConnection = getPoolConnection() pooledConnection = getPoolConnection()
commands = pydapper.using(pooledConnection) commands = pydapper.using(pooledConnection)
notification_type = 3 # RED (mapped to time_conflict) notification_type = 3 # RED (mapped to time_conflict)
if evaluation_states_new[0] == 2: if evaluation_states_new[0] == 2:
match evaluation_states_old[0]: match evaluation_states_old[0]:
case 0: case 0: