Add email notifier

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)