Merge pull request #34 from Cyber-Syntax:refactor/improve-code-structure

refactor: code style and add cleanup support for decrypted files

Author: Cyber-Syntax

AGENTS.md

@@ -5,17 +5,20 @@ This file provides guidance to agents when working with code in this repository.
 ## General Guidelines
 
 1. Modern Python First: Use Python 3.12+ features extensively - built-in generics, pattern matching, and dataclasses.
-2. Async-First Architecture: All I/O operations must be async. Use modern async patterns like `asyncio.TaskGroup` for concurrency.
-3. Type Safety: Full type annotations on all functions including return types. Use modern syntax (`dict[str, int]`, `str | None`).
-4. KISS Principle: Aim for simplicity and clarity. Avoid unnecessary abstractions or metaprogramming.
-5. DRY with Care: Reuse code appropriately but avoid over-engineering. Each command handler has single responsibility.
-6. Performance-Conscious: Use `@dataclass(slots=True)` when object count justifies it, orjson for JSON, and async-safe patterns over explicit locks.
+2. Type Safety: Full type annotations on all functions including return types. Use modern syntax (`dict[str, int]`, `str | None`).
+3. KISS Principle: Aim for simplicity and clarity. Avoid unnecessary abstractions or metaprogramming.
+4. DRY with Care: Reuse code appropriately but avoid over-engineering. Each command handler has single responsibility.
 
-## Activate venv before any test execution
+## Test after any change
 
-Unit test located in `tests/` directory
+1. Activate venv before any test execution:
 
 ```bash
 source .venv/bin/activate
+```
+
+2. Run pytest with following command to ensure all tests pass:
+
+```bash
 pytest -v -qa --strict-markers
 ```

autotarcompress/commands.py

@@ -1,21 +1,14 @@
-"""Command pattern implementations for backup operations.
+"""Re-export command classes for backward compatibility in backup operations."""
 
-This module re-exports command classes from the commands package
-for backward compatibility.
-"""
+from autotarcompress.commands.backup import BackupCommand
+from autotarcompress.commands.cleanup import CleanupCommand
+from autotarcompress.commands.command import Command
+from autotarcompress.commands.decrypt import DecryptCommand
+from autotarcompress.commands.encrypt import EncryptCommand
+from autotarcompress.commands.extract import ExtractCommand
+from autotarcompress.commands.info import InfoCommand
 
