feat: refine logging and add html mailing (#34)

* feat: set logger to own class

* feat: add custom log.success

* feat: add CustomLogger success

* fix: comment varChecker

* style: minor logging changes

* style: run linter

* feat: update chromedriver version

* feat: add custom success message

* feat: beautiful html email

* fix: linting2

* feat: add neutral mail and appendix

* feat: check linting

* feat: check linting

* fix: testing

* fix: testing

.github/workflows/build-and-push-on-merge.yml

@@ -7,9 +7,12 @@ env:
 on:
   workflow_dispatch:
   push:
-    branches:
-      - master
-
+    branches: master
+    paths:
+      - src/**
+      - poetry.Dockerfile
+      - pyproject.toml
+      - CHANGELOG.md
 
 jobs:
   get-tag:

justfile

@@ -91,6 +91,7 @@ fmt:
     find . -type d -name ".tox" -exec rm -rf {} +
     find . -type d -name "_build" -path "*/docs/_build" -exec rm -rf {} +
     find . -type d -name "logs" -exec rm -rf {} +
+    find . -type d -name ".ruff_cache" -exec rm -rf {} +
 
 # Remove the installed virtual environment
 clean-venv: 

poetry.Dockerfile

@@ -13,7 +13,7 @@ ENV TZ="Europe/Berlin"
 
 # Install Chromedriver
 RUN apt-get install -yqq unzip
-ENV ChromedriverVersion="126.0.6478.126"
+ENV ChromedriverVersion="126.0.6478.182"
 RUN wget -O /tmp/chromedriver.zip https://storage.googleapis.com/chrome-for-testing-public/$ChromedriverVersion/linux64/chromedriver-linux64.zip
 RUN unzip /tmp/chromedriver.zip chromedriver-linux64/chromedriver -d /usr/local/bin/
 

pyproject.toml

@@ -3,7 +3,12 @@ name = "slotBooker"
 version = "2.3.0"
 description = ""
 authors = ["Sebastian Blum <hello@seblum.com>"]
-packages = [{include = "slotbooker", from = "src"}]
+# packages = [{include = "slotbooker", from = "src"}]
+
+
+[tool.poetry.plugins.dotenv]
+ignore = "false"
+location = ".env"
 
 [tool.poetry.dependencies]
 python = "^3.11"
@@ -19,7 +24,6 @@ coverage = "^7.5.3"
 pytest = "^8.2.2"
 ruff = "^0.5.0"
 
-
 [tool.poetry.scripts]
 slotBooker = "slotbooker.booker:main"
 slotBookerDev = "slotbooker.booker_dev:main"
@@ -43,4 +47,4 @@ exclude = ["migrations", "node_modules", "venv"]
 # # Configure per-file-ignores
 # per-file-ignores = [
 #     "tests/test_*.py: F401, F403"
-# ]
+# ]

src/slotbooker/alerts_and_errors.py

@@ -42,23 +42,22 @@ def alert_is_present(self) -> Optional[object]:
             error_message="Timed out waiting for alert to appear.",
         )
         if alert:
-            logging.info("! Alert present")
+            logging.warning("Alert present")
             return self.driver.switch_to.alert
         return None
 
     def evaluate_alert(self, alert_obj: object, prioritize_waiting_list: Any) -> Enum:
         """Determines the type of alert based on its text."""
         alert_text = alert_obj.text
         if self._contains_keywords(alert_text, ["waiting list", "Warteliste"]):
-            logging.info("! Class full")
+            logging.warning("Class full")
             return self._handle_waiting_list_booking(prioritize_waiting_list, alert_obj)
         elif self._contains_keywords(
             alert_text, ["wirklich", "stornieren", "stornieren?"]
         ):
             return self._handle_cancel_slot(alert_obj)
         else:
-            logging.info(AlertTypes.NotIdentifyAlert.value)
-            logging.info(f"! Alert message: {alert_text}")
+            logging.warning(f"{AlertTypes.NotIdentifyAlert.value}: {alert_text}")
             return self.booking_helper.continue_booking_process()
 
     def _contains_keywords(self, text: str, keywords: list) -> bool:
@@ -69,23 +68,23 @@ def _handle_waiting_list_booking(
     ) -> bool:
         """Handle booking waiting list option in the alert."""
         if prioritize_waiting_list:
