-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
bd7ddd1
commit e7a20f6
Showing
5 changed files
with
173 additions
and
141 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| import logging | ||
| import os | ||
| import smtplib | ||
| from email.message import EmailMessage | ||
|
|
||
| __all__ = ["EmailAlert"] | ||
|
|
||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class EmailAlert: | ||
| def __init__(self): | ||
| self.from_email: str = os.getenv("FROM_EMAIL") | ||
| self.from_password: str = os.getenv("FROM_PASSWORD") | ||
| self.smtp_server: str = os.getenv("SMTP_SERVER") | ||
| self.smtp_port: int = int(os.getenv("SMTP_PORT")) | ||
|
|
||
| @staticmethod | ||
| def _anonymize_email(email): | ||
| local_part, domain = email.split("@") | ||
| return f"{'*' * len(local_part)}@{domain}" | ||
|
|
||
| def send(self, subject: str, body: str, to_email: str): | ||
| if not all( | ||
| isinstance(i, str) | ||
| for i in [subject, body, to_email, self.from_email, self.from_password] | ||
| ): | ||
| raise ValueError("All parameters must be of type str") | ||
|
|
||
| logger.info(f"Preparing to send email to {self._anonymize_email(to_email)}") | ||
|
|
||
| # Create the email message | ||
| msg = EmailMessage() | ||
| msg.set_content(body.rstrip()) | ||
| msg["Subject"] = subject | ||
| msg["From"] = self.from_email | ||
| msg["To"] = to_email | ||
|
|
||
| try: | ||
| # Connect to the mail server using TLS | ||
| with smtplib.SMTP(self.smtp_server, self.smtp_port) as server: | ||
| # Enable TLS encryption | ||
| server.starttls() | ||
| # Log in to your email account | ||
| server.login(self.from_email, self.from_password) | ||
|
|
||
| # Send the email | ||
| server.send_message(msg) | ||
| logger.info(f"Email sent to {self._anonymize_email(to_email)}") | ||
| except smtplib.SMTPException as e: | ||
| logger.error( | ||
| f"SMTP error occurred when sending email to {self._anonymize_email(to_email)}: {e}" | ||
| ) | ||
| except Exception as e: | ||
| logger.error( | ||
| f"Unexpected error occurred when sending email to {self._anonymize_email(to_email)}: {e}" | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,129 +1,66 @@ | ||
| import logging | ||
| import importlib | ||
| import logging.config | ||
| import os | ||
| import smtplib | ||
| import sys | ||
| from email.message import EmailMessage | ||
| from logging.handlers import RotatingFileHandler | ||
|
|
||
| from dotenv import load_dotenv | ||
|
|
||
| __all__ = ["logger", "API_KEY", "GET_URL", "SERVERS_URL", "MAX_RETRIES", "RETRY_BACKOFF_FACTOR", "ROTATE", "POST_BACKUP_SCRIPT"] | ||
|
|
||
| # Environment .env | ||
| load_dotenv() | ||
|
|
||
| # Logger | ||
| LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s" | ||
| MAX_LOG_FILE_BYTES = 500000 | ||
| BACKUP_COUNT = 15 | ||
| LOG_LEVEL = logging.__getattribute__(os.getenv("LOG_LEVEL") or "ERROR") | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
| logger.setLevel(logging.DEBUG) | ||
|
|
||
| # console log (stderr) | ||
| handler = logging.StreamHandler(stream=sys.stderr) | ||
| handler.setLevel(LOG_LEVEL) | ||
| handler.setFormatter(logging.Formatter(LOG_FORMAT)) | ||
| logger.addHandler(handler) | ||
|
|
||
| # file logger | ||
| handler = RotatingFileHandler( | ||
| "backup.log", maxBytes=MAX_LOG_FILE_BYTES, backupCount=BACKUP_COUNT | ||
| ) | ||
| handler.setLevel(logging.INFO) # log file log level is always INFO | ||
| handler.setFormatter(logging.Formatter(LOG_FORMAT)) | ||
| logger.addHandler(handler) | ||
|
|
||
| # API key stuff | ||
| API_KEY = os.getenv("API_KEY") or "" | ||
| GET_URL = os.getenv("GET_URL") or "" | ||
| SERVERS_URL = os.getenv("SERVERS_URL") or "" | ||
| MAX_RETRIES = int(os.getenv("MAX_RETRIES") or "") or 5 | ||
| RETRY_BACKOFF_FACTOR = int(os.getenv("RETRY_BACKOFF_FACTOR") or "0") or 1 | ||
| SEND_EMAILS = (str(os.getenv("SEND_EMAILS") or "").lower() or "true") == "true" | ||
| ROTATE = (str(os.getenv("ROTATE") or "").lower() or "false") == "true" | ||
| POST_BACKUP_SCRIPT = os.getenv("POST_BACKUP_SCRIPT") # Optional | ||
|
|
||
|
|
||
|
|
||
| class EmailAlert: | ||
| def __init__(self, from_email, from_password, smtp_server, smtp_port): | ||
| self.from_email = from_email | ||
| self.from_password = from_password | ||
| self.smtp_server = smtp_server | ||
| self.smtp_port = int(smtp_port) | ||
|
|
||
| def anonymize_email(self, email): | ||
| local_part, domain = email.split("@") | ||
| return f"{'*' * len(local_part)}@{domain}" | ||
|
|
||
| def send(self, subject, body, to_email): | ||
| if not all( | ||
| isinstance(i, str) | ||
| for i in [subject, body, to_email, self.from_email, self.from_password] | ||
| ): | ||
| raise ValueError("All parameters must be of type str") | ||
|
|
||
| logger.info(f"Preparing to send email to {self.anonymize_email(to_email)}") | ||
|
|
||
| # Create the email message | ||
| msg = EmailMessage() | ||
| msg.set_content(body) | ||
| msg["Subject"] = subject | ||
| msg["From"] = self.from_email | ||
| msg["To"] = to_email | ||
|
|
||
| try: | ||
| # Connect to the mail server using TLS | ||
| with smtplib.SMTP(self.smtp_server, self.smtp_port) as server: | ||
| # Enable TLS encryption | ||
| server.starttls() | ||
| # Log in to your email account | ||
| server.login(self.from_email, self.from_password) | ||
|
|
||
| # Send the email | ||
| server.send_message(msg) | ||
| logger.info(f"Email sent to {self.anonymize_email(to_email)}") | ||
| except smtplib.SMTPException as e: | ||
| logger.error( | ||
| f"SMTP error occurred when sending email to {self.anonymize_email(to_email)}: {e}" | ||
| ) | ||
| except Exception as e: | ||
| logger.error( | ||
| f"Unexpected error occurred when sending email to {self.anonymize_email(to_email)}: {e}" | ||
| ) | ||
|
|
||
|
|
||
| # Instantiate EmailAlert object | ||
| email_alert = EmailAlert( | ||
| os.getenv("FROM_EMAIL"), | ||
| os.getenv("FROM_PASSWORD"), | ||
| os.getenv("SMTP_SERVER"), | ||
| os.getenv("SMTP_PORT"), | ||
| ) | ||
|
|
||
|
|
||
| def notify_error(): | ||
| from alert import EmailAlert | ||
| from config import LOG_LEVEL, SEND_EMAILS | ||
|
|
||
| __all__ = ["notify_error"] | ||
|
|
||
| LOGGING_CONFIG = { | ||
| "version": 1, | ||
| "formatters": {"standard": {"format": "%(asctime)s - %(levelname)s - %(message)s"}}, | ||
| "handlers": { | ||
| "default": { | ||
| "level": "DEBUG", | ||
| "formatter": "standard", | ||
| "class": "logging.StreamHandler", | ||
| "stream": "ext://sys.stdout", | ||
| }, | ||
| "file": { | ||
| "class": "logging.handlers.RotatingFileHandler", | ||
| "level": "INFO", | ||
| "formatter": "standard", | ||
| "filename": "backup.log", | ||
| "maxBytes": 500000, | ||
| "backupCount": 15, | ||
| }, | ||
| }, | ||
| "loggers": {"": {"level": LOG_LEVEL, "handlers": ["default", "file"]}}, | ||
| } | ||
| logging.config.dictConfig(LOGGING_CONFIG) | ||
|
|
||
|
|
||
| def notify_error(error: str = "") -> None: | ||
| if not SEND_EMAILS: | ||
| return | ||
|
|
||
| if not ( | ||
| os.getenv("EMAIL_SUBJECT") | ||
| and os.getenv("EMAIL_BODY") | ||
| and os.getenv("TO_EMAIL") | ||
| ): | ||
| logger.error( | ||
| "One or more email environment variables are not set. Can't send notification email." | ||
| ) | ||
| return | ||
| email_alert.send( | ||
| os.getenv("EMAIL_SUBJECT"), os.getenv("EMAIL_BODY"), os.getenv("TO_EMAIL") | ||
| EmailAlert().send( | ||
| os.getenv("EMAIL_SUBJECT"), | ||
| os.getenv("EMAIL_BODY") + f"\n\n{error}", | ||
| os.getenv("TO_EMAIL"), | ||
| ) | ||
|
|
||
|
|
||
| for required_url in ["API_KEY", "GET_URL", "SERVERS_URL"]: | ||
| if not locals()[required_url]: | ||
| logger.error(f"{required_url} environment variable not set. Can't proceed without it.") | ||
| def _check_required(name: str) -> None: | ||
| mod = importlib.import_module("config") | ||
| if name not in mod or not mod[name]: | ||
| logging.getLogger(__name__).error( | ||
| f"{name} environment variable missing. Please set it in your .env file." | ||
| ) | ||
| notify_error() | ||
| exit(1) | ||
|
|
||
|
|
||
| _check_required("API_KEY") | ||
| _check_required("GET_URL") | ||
| _check_required("SERVERS_URL") | ||
|
|
||
| if SEND_EMAILS: | ||
| _check_required("FROM_EMAIL") | ||
| _check_required("SMTP_SERVER") | ||
| _check_required("SMTP_PORT") | ||
|
|
||
| _check_required("EMAIL_SUBJECT") | ||
| _check_required("EMAIL_BODY") | ||
| _check_required("TO_EMAIL") |
Oops, something went wrong.