-# Re-export command classes for backward compatibility
-from autotarcompress.commands import (
-    BackupCommand,
-    CleanupCommand,
-    Command,
-    DecryptCommand,
-    EncryptCommand,
-    ExtractCommand,
-    InfoCommand,
-)
-
-__all__ = [
+__all__: list[str] = [
     "BackupCommand",
     "CleanupCommand",
     "Command",

autotarcompress/commands/backup.py

@@ -21,46 +21,78 @@
 
 
 class BackupCommand(Command):
-    """Concrete command to perform backup"""
+    """Command to create compressed backup archives using tar and xz."""
 
-    def __init__(self, config: BackupConfig):
-        self.config = config
-        self.logger = logging.getLogger(__name__)
+    def __init__(self, config: BackupConfig) -> None:
+        """Initialize BackupCommand.
+
+        Args:
+            config (BackupConfig): Backup configuration object.
+
+        """
+        self.config: BackupConfig = config
+        self.logger: logging.Logger = logging.getLogger(__name__)
 
     def execute(self) -> bool:
-        """Execute backup process."""
-        if not self.config.dirs_to_backup:
-            self.logger.error("No directories configured for backup")
-            return False
+        """Execute backup process.
 
-        total_size = self._calculate_total_size()
-        success = self._run_backup_process(total_size)
+        Returns:
+            bool: True if backup succeeded, False otherwise.
 
-        # Save backup info if backup was successful
+        """
+        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.")
+            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.")
+            return False
+        success: bool = self._run_backup_process(total_size)
         if success:
             self._save_backup_info(total_size)
-
         return success
 
     def _calculate_total_size(self) -> int:
+        """Calculate total size of all directories to back up.
+
+        Returns:
+            int: Total size in bytes.
+
+        """
         calculator = SizeCalculator(self.config.dirs_to_backup, self.config.ignore_list)
         return calculator.calculate_total_size()
 
     def _run_backup_process(self, total_size: int) -> bool:
-        """Run the backup process and return success status."""
-        # Check is there any file exist with same name
+        """Run the backup process and return success status.
+
+        Args:
+            total_size (int): Total size of files to back up, in bytes.
+
+        Returns:
+            bool: True if backup succeeded, False otherwise.
+
+        """
         if os.path.exists(self.config.backup_path):
-            print(f"File already exist: {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":
-                os.remove(self.config.backup_path)
+                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.")
                 return False
 
-        exclude_options = " ".join([f"--exclude={path}" for path in self.config.ignore_list])
-
-        # TODO: need to fix this exclude option
-        # TEST: without os.path.basename which it is not working
-        # exclude_options += f" --exclude={self.config.backup_folder}"
+        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]
 
@@ -73,13 +105,14 @@ def _run_backup_process(self, total_size: int) -> bool:
 
         # HACK: h option is used to follow symlinks
         cmd = (
-            f"tar -chf - --one-file-system {exclude_options} {' '.join(quoted_paths)} | "
+            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} GB")
+        self.logger.info(f"Total size: {total_size_gb:.2f} GB")
 
         try:
             # FIX: later spinner not working for now
@@ -88,8 +121,10 @@ def _run_backup_process(self, total_size: int) -> bool:
             # 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
 

autotarcompress/commands/cleanup.py

@@ -14,37 +14,49 @@
 
 
 class CleanupCommand(Command):
-    """Concrete command to perform cleanup of old backups"""
+    """Command to clean up old backup, encrypted, and decrypted files."""
 
-    def __init__(self, config: BackupConfig):
-        self.config = config
-        self.logger = logging.getLogger(__name__)
+    def __init__(self, config: BackupConfig) -> None:
+        """Initialize CleanupCommand.
+
+        Args:
+            config (BackupConfig): Backup configuration with retention and folder settings.
+
+        """
+        self.config: BackupConfig = config
+        self.logger: logging.Logger = logging.getLogger(__name__)
 
     def execute(self) -> bool:
-        """Execute cleanup process for old backup files"""
+        """Delete old backup, encrypted, and decrypted files per retention policy.
+
+        Returns:
+            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)
         return True
 
     def _cleanup_files(self, ext: str, keep_count: int) -> None:
-        """Delete old backup files based on file extension and retention count.
+        """Delete old files by extension, keeping only the most recent as configured.
 
         Args:
-            ext: File extension to filter for cleanup
-            keep_count: Number of recent files to keep
+            ext (str): File extension to filter for cleanup.
+            keep_count (int): Number of recent files to keep.
 
         """
-        backup_folder = Path(self.config.backup_folder)
-
-        # Get all files with the specified extension
-        files = sorted(
+        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"),
         )
-
-        # Delete files exceeding the retention count
-        files_to_delete = files if keep_count == 0 else files[:-keep_count]
-
+        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
         for old_file in files_to_delete:
             file_path = backup_folder / old_file
             try:

autotarcompress/commands/command.py

@@ -7,8 +7,13 @@
 
 
 class Command(ABC):
-    """Command interface for backup manager"""
+    """Abstract command interface for backup manager operations."""
 
     @abstractmethod
     def execute(self) -> bool:
-        """Execute the command operation"""
+        """Execute the command operation.
+
+        Returns:
+            bool: True if command succeeded, False otherwise.
+
+        """

autotarcompress/commands/decrypt.py

@@ -16,27 +16,45 @@
 
 
 class DecryptCommand(Command):
-    """Concrete command to perform decryption with secure parameters"""
-
-    PBKDF2_ITERATIONS = 600000  # Must match encryption iterations
-
-    def __init__(self, config: BackupConfig, file_path: str):
-        self.config = config
-        self.file_path = file_path
-        self.logger = logging.getLogger(__name__)
+    """
+    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.
+    """
+
+    PBKDF2_ITERATIONS: int = 600000  # Must match encryption iterations
+
+    def __init__(self, config: BackupConfig, file_path: str) -> None:
+        """
+        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
+        self.logger: logging.Logger = logging.getLogger(__name__)
         self._password_context = ContextManager()._password_context
         self._safe_cleanup = ContextManager()._safe_cleanup
 
     def execute(self) -> bool:
-        """Secure decryption with matched PBKDF2 parameters"""
-        output_path = os.path.splitext(self.file_path)[0]
-        decrypted_path = f"{output_path}-decrypted"
+        """
+        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:
                 return False
 
-            cmd = [
+            cmd: list[str] = [
                 "openssl",
                 "enc",
                 "-d",
@@ -66,7 +84,10 @@ def execute(self) -> bool:
                 self._verify_integrity(decrypted_path)
                 return True
             except subprocess.CalledProcessError as e:
-                self.logger.error(f"Decryption failed: {self._sanitize_logs(e.stderr)}")
+                # Log sanitized error output to avoid leaking sensitive data
+                self.logger.error(
+                    "Decryption failed: %s", self._sanitize_logs(e.stderr)
+                )
                 self._safe_cleanup(decrypted_path)
                 return False
             except subprocess.TimeoutExpired:
@@ -75,13 +96,18 @@ def execute(self) -> bool:
                 return False
 
     def _verify_integrity(self, decrypted_path: str) -> None:
-        """Verify decrypted file matches original backup checksum"""
-        original_path = os.path.splitext(self.file_path)[0]
+        """
+        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):
-            decrypted_hash = self._calculate_sha256(decrypted_path)
-            original_hash = self._calculate_sha256(original_path)
+            decrypted_hash: str = self._calculate_sha256(decrypted_path)
+            original_hash: str = self._calculate_sha256(original_path)
 
-            # Compare hashes
+            # Print hashes for manual inspection (acceptable for CLI tools)
             print(f"Decrypted file hash: {decrypted_hash}")
             print(f"Original file hash: {original_hash}")
 
@@ -91,7 +117,15 @@ 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:
             while True:
@@ -102,8 +136,16 @@ def _calculate_sha256(self, file_path: str) -> str:
         return sha256.hexdigest()
 
     def _sanitize_logs(self, output: bytes) -> str:
-        """Safe log sanitization without modifying bytes"""
-        # Replace password=<value> with password=[REDACTED]
+        """
+        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)
         return sanitized.decode("utf-8", errors="replace")