-            logging.info("! Booking waiting list...")
+            logging.info("Booking waiting list...")
             alert_obj.accept()
-            logging.info("| Waiting list booked")
+            logging.info("Waiting list booked")
             return self.booking_helper.stop_booking_process()
         else:
             logging.info(
-                f"! Parameter 'wl' is set to {prioritize_waiting_list} > Skipping waiting list"
+                f"Parameter 'wl' is set to {prioritize_waiting_list} > Skipping waiting list"
             )
             alert_obj.dismiss()
-            logging.info("> Looking for further slots...")
+            logging.info("Looking for further slots...")
             return self.booking_helper.continue_booking_process()
 
     def _handle_cancel_slot(self, alert_obj: object) -> bool:
         """Handle aborting the canceling of a slot."""
-        logging.info("! Aborted canceling slot...")
+        logging.warning("Aborted canceling slot...")
         alert_obj.dismiss()
-        logging.info("> Looking for further slots...")
+        logging.info("Looking for further slots...")
         return self.booking_helper.continue_booking_process()
 
     def error_is_present(self) -> Optional[str]:
@@ -96,7 +95,7 @@ def error_is_present(self) -> Optional[str]:
             )
         )
         if error_window:
-            logging.info("! Error !")
+            # logging.error("! Error !")
             error_text = self.driver.find_element(
                 By.XPATH, self.xpath_helper.get_xpath_error_text_window()
             ).text
@@ -112,10 +111,9 @@ def evaluate_error(self, error_text: str) -> bool:
         }
         result = error_map.get(error_text, False)
         if result is False:
-            logging.info(f"! {AlertTypes.NotIdentifyError.value}")
-            logging.info(f"! Error message: {error_text}")
+            logging.error(f"{AlertTypes.NotIdentifyError.value}: {error_text}")
         else:
-            logging.info(f"! {error_text}")
+            logging.error(f"Another Error occured: {error_text}")
         return result
 
     def login_error_is_present(self):
@@ -126,12 +124,8 @@ def login_error_is_present(self):
         )
         if alert_div:
             alert_text = alert_div.text
-            print(alert_div)
-            print(alert_text)
-            print("Alert message is present.")
             if self._contains_keywords(alert_text, ["credentials", "Fehler"]):
-                print(f"! Credentials wrong {alert_text}")
-                logging.info(f"! Credentials wrong {alert_text}")
+                logging.error(f"Credentials wrong {alert_text}")
                 return True
         else:
             print("Alert message is not present.")

src/slotbooker/booker.py

@@ -4,9 +4,9 @@
 from selenium.common.exceptions import SessionNotCreatedException, NoSuchDriverException
 
 from .utils.driver import close_driver, get_driver
-from .utils.logging import LogHandler
+from .utils.custom_logger import LogHandler
+from .utils.mailhandler import MailHandler
 from .ui_interaction import Booker
-from .utils.settings import set_credentials
 
 # Load configuration files
 config_path = os.path.join(os.path.dirname(__file__), "utils/config.yaml")
@@ -17,6 +17,16 @@
 with open(classes_path, "r") as file:
     classes = yaml.safe_load(file)
 
+env_vars_to_check = [
+    "OCTIV_USERNAME",
+    "OCTIV_PASSWORD",
+    "DAYS_BEFORE_BOOKABLE",
+    "EXECUTION_BOOKING_TIME",
+]
+# Create an instance of ClassVarHelper
+# helper = ClassVarHelper(env_vars_to_check)
+# helper.check_vars()
+
 
 def main(retry: int = 3):
     """Slotbooker Main Function.
@@ -38,14 +48,11 @@ def main(retry: int = 3):
 
         >>> main()
     """
-    # start writing output to logfile
-    # file, orig_stdout, dir_log_file = start_logging()
+    log_hander = LogHandler()
+    driver = get_driver(chromedriver=config.get("chromedriver"))
 
     if os.environ.get("IS_TEST"):
