""" Genesis System - Backup Manager This module provides functionality for automated backups of critical system data. It includes features for scheduling backups, managing backup storage, and ensuring data integrity. """ import os import shutil import time import datetime import logging import configparser from typing import List, Optional # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) class BackupManager: """ Manages automated backups of system data. """ def __init__(self, config_file: str = "/mnt/e/genesis-system/config/backup.ini") -> None: """ Initializes the BackupManager with configuration from a file. Args: config_file (str): Path to the configuration file. """ self.config_file = config_file self.config = configparser.ConfigParser() self.load_config() self.backup_source = self.config.get("backup", "source_directory") self.backup_destination = self.config.get("backup", "destination_directory") self.backup_retention = self.config.getint("backup", "retention_period_days") # Ensure destination directory exists os.makedirs(self.backup_destination, exist_ok=True) def load_config(self) -> None: """ Loads configuration from the specified file. """ try: self.config.read(self.config_file) logging.info(f"Configuration loaded from {self.config_file}") except Exception as e: logging.error(f"Error loading configuration: {e}") raise def create_backup(self) -> bool: """ Creates a full backup of the source directory. Returns: bool: True if the backup was successful, False otherwise. """ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_name = f"backup_{timestamp}" backup_path = os.path.join(self.backup_destination, backup_name) try: shutil.copytree(self.backup_source, backup_path) logging.info(f"Backup created successfully at: {backup_path}") return True except Exception as e: logging.error(f"Backup failed: {e}") return False def verify_backup(self, backup_path: str) -> bool: """ Verifies the integrity of a backup by checking for file existence and size. Args: backup_path (str): Path to the backup directory. Returns: bool: True if the backup is valid, False otherwise. """ try: # Basic check: ensure the backup directory exists if not os.path.exists(backup_path): logging.warning(f"Backup path does not exist: {backup_path}") return False # More thorough check: compare the number of files and their sizes between source and backup source_file_count = self._count_files(self.backup_source) backup_file_count = self._count_files(backup_path) if source_file_count != backup_file_count: logging.warning(f"File count mismatch: Source={source_file_count}, Backup={backup_file_count}") return False # Further (simplified) size comparison source_size = self._calculate_total_size(self.backup_source) backup_size = self._calculate_total_size(backup_path) size_difference_percent = abs(source_size - backup_size) / max(source_size, backup_size, 1) * 100 if size_difference_percent > 5: # Allow for small differences logging.warning(f"Significant size difference: Source={source_size}, Backup={backup_size}") return False logging.info(f"Backup verified successfully at: {backup_path}") return True except Exception as e: logging.error(f"Backup verification failed: {e}") return False def _count_files(self, directory: str) -> int: """ Counts the number of files in a directory recursively. Args: directory (str): The directory to count files in. Returns: int: The number of files. """ count = 0 for root, _, files in os.walk(directory): count += len(files) return count def _calculate_total_size(self, directory: str) -> int: """ Calculates the total size (in bytes) of all files within a directory recursively. Args: directory (str): The directory to calculate the size of. Returns: int: The total size in bytes. """ total_size = 0 for root, _, files in os.walk(directory): for file in files: file_path = os.path.join(root, file) try: total_size += os.path.getsize(file_path) except OSError as e: logging.warning(f"Could not get size of {file_path}: {e}") return total_size def manage_retention(self) -> None: """ Deletes old backups based on the retention period specified in the configuration. """ cutoff_date = datetime.datetime.now() - datetime.timedelta(days=self.backup_retention) logging.info(f"Deleting backups older than: {cutoff_date}") for backup_name in os.listdir(self.backup_destination): backup_path = os.path.join(self.backup_destination, backup_name) try: # Attempt to parse the timestamp from the backup name backup_timestamp_str = backup_name.split("_")[1] # Assuming format backup_YYYYMMDD_HHMMSS backup_timestamp = datetime.datetime.strptime(backup_timestamp_str, "%Y%m%d_%H%M%S") if backup_timestamp < cutoff_date: shutil.rmtree(backup_path) logging.info(f"Deleted old backup: {backup_path}") except (ValueError, IndexError) as e: logging.warning(f"Could not parse timestamp from backup name '{backup_name}'. Skipping. Error: {e}") except OSError as e: logging.error(f"Error deleting backup {backup_path}: {e}") def run_backup_cycle(self) -> None: """ Executes a complete backup cycle, including creating a backup, verifying it, and managing retention. """ logging.info("Starting backup cycle...") if self.create_backup(): timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_name = f"backup_{timestamp}" backup_path = os.path.join(self.backup_destination, backup_name) if self.verify_backup(backup_path): self.manage_retention() else: logging.error("Backup verification failed. Retention management skipped.") else: logging.error("Backup creation failed. Cycle aborted.") logging.info("Backup cycle complete.") def main(): """ Main function to run the backup manager. """ try: backup_manager = BackupManager() backup_manager.run_backup_cycle() except Exception as e: logging.error(f"An unhandled error occurred: {e}") if __name__ == "__main__": main()