Bugfixes and error checks.

release-versions/latest.txt

@@ -1 +1 @@
-2024.05.27-11
+2024.05.27-12

root/app/calibre_info.py

@@ -38,9 +38,8 @@ def __init__(self, toml_path: str, manager: mp.Manager):
         self.location = calibre_config.get("path")
         self.username = calibre_config.get("username")
         self.password = calibre_config.get("password")
-        self.default_ini = self._append_filename(calibre_config.get("default_ini"), "defaults.ini")
-        self.personal_ini = self._append_filename(calibre_config.get("personal_ini"), "personal.ini")
-
+        self.default_ini = self._get_ini_file(calibre_config, "default_ini", "defaults.ini")
+        self.personal_ini = self._get_ini_file(calibre_config, "personal_ini", "personal.ini")
         # Create a lock for thread-safe operations
         self.lock = manager.Lock()
 
@@ -60,6 +59,14 @@ def _append_filename(path: str, filename: str) -> str:
         if path and not path.endswith(filename):
             return os.path.join(path, filename)
         return path
+
+
+    def _get_ini_file(self, calibre_config: dict, config_key:str, default_filename: str):
+        ini_file = self._append_filename(calibre_config.get(config_key), default_filename)
+        if ini_file and not os.path.isfile(ini_file):
+            ff_logging.log_failure(f"File {ini_file} does not exist.")
+            ini_file = ""
+        return ini_file
 
     # Check if Calibre is installed
     def check_installed(self) -> bool:

root/app/calibre_info_test.py

@@ -70,14 +70,17 @@ class ConfigCase(NamedTuple):
             ),
         ]
     )
+    @patch("os.path.isfile")
     @patch("builtins.open", new_callable=mock_open)
     @patch("multiprocessing.Manager")
     @patch("calibre_info.ff_logging.log_failure")
     def test_calibre_info_init(
-        self, toml_path, config, expected_config, mock_log, mock_manager, mock_file
+        self, toml_path, config, expected_config, mock_log, mock_manager, mock_file, mock_isfile
     ):
         mock_file.return_value.read.return_value = str(config).encode()
         mock_manager.return_value = MagicMock()
+        # TODO: Actually test this.
+        mock_isfile.return_value = True
         if isinstance(expected_config, dict):
             calibre_info = CalibreInfo(toml_path, mock_manager())
             self.assertEqual(calibre_info.location, expected_config["calibre"]["path"])

root/app/pushbullet_notification.py

@@ -40,7 +40,7 @@ def send_notification(self, title: str, body: str) -> None:
         # If Pushbullet is enabled, send the notification
         if self.enabled:
             try:
-                ff_logging.log(f"\tSending Pushbullet notification: {title} - {body}", "OKBLUE")
+                ff_logging.log(f"\tSending Pushbullet notification: {title} - {body}", "OKGREEN")
                 self.pb.push_note(title, body)
             except PushbulletError as e:
                 message = f"\tFailed to send Pushbullet notification: {e}"

root/app/url_ingester.py

@@ -2,8 +2,6 @@
 import socket
 from contextlib import contextmanager
 import time
-import contextlib
-import os
 import logging
 
 from fanficfare import geturls
@@ -20,6 +18,16 @@ def set_timeout(time):
         yield
     finally:
         socket.setdefaulttimeout(old_timeout)
+
+@contextmanager
+def set_logging_level():
+    """Set the logging level to CRITICAL."""
+    old_level = logging.root.manager.disable
+    logging.disable(logging.CRITICAL)
+    try:
+        yield
+    finally:
+        logging.disable(old_level)
 
 class EmailInfo:
     """
@@ -60,19 +68,14 @@ def get_urls(self) -> set[str]:
         """
         urls = set()
                 # Save the current logging level
-        old_level = logging.root.manager.disable
 
-        # Set the logging level to CRITICAL
-        logging.disable(logging.CRITICAL)
         with set_timeout(55):
             try:
                 # Get URLs from the email account
-                urls = geturls.get_urls_from_imap(self.server, self.email, self.password, self.mailbox)
+                with set_logging_level():
+                    urls = geturls.get_urls_from_imap(self.server, self.email, self.password, self.mailbox)
             except Exception as e:
                 ff_logging.log_failure(f"Failed to get URLs: {e}")
-            finally:
-                # Restore the old logging level
-                logging.disable(old_level)
         return urls
 
 

root/app/url_worker.py

@@ -235,7 +235,7 @@ def url_worker(queue: mp.Queue, cdb: calibre_info.CalibreInfo, pushbullet_info:
                 if cdb.personal_ini:
                     copyfile(cdb.personal_ini, join(temp_dir, "personal.ini"))
                 # Execute the command and get the output
-                output = check_output(command, shell=True, stderr=STDOUT, stdin=PIPE,).decode("utf-8")
+                output = check_output(command, shell=True, stderr=STDOUT, stdin=PIPE).decode("utf-8")
                 ff_logging.log(f"\t({site}) Output: {output}", "OKBLUE")
             except Exception as e:
                 # If the command fails, log the failure and continue to the next iteration