From 060190be6121e804452af71b0deec3846c771141 Mon Sep 17 00:00:00 2001 From: Blallo Date: Wed, 24 Aug 2022 11:13:12 +0200 Subject: [PATCH] Add email notifier --- latecomers/notifier.py | 92 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 latecomers/notifier.py diff --git a/latecomers/notifier.py b/latecomers/notifier.py new file mode 100644 index 0000000..f30b12e --- /dev/null +++ b/latecomers/notifier.py @@ -0,0 +1,92 @@ +# -*- encoding: utf-8 -*- +from datetime import datetime +from email import encoders +from email.mime.base import MIMEBase +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText +import logging +import smtplib +import ssl +import typing as T + + +RETRIES = 3 +logger = logging.getLogger(__name__) + + +def retry_and_log(func: T.Callable[..., T.Any]) -> T.Callable[..., T.Any]: + """ + This decorator wraps a function, logs at info in case of success and logs + error in case an exception is raised. + """ + def inner(*args, **kwargs): + for i in range(RETRIES): + try: + func(*args, **kwargs) + logger.info(f"Success ({i+1} tentative): {func.__name__}(args={args}, kwargs={kwargs})") # noqa: E501 + return + except Exception as e: + logger.error(f"Failure ({i+1} tentative): {e}") + + return inner + + +class Notifier(object): + def __init__( + self, + smtp_addr: T.Text, + smtp_port: int, + smtp_starttls: bool, + smtp_user: T.Text, + smtp_pass: T.Text, + smtp_from: T.Optional[T.Text] = None, + ) -> None: + context = ssl.create_default_context() + if smtp_starttls: + self._conn = smtplib.SMTP_SSL(smtp_addr, smtp_port) + self._greet = self._conn.ehlo + else: + self._conn = smtplib.SMTP(smtp_addr, smtp_port) + self._greet = lambda: self._conn.starttls(context=context) + + self._addr = smtp_addr + self._port = smtp_port + self._user = smtp_user + self._pass = smtp_pass + self._from = smtp_from if smtp_from else smtp_user + logger.info("%s initialized", self) + + def __str__(self) -> T.Text: + return f"Notifier<{self._addr}:{self._port},from={self._from}>" + + def send(self, to: T.List[T.Text], email: T.Text) -> None: + self._greet() + self._conn.login(self._user, self._pass) + self._conn.sendmail(self._from, to, email) + self._conn.close() + + @retry_and_log + def send_result(self, to: T.List[T.Text], result: bytes) -> None: + date = datetime.now().strftime("%Y-%m-%d") + body = f"Resoconto dei voli dal sito di AdR per l'aereoporto di Ciampino in data {date}" # noqa: E501 + + message = MIMEMultipart() + message["From"] = self._from + message["To"] = ",".join(to) + message["Subject"] = f"[{date}] Resoconto CIA da AdR" + + message.attach(MIMEText(body, "plain")) + + payload = MIMEBase("application", "octet-stream") + payload.set_payload(result) + encoders.encode_base64(payload) + payload.add_header( + "Content-Disposition", + f"attachment; filename={date}.xlsx", + ) + + message.attach(payload) + + email = message.as_string() + + self.send(to, email)