diff --git a/alert.py b/alert.py new file mode 100644 index 0000000..87a55ee --- /dev/null +++ b/alert.py @@ -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}" + ) diff --git a/api_request.py b/api_request.py index f64e41b..cb35272 100644 --- a/api_request.py +++ b/api_request.py @@ -11,19 +11,21 @@ @lru_cache(1) -def get_session(): +def _get_session(): retry_strategy = Retry( total=MAX_RETRIES, backoff_factor=RETRY_BACKOFF_FACTOR, ) session = requests.Session() - session.headers.update({ - "Authorization": "Bearer " + API_KEY, - # https://dashflo.net/docs/api/pterodactyl/v1/ - "Content-Type": "application/json", - "Accept": "application/json", - }) + session.headers.update( + { + "Authorization": "Bearer " + API_KEY, + # https://dashflo.net/docs/api/pterodactyl/v1/ + "Content-Type": "application/json", + "Accept": "application/json", + } + ) adapter = HTTPAdapter(max_retries=retry_strategy) @@ -36,11 +38,12 @@ def get_session(): def request(url, method: str = "GET", data=None) -> dict: for retry in range(MAX_RETRIES): try: - response = get_session().request(method=method, url=url, data=data) + response = _get_session().request(method=method, url=url, data=data) if not response: - raise requests.exceptions.RequestException(response.json()["errors"][0]["detail"]) + raise requests.exceptions.RequestException( + response.json()["errors"][0]["detail"] + ) - response.raise_for_status() return response.json() except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: logger.error(f"Network Error: {str(e)}") diff --git a/backup.py b/backup.py index f53d838..343aec6 100644 --- a/backup.py +++ b/backup.py @@ -1,11 +1,13 @@ +import logging import os import time -from pprint import pprint import requests from api_request import request -from common import GET_URL, POST_BACKUP_SCRIPT, ROTATE, SERVERS_URL, logger +from config import GET_URL, POST_BACKUP_SCRIPT, ROTATE, SERVERS_URL + +logger = logging.getLogger(__name__) # remove backups when limits reached @@ -17,9 +19,7 @@ def remove_old_backup(server): url = f"{SERVERS_URL}{server_id}/backups" response = request(url) - backups = sorted( - response["data"], key=lambda b: b["attributes"]["created_at"] - ) + backups = sorted(response["data"], key=lambda b: b["attributes"]["created_at"]) if len(backups) >= backup_limit: # backup limit reached @@ -43,7 +43,7 @@ def remove_old_backup(server): f" removing backup: \"{backups[i]['attributes']['name']}\"" ) - request(url, method='DELETE') + request(url, method="DELETE") time.sleep(2) else: @@ -55,12 +55,12 @@ def remove_old_backup(server): ) -def backup_servers(server_list): +def backup_servers(all_servers): failed_servers = [] - server_list = server_list["data"] + all_servers = all_servers["data"] - for server in server_list: + for server in all_servers: server_attr = server["attributes"] server_id = server_attr["identifier"] server_name = server_attr["name"] @@ -77,14 +77,14 @@ def backup_servers(server_list): continue url = f"{SERVERS_URL}{server_id}/backups" - backup = request(url, method='POST') + backup = request(url, method="POST") backup_uuid = backup["attributes"]["uuid"] logger.info(" backup started") if POST_BACKUP_SCRIPT: # we should only run this when the backup has been finished.... - # this will prevent that the backups are made concurrently, thats OK, maybe it is + # this will prevent that the backups are made concurrently, that's OK, maybe it is # even better as it reduces overall load logger.info(" waiting for backup to finish...") while True: @@ -111,6 +111,6 @@ def run_script(server_id, backup_uuid): logger.error(f" post backup script: failed with exit status {exit_status}") -if __name__ == '__main__': +if __name__ == "__main__": server_list = request(GET_URL) backup_servers(server_list) diff --git a/common.py b/common.py index 99a3390..ede8c8b 100644 --- a/common.py +++ b/common.py @@ -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") diff --git a/config.py b/config.py new file mode 100644 index 0000000..1254076 --- /dev/null +++ b/config.py @@ -0,0 +1,34 @@ +import os + +from dotenv import load_dotenv + +__all__ = [ + "API_KEY", + "SEND_EMAILS", + "GET_URL", + "SERVERS_URL", + "MAX_RETRIES", + "RETRY_BACKOFF_FACTOR", + "ROTATE", + "POST_BACKUP_SCRIPT", + "LOG_LEVEL", +] + + +# Environment .env +load_dotenv() + +# 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 +LOG_LEVEL = os.getenv("LOG_LEVEL") or "ERROR" + +EMAIL_SUBJECT = os.getenv("EMAIL_SUBJECT") +EMAIL_BODY = os.getenv("EMAIL_BODY") +TO_EMAIL = os.getenv("TO_EMAIL")