Merge pull request #37 from Cyber-Syntax:refactor/cleanup

refactor: codebase and enhance cleanup processes

Author: Cyber-Syntax

AGENTS.md

@@ -20,5 +20,5 @@ source .venv/bin/activate
 2. Run pytest with following command to ensure all tests pass:
 
 ```bash
-pytest -v -qa --strict-markers
+pytest -v -q --strict-markers
 ```

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 All notable changes to this project will be documented in this file. Commits automatically generated by github actions.
 
+## v0.6.2-beta
 ## v0.6.1-beta
 ## v0.6.0-beta
 ### BREAKING CHANGES

autotarcompress/commands/backup.py

@@ -1,7 +1,7 @@
 """Backup command implementation for creating compressed archives.
 
-This module contains the BackupCommand class that handles the creation of backup archives
-using tar and xz compression.
+This module contains the BackupCommand class that handles the creation of
+backup archives using tar and xz compression.
 """
 
 import datetime
@@ -41,15 +41,15 @@ def execute(self) -> bool:
 
         """
         if not self.config.dirs_to_backup:
-            msg = "No directories configured for backup. Skipping backup."
-            print(msg)
-            self.logger.error("No directories configured for backup. Skipping backup.")
+            self._print_and_log(
+                "No directories configured for backup. Skipping backup.", level="error"
+            )
             return False
         total_size: int = self._calculate_total_size()
         if total_size == 0:
-            msg = "Total backup size is 0 bytes. Nothing to back up."
-            print(msg)
-            self.logger.warning("Total backup size is 0 bytes. Nothing to back up.")
+            self._print_and_log(
+                "Total backup size is 0 bytes. Nothing to back up.", level="warning"
+            )
             return False
         success: bool = self._run_backup_process(total_size)
         if success:
@@ -77,56 +77,64 @@ def _run_backup_process(self, total_size: int) -> bool:
 
         """
         if os.path.exists(self.config.backup_path):
-            print(f"File already exists: {self.config.backup_path}")
-            self.logger.warning("Backup file already exists: %s", self.config.backup_path)
-            if input("Do you want to remove it? (y/n): ").lower() == "y":
-                try:
-                    os.remove(self.config.backup_path)
-                    self.logger.info("Removed existing backup file: %s", self.config.backup_path)
-                except Exception as e:
-                    print(f"Failed to remove existing backup: {e}")
-                    self.logger.error("Failed to remove existing backup: %s", e)
-                    return False
-            else:
-                print("Backup aborted by user.")
-                self.logger.info("Backup aborted by user due to existing file.")
+            self._print_and_log(f"File already exists: {self.config.backup_path}", level="warning")
+            if not self._prompt_overwrite():
+                self._print_and_log("Backup aborted by user due to existing file.", level="info")
+                return False
+            try:
+                os.remove(self.config.backup_path)
+                self.logger.info("Removed existing backup file: %s", self.config.backup_path)
+            # NOTE: Broad except is used here to ensure any file removal error
+            # is caught during backup overwrite prompt. This is a critical IO
+            # operation.
+            except (OSError, PermissionError) as e:
+                self._print_and_log(f"Failed to remove existing backup: {e}", level="error")
                 return False
 
-        exclude_options = " ".join(f"--exclude={path}" for path in self.config.ignore_list)
+        cmd = self._build_tar_command()
+        total_size_gb = total_size / 1024**3
 
-        dir_paths = [os.path.expanduser(path) for path in self.config.dirs_to_backup]
+        self.logger.info("Starting backup to %s", self.config.backup_path)
+        self.logger.info("Total size: %.2f GB", total_size_gb)
 
-        # Properly quote directory paths to handle spaces and special characters
-        quoted_paths = [shlex.quote(path) for path in dir_paths]
+        try:
+            subprocess.run(cmd, shell=True, check=True)
+            self.logger.info("Backup completed successfully")
+            self._print_and_log("Backup completed successfully.", level="info")
+            return True
+        except subprocess.CalledProcessError as e:
+            self._print_and_log(f"Backup failed: {e}", level="error")
+            return False
 
-        # Get CPU count safely
+    def _build_tar_command(self) -> str:
+        """Build the tar+xz command string for backup."""
+        exclude_options = " ".join(f"--exclude={path}" for path in self.config.ignore_list)
+        dir_paths = [os.path.expanduser(path) for path in self.config.dirs_to_backup]
+        quoted_paths = [shlex.quote(path) for path in dir_paths]
         cpu_count = os.cpu_count() or 1
         threads = max(1, cpu_count - 1)