-        logging.info("! Test env")
-        print("! Test env")
-        driver = get_driver(chromedriver=config.get("chromedriver"))
-
+        logging.info("Testing Docker Container")
         booker = Booker(
             driver=driver,
             days_before_bookable=0,
@@ -57,70 +64,70 @@ def main(retry: int = 3):
 
         login_failed = booker.login(username=user, password=password)
         if login_failed:
-            logging.info("! Login failed as expected")
-            print("TEST OK")
-        exit()
+            logging.success(message="TEST OK | Login failed as expected")
 
-    log_hander = LogHandler()
-    dir_log_file = log_hander.setup_log_dir()
-    logging.basicConfig(
-        filename=dir_log_file,
-        filemode="w",
-        encoding="utf-8",
-        format="%(asctime)s %(message)s",
-        level=logging.INFO,
-    )
+        exit()
 
     # Retrieve environment variables
     user = os.environ.get("OCTIV_USERNAME")
     password = os.environ.get("OCTIV_PASSWORD")
     days_before_bookable = int(os.environ.get("DAYS_BEFORE_BOOKABLE", 0))
     execution_booking_time = os.environ.get("EXECUTION_BOOKING_TIME")
-
-    # Ensure credentials are set
-    if not user or not password:
-        logging.info("USERNAME and PASSWORD not set")
-        set_credentials()
+    booked_successful = False
+
+    logging.info(f"Log in as: {user}")
+
+    for attempt in range(1, retry + 1):
+        try:
+            booker = Booker(
+                driver=driver,
+                days_before_bookable=days_before_bookable,
+                base_url=config.get("base_url"),
+                execution_booking_time=execution_booking_time,
+            )
+
+            booker.login(username=user, password=password)
+            booking_day, booking_date = booker.switch_day()
+            booking_date = f"{booking_day}, {booking_date}"
+
+            booked_successful, class_slot, time_slot = booker.book_class(
+                class_dict=classes.get("class_dict"),
+                booking_action=classes.get("book_class"),
+            )
+
+            close_driver(driver)
+            logging.success(f"Attempt {attempt}: OctivBooker succeeded")
+            break
+        except (SessionNotCreatedException, NoSuchDriverException) as e:
+            logging.warning(
+                f"Attempt {attempt}: OctivBooker failed due to driver issue"
+            )
+            logging.error(e, exc_info=True)
+        except Exception as e:
+            logging.warning(
+                f"Attempt {attempt}: OctivBooker failed due to unexpected error"
+            )
+            logging.error(e, exc_info=True)
+
+    mail_handler = MailHandler(format="html")
+    if booked_successful:
+        mail_handler.send_successful_booking_email(
+            booking_date=booking_date,
+            booking_time=time_slot,
+            booking_name=class_slot,
+            attachment_path=log_hander.get_log_file_path(),
+        )
+    elif not booked_successful and class_slot is None and time_slot is None:
+        mail_handler.send_no_classes_email(
+            booking_date=booking_date, attachment_path=log_hander.get_log_file_path()
+        )
     else:
-        logging.info(f"USER: {user}")
-
-        for attempt in range(retry):
-            try:
-                driver = get_driver(chromedriver=config.get("chromedriver"))
-
-                booker = Booker(
-                    driver=driver,
-                    days_before_bookable=days_before_bookable,
-                    base_url=config.get("base_url"),
-                    execution_booking_time=execution_booking_time,
-                )
-
-                booker.login(username=user, password=password)
-                booker.switch_day()
-                booker.book_class(
-                    class_dict=classes.get("class_dict"),
-                    booking_action=classes.get("book_class"),
-                )
-
-                close_driver(driver)
-                logging.info(f"| [{attempt + 1}] OctivBooker succeeded")
-                response = "SUCCESS"
-                break
-            except (SessionNotCreatedException, NoSuchDriverException) as e:
-                logging.info(f"| [{attempt + 1}] OctivBooker failed")
-                logging.error(e, exc_info=True)
-                response = "FAILED"
-                continue
-            except Exception as e:
-                logging.info(f"| [{attempt + 1}] OctivBooker failed")
-                logging.error(e, exc_info=True)
-                response = "FAILED"
-                continue
-
-        html_file = log_hander.convert_logs_to_html()
-        # stop_logging(file, orig_stdout)
-        # log_hander.send_logs_to_mail(dir_log_file,response)
-        log_hander.send_logs_to_mail(html_file,response,format="html")
+        mail_handler.send_unsuccessful_booking_email(
+            booking_date=booking_date,
+            booking_time=time_slot,
+            booking_name=class_slot,
+            attachment_path=log_hander.get_log_file_path(),
+        )
 
 
 if __name__ == "__main__":