feat(notifier): ✨ add mail notification

.env-sample

@@ -2,6 +2,11 @@
 # for the IMAP connection to an OWL Exchange Server
 # The connection expect an Exchange account
 
-EMAIL=mail@intersport.fr
+EMAIL=mail@client.fr
 PASSWORD=password123
-EWS_URL=https://mail.intersport.fr/EWS/Exchange.asmx  # Default mail server
+EWS_URL=https://mail.client.fr/EWS/Exchange.asmx  # Default mail server
+
+# send mail report to
+N2_REPORT_RECIPIENTS=n2-support@client.fr
+# default report mail subject
+N2_REPORT_SUBJECT=[Logwatcher] Rapport erreurs N2

src/logwatcher/cli.py

@@ -16,6 +16,7 @@ from logwatcher.mail_reader import (
     move_analyzed_mails,
 )
 from logwatcher.models import LogEntry
+from logwatcher.notifier import send_n2_report
 from logwatcher.parser import parse_file, parse_lines
 from logwatcher.reporter import build_mail_summary, get_period, write_log_report
 
@@ -102,6 +103,8 @@ def from_mail(
         end_date,
         output_dir=output_dir,
     )
+    
+    # send mail with n2.log as attachment
     mail_body = build_mail_summary(
         nb_files=nb_files,
         nb_errors_total=len(relevant + irrelevant),
@@ -110,8 +113,7 @@ def from_mail(
         start_date=start_date,
         end_date=end_date,
     )
-
+    send_n2_report(mail_body, account, output_dir)
-    # send mail with n2.log as attachment
 
     move_analyzed_mails(account, messages)
     logger.info("============ logwatcher mail ended ============")

src/logwatcher/mail_reader.py

@@ -9,17 +9,16 @@ from exchangelib import (
     Configuration,
     Credentials,
     FileAttachment,
-    Folder,
     Message,
     Version,
 )
 from exchangelib.errors import UnauthorizedError
 
+from logwatcher.utils import get_or_create_folder
+
 logger = logging.getLogger(__name__)
 
-
+ANALYZED_FOLDER = "Analyzed"
-LOG_DIR = "Logs"
-ANALYZED_LOG_DIR = "Analyzed"
 LOG_IN_ATTACHMENT_PATTERN = "Le compte-rendu contient plus de 100 lignes."
 
 
@@ -140,7 +139,7 @@ def fetch_log_messages(account: Account) -> list[Message]:
     Returns:
         queryset of log-related mails
     """
-    logs_folder = account.msg_folder_root / LOG_DIR
+    logs_folder = account.msg_folder_root / "Logs"
     return list(logs_folder.all())
 
 
@@ -182,30 +181,6 @@ def extract_logs_from_mails(messages: list[Message]) -> list[str]:
     return logs
 
 
-def _get_or_create_analyzed_folder(account: Account) -> Folder:
-    """
-    Return the 'Analyzed' folder, creating it if it doesn't exist.
-
-    Args:
-        account: The mailbox account owning the Logs folder.
-
-    Returns:
-        The existing or newly created 'Analyzed' folder.
-
-    """
-    log_folder = account.msg_folder_root / LOG_DIR
-    matches = list(log_folder.glob(ANALYZED_LOG_DIR))
-
-    if matches:
-        logger.info(f"\tFolder '{ANALYZED_LOG_DIR}' found.")
-        return matches[0]
-
-    logger.info(f"\tCreating '{ANALYZED_LOG_DIR}' folder.")
-    analyzed_folder = Folder(parent=log_folder, name=ANALYZED_LOG_DIR)
-    analyzed_folder.save()
-    return analyzed_folder
-
-
 def move_analyzed_mails(account: Account, messages: list[Message]) -> None:
     """
     Move processed mails from the Logs folder to the Analyzed subfolder.
@@ -223,14 +198,14 @@ def move_analyzed_mails(account: Account, messages: list[Message]) -> None:
         return
 
     logger.info(
-        f"Moving {len(messages)} mails from '{LOG_DIR}' directory to '{ANALYZED_LOG_DIR}'"
+        f"Moving {len(messages)} mails to '{ANALYZED_FOLDER}'"
     )
 
-    analyzed_folder = _get_or_create_analyzed_folder(account)
+    analyzed_folder = get_or_create_folder(account, ANALYZED_FOLDER)
 
     ids = [(m.id, m.changekey) for m in messages]
     account.bulk_move(ids=ids, to_folder=analyzed_folder)
 
     logger.info(
-        f"Mails were successfully moved from '{LOG_DIR}' to '{ANALYZED_LOG_DIR}'."
+        f"Mails were successfully moved to '{ANALYZED_FOLDER}'."
     )

src/logwatcher/notifier.py

@@ -1,5 +1,38 @@
+import logging
+import os
+from pathlib import Path
+
 from exchangelib import Account, FileAttachment, Message
 
+from logwatcher.utils import get_or_create_folder
 
-def send_n2_report(summary: str):
+LOG_FOLDER = "Logs"
-    pass
+SENT_FOLDER = "Sent"
+
+logger = logging.getLogger(__name__)
+
+
+def send_n2_report(summary: str, account: Account, output_path: Path):
+    to = os.environ["N2_REPORT_RECIPIENTS"]
+    subject = os.environ["N2_REPORT_SUBJECT"]
+    
+    # generate file attachments
+    n2_log_file = output_path / "n2.log"
+    logger.info(f"Create attachment based on '{n2_log_file}' file.")
+    n2_log_attachment = FileAttachment(name=n2_log_file.name, content=n2_log_file.read_bytes(), is_inline=False)
+    logger.info(f"Attachment '{n2_log_attachment.name}' was successfully created.")
+
+    # get sent folder
+    sent_folder = get_or_create_folder(account, SENT_FOLDER)
+    
+    # create mail
+    logger.info(f"Create mail with account='{account.fullname}', recipients='{to}', subject='{subject}'.")
+    mail = Message(account=account, to_recipients=to, subject=subject, body=summary)
+    mail.attach(n2_log_attachment)
+    logger.info(f"Mail '{mail.subject}' was successfully created with attachment '{mail.attachments[0].name}'.")
+
+    # send and keep a copy to the Logs/Sent folder
+    logger.info(f"Attempt to send mail '{mail.subject}'...")
+    mail.send(copy_to_folder=sent_folder)  # TODO check if message was successfully sent
+    logger.info(f"Mail '{subject}' was successfully sent to '{to}'.")
+    logger.info(f"Mail is saved in '{sent_folder}' folder.")

src/logwatcher/utils.py

@@ -0,0 +1,28 @@
+import logging
+
+from exchangelib import Account, Folder
+
+logger = logging.getLogger(__name__)
+
+def get_or_create_folder(account: Account, folder_name: str) -> Folder:
+    """
+    Return the 'Analyzed' folder, creating it if it doesn't exist.
+
+    Args:
+        account: The mailbox account owning the Logs folder.
+
+    Returns:
+        The existing or newly created 'Analyzed' folder.
+
+    """
+    log_folder = account.msg_folder_root / "Logs"
+    matches = list(log_folder.glob(folder_name))
+
+    if matches:
+        logger.info(f"\tFolder '{folder_name}' found.")
+        return matches[0]
+
+    logger.info(f"\tCreating '{folder_name}' folder.")
+    folder = Folder(parent=log_folder, name=folder_name)
+    folder.save()
+    return folder