-
-        # HACK: h option is used to follow symlinks
-        cmd = (
+        return (
             f"tar -chf - --one-file-system {exclude_options} "
             f"{' '.join(quoted_paths)} | "
             f"xz --threads={threads} > {self.config.backup_path}"
         )
-        total_size_gb = total_size / 1024**3
-
-        self.logger.info(f"Starting backup to {self.config.backup_path}")
-        self.logger.info(f"Total size: {total_size_gb:.2f} GB")
 
-        try:
-            # FIX: later spinner not working for now
-            # FAILED: not work as expected because of
-            # "| tar: Removing leading `/' from member names" outputs
-            # self._show_spinner(subprocess.Popen(cmd, shell=True))
-            subprocess.run(cmd, shell=True, check=True)
-            self.logger.info("Backup completed successfully")
-            print("Backup completed successfully.")
-            return True
-        except subprocess.CalledProcessError as e:
-            print(f"Backup failed: {e}")
-            self.logger.error(f"Backup failed: {e}")
-            return False
+    def _prompt_overwrite(self) -> bool:
+        """Prompt user to overwrite existing backup file."""
+        response = input("Do you want to remove it? (y/n): ").strip().lower()
+        return response == "y"
+
+    def _print_and_log(self, message: str, level: str = "info") -> None:
+        """Print and log a message at the specified level."""
+        print(message)
+        if level == "info":
+            self.logger.info(message)
+        elif level == "warning":
+            self.logger.warning(message)
+        elif level == "error":
+            self.logger.error(message)
+        else:
+            self.logger.debug(message)
 
     def _save_backup_info(self, total_size: int) -> None:
         """Save backup information to last-backup-info.json."""
@@ -146,10 +154,12 @@ def _save_backup_info(self, total_size: int) -> None:
             with open(info_file_path, "w", encoding="utf-8") as f:
                 json.dump(backup_info, f, indent=2)
 
-            self.logger.info(f"Backup info saved to {info_file_path}")
+            self.logger.info("Backup info saved to %s", info_file_path)
 
-        except Exception as e:
-            self.logger.error(f"Failed to save backup info: {e}")
+            # NOTE: Broad except is used here to ensure any error during backup
+            # info saving is logged, as this is a non-critical reporting step.
+        except (OSError, PermissionError, ValueError) as e:
+            self.logger.error("Failed to save backup info: %s", e)
 
     def _format_size(self, size_bytes: int) -> str:
         """Format size in bytes to human readable format."""

autotarcompress/commands/cleanup.py

@@ -7,6 +7,7 @@
 import datetime
 import logging
 import os
+import shutil
 from pathlib import Path
 
 from autotarcompress.commands.command import Command
@@ -20,49 +21,78 @@ def __init__(self, config: BackupConfig) -> None:
         """Initialize CleanupCommand.
 
         Args:
-            config (BackupConfig): Backup configuration with retention and folder settings.
+            config (BackupConfig): Backup configuration with retention and
+                folder settings.
 
         """
         self.config: BackupConfig = config
         self.logger: logging.Logger = logging.getLogger(__name__)
 
     def execute(self) -> bool:
-        """Delete old backup, encrypted, and decrypted files per retention policy.
+        """Delete old backup, encrypted, and decrypted files.
+
+        Per retention policy.
 
         Returns:
-            bool: Always True (cleanup always completes, even if nothing to delete).
+            bool: Always True (cleanup always completes, even if nothing
+                to delete).
 
         """
         self._cleanup_files(".tar.xz", self.config.keep_backup)
-        self._cleanup_files(".tar.xz.enc", self.config.keep_enc_backup)
         self._cleanup_files(".tar.xz-decrypted", self.config.keep_backup)
+        self._cleanup_files(".tar-extracted", self.config.keep_backup)
+        self._cleanup_files(".tar.xz.enc", self.config.keep_enc_backup)
         return True
 
     def _cleanup_files(self, ext: str, keep_count: int) -> None:
-        """Delete old files by extension, keeping only the most recent as configured.
+        """Delete old files by extension.
+
+        Keeping only the most recent as configured.
 
         Args:
             ext (str): File extension to filter for cleanup.
             keep_count (int): Number of recent files to keep.
 
         """
+
+        def _extract_date_from_filename(filename: str) -> datetime.datetime:
+            """Extract datetime from filename for sorting.
+
+            Args:
+                filename (str): Filename with date format at start.
+
+            Returns:
+                datetime.datetime: Parsed datetime object.
+
+            """
+            return datetime.datetime.strptime(filename.split(".")[0], "%d-%m-%Y")
+
         backup_folder: Path = Path(self.config.backup_folder)
         files: list[str] = sorted(
             [f for f in os.listdir(backup_folder) if f.endswith(ext)],
-            key=lambda x: datetime.datetime.strptime(x.split(".")[0], "%d-%m-%Y"),
+            key=_extract_date_from_filename,
         )
+
         files_to_delete: list[str] = files if keep_count == 0 else files[:-keep_count]
         if not files_to_delete:
             msg = f"No old '{ext}' files to remove."
             print(msg)
             self.logger.info("No old '%s' files to remove.", ext)
-            return
+            return None
+
         for old_file in files_to_delete:
             file_path = backup_folder / old_file
             try:
-                file_path.unlink()
-                self.logger.info("Deleted old backup: %s", old_file)
-                print(f"Deleted old backup: {old_file}")
-            except Exception as e:
+                if file_path.is_dir():
+                    # Remove directory (recursively if not empty)
+                    shutil.rmtree(file_path)
+                    self.logger.info("Deleted old backup directory: %s", old_file)
+                    print(f"Deleted old backup directory: {old_file}")
+                else:
+                    file_path.unlink()
+                    self.logger.info("Deleted old backup: %s", old_file)
+                    print(f"Deleted old backup: {old_file}")
+            except (OSError, PermissionError) as e:
                 self.logger.error("Failed to delete %s: %s", old_file, e)
                 print(f"Failed to delete {old_file}: {e}")
+        return None

autotarcompress/commands/decrypt.py

@@ -16,8 +16,7 @@
 
 
 class DecryptCommand(Command):
-    """
-    Command to securely decrypt backup archives using OpenSSL with PBKDF2.
+    """Command to securely decrypt backup archives using OpenSSL with PBKDF2.
 
     This class ensures decryption parameters match those used for encryption and
     verifies file integrity post-decryption. Logging uses %s formatting for performance.
@@ -26,12 +25,12 @@ class DecryptCommand(Command):
     PBKDF2_ITERATIONS: int = 600000  # Must match encryption iterations
 
     def __init__(self, config: BackupConfig, file_path: str) -> None:
-        """
-        Initialize the DecryptCommand.
+        """Initialize the DecryptCommand.
 
         Args:
             config (BackupConfig): The backup configuration object.
             file_path (str): Path to the encrypted file to decrypt.
+
         """
         self.config: BackupConfig = config
         self.file_path: str = file_path
@@ -40,18 +39,18 @@ def __init__(self, config: BackupConfig, file_path: str) -> None:
         self._safe_cleanup = ContextManager()._safe_cleanup
 
     def execute(self) -> bool:
-        """
-        Perform secure decryption with matched PBKDF2 parameters.
+        """Perform secure decryption with matched PBKDF2 parameters.
 
         Returns:
             bool: True if decryption and integrity check succeed, False otherwise.
+
         """
         output_path: str = os.path.splitext(self.file_path)[0]
         decrypted_path: str = f"{output_path}-decrypted"
 
         # Use password context manager to securely obtain password
-        with self._password_context() as password:
-            if not password:
+        with self._password_context() as password:  # type: ignore[var-annotated]
+            if password is None:
                 return False
 
             cmd: list[str] = [
@@ -81,12 +80,11 @@ def execute(self) -> bool:
                     timeout=300,
                     shell=False,
                 )
-                self._verify_integrity(decrypted_path)
-                return True
-            except subprocess.CalledProcessError as e:
+            except subprocess.CalledProcessError as exc:
                 # Log sanitized error output to avoid leaking sensitive data
                 self.logger.error(
-                    "Decryption failed: %s", self._sanitize_logs(e.stderr)
+                    "Decryption failed: %s",
+                    self._sanitize_logs(exc.stderr),
                 )
                 self._safe_cleanup(decrypted_path)
                 return False
@@ -95,12 +93,15 @@ def execute(self) -> bool:
                 self._safe_cleanup(decrypted_path)
                 return False
 
+            self._verify_integrity(decrypted_path)
+            return True
+
     def _verify_integrity(self, decrypted_path: str) -> None:
-        """
-        Verify decrypted file matches original backup checksum.
+        """Verify decrypted file matches original backup checksum.
 
         Args:
             decrypted_path (str): Path to the decrypted file.
+
         """
         original_path: str = os.path.splitext(self.file_path)[0]
         if os.path.exists(original_path):
@@ -117,35 +118,39 @@ def _verify_integrity(self, decrypted_path: str) -> None:
                 self.logger.error("Integrity check failed")
 
     def _calculate_sha256(self, file_path: str) -> str:
-        """
-        Calculate SHA256 checksum for a file.
+        """Calculate SHA256 checksum for a file.
 
         Args:
             file_path (str): Path to the file.
 
         Returns:
             str: SHA256 hex digest of the file contents.
+
         """
         sha256 = hashlib.sha256()
-        with open(file_path, "rb") as f:
+        with open(file_path, "rb") as file_obj:
             while True:
-                data = f.read(65536)
+                data = file_obj.read(65536)
                 if not data:
                     break
                 sha256.update(data)
         return sha256.hexdigest()
 
     def _sanitize_logs(self, output: bytes) -> str:
-        """
-        Sanitize log output to redact sensitive information.
+        """Sanitize log output to redact sensitive information.
 
         Args:
             output (bytes): Raw stderr output from subprocess.
 
         Returns:
             str: Sanitized string safe for logging.
+
         """
         # Redact password and IP addresses from logs for security
         sanitized = re.sub(rb"password=[^\s]*", b"password=[REDACTED]", output)
-        sanitized = re.sub(rb"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b", b"[IP_REDACTED]", sanitized)
+        sanitized = re.sub(
+            rb"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b",
+            b"[IP_REDACTED]",
+            sanitized,
+        )
         return sanitized.decode("utf-8", errors="replace")