276 lines
13 KiB
Python
276 lines
13 KiB
Python
# backup_handler.py
|
|
import os
|
|
import datetime
|
|
import zipfile
|
|
import logging
|
|
|
|
# Removed 'stat' import as it's not strictly needed for the core logic here
|
|
|
|
|
|
class BackupHandler:
|
|
"""Handles the creation of ZIP backups with file and directory exclusions."""
|
|
|
|
def __init__(self, logger):
|
|
"""
|
|
Initializes the BackupHandler.
|
|
|
|
Args:
|
|
logger (logging.Logger or logging.LoggerAdapter): Logger instance.
|
|
"""
|
|
if not isinstance(logger, (logging.Logger, logging.LoggerAdapter)):
|
|
raise TypeError("BackupHandler requires a valid Logger or LoggerAdapter.")
|
|
self.logger = logger
|
|
|
|
def _log_walk_error(self, os_error):
|
|
"""Error handler callback for os.walk to log issues during traversal."""
|
|
# Log PermissionError and other OSError subclasses encountered during walk
|
|
if isinstance(os_error, OSError):
|
|
self.logger.warning(
|
|
f"OS error during directory walk: Cannot access '{os_error.filename}'. "
|
|
f"Reason: {os_error.strerror}. Skipping this item/directory."
|
|
)
|
|
else:
|
|
# Log any other unexpected errors during walk
|
|
self.logger.error(
|
|
f"Unexpected error during directory walk: {os_error}", exc_info=True
|
|
)
|
|
# Returning None (implicitly) tells os.walk to continue if possible
|
|
|
|
def create_zip_backup(
|
|
self,
|
|
source_repo_path, # Directory to back up
|
|
backup_base_dir, # Destination base directory for the zip
|
|
profile_name, # Profile name for backup filename
|
|
excluded_extensions, # Set of lowercase extensions like {'.log', '.tmp'}
|
|
excluded_dirs_base, # Set of lowercase directory base names like {'__pycache__', '.git'}
|
|
):
|
|
"""
|
|
Creates a timestamped ZIP backup of the source directory, respecting
|
|
provided file extension and directory name exclusions.
|
|
|
|
Args:
|
|
source_repo_path (str): Absolute path to the directory to back up.
|
|
backup_base_dir (str): Absolute path to the base backup directory.
|
|
profile_name (str): Name of the current profile (for filename).
|
|
excluded_extensions (set): Set of lowercase file extensions to exclude.
|
|
excluded_dirs_base (set): Set of lowercase directory base names to exclude.
|
|
|
|
Returns:
|
|
str or None: The full path of the created ZIP file on success.
|
|
None if the backup is empty and removed, or on critical error before zip creation.
|
|
|
|
Raises:
|
|
ValueError: If input paths are invalid or required args missing/wrong type.
|
|
IOError: If directory creation or file writing fails critically. (Includes PermissionError)
|
|
Exception: For other unexpected errors during ZIP creation.
|
|
"""
|
|
self.logger.info(f"Starting ZIP backup for profile '{profile_name}'...")
|
|
self.logger.debug(f"Source path: '{source_repo_path}'")
|
|
self.logger.debug(f"Backup base directory: '{backup_base_dir}'")
|
|
self.logger.debug(f"Excluded extensions: {excluded_extensions}")
|
|
# --- MODIFICA: Log delle directory escluse ---
|
|
self.logger.debug(f"Excluded directory names (base): {excluded_dirs_base}")
|
|
# --- FINE MODIFICA ---
|
|
|
|
# --- 1. Validate Inputs ---
|
|
if not source_repo_path or not os.path.isdir(source_repo_path):
|
|
raise ValueError(f"Invalid source directory path: {source_repo_path}")
|
|
if not backup_base_dir:
|
|
raise ValueError("Backup base directory cannot be empty.")
|
|
if not isinstance(excluded_extensions, set):
|
|
raise TypeError("excluded_extensions must be a set.")
|
|
if not isinstance(excluded_dirs_base, set):
|
|
# Ensure the type passed from the controller is correct
|
|
raise TypeError("excluded_dirs_base must be a set.")
|
|
|
|
# --- 2. Prepare Destination Directory ---
|
|
try:
|
|
if not os.path.isdir(backup_base_dir):
|
|
self.logger.info(f"Creating backup base directory: {backup_base_dir}")
|
|
os.makedirs(backup_base_dir, exist_ok=True)
|
|
# Check write permissions (important!)
|
|
if not os.access(backup_base_dir, os.W_OK | os.X_OK):
|
|
raise PermissionError(
|
|
f"Write/traverse permission denied for backup directory: {backup_base_dir}"
|
|
)
|
|
except OSError as e:
|
|
self.logger.error(
|
|
f"Cannot create or access backup directory '{backup_base_dir}': {e}",
|
|
exc_info=True,
|
|
)
|
|
raise IOError(
|
|
f"Could not prepare backup directory: {e}"
|
|
) from e # Re-raise as IOError
|
|
|
|
# --- 3. Construct Backup Filename ---
|
|
try:
|
|
now_str = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
safe_profile = (
|
|
"".join(
|
|
c for c in profile_name if c.isalnum() or c in ("_", "-")
|
|
).rstrip()
|
|
or "profile"
|
|
)
|
|
backup_filename = f"{now_str}_backup_{safe_profile}.zip"
|
|
backup_full_path = os.path.join(backup_base_dir, backup_filename)
|
|
self.logger.info(f"Target backup ZIP file: {backup_full_path}")
|
|
except Exception as e:
|
|
raise ValueError(f"Could not construct valid backup filename: {e}") from e
|
|
|
|
# --- 4. Create ZIP Archive ---
|
|
files_added = 0
|
|
files_excluded = 0
|
|
dirs_excluded_count = 0 # Use a distinct name for the counter
|
|
zip_f = None # Initialize zip file object reference
|
|
|
|
try:
|
|
# Open ZIP file with compression and large file support
|
|
zip_f = zipfile.ZipFile(
|
|
backup_full_path, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True
|
|
)
|
|
|
|
# Walk through the source directory tree.
|
|
# topdown=True allows modification of 'dirs' list to prune traversal.
|
|
# onerror=_log_walk_error handles issues like permission errors during walk gracefully.
|
|
for root, dirs, files in os.walk(
|
|
source_repo_path, topdown=True, onerror=self._log_walk_error
|
|
):
|
|
|
|
# --- MODIFICA: Directory Exclusion Logic ---
|
|
original_dirs = list(dirs) # Keep original list for logging/counting
|
|
# Modify 'dirs' in-place: keep only directories whose lowercase name
|
|
# is NOT in the excluded_dirs_base set.
|
|
dirs[:] = [d for d in dirs if d.lower() not in excluded_dirs_base]
|
|
|
|
# Log and count the excluded directories at this level
|
|
excluded_now = set(original_dirs) - set(dirs)
|
|
if excluded_now:
|
|
current_level_excluded_count = len(excluded_now)
|
|
dirs_excluded_count += (
|
|
current_level_excluded_count # Increment total count
|
|
)
|
|
for ex_dir in excluded_now:
|
|
# Log the full path of the excluded directory
|
|
self.logger.debug(
|
|
f"Excluding directory (and contents): {os.path.join(root, ex_dir)}"
|
|
)
|
|
# --- FINE MODIFICA ---
|
|
|
|
# --- File Exclusion and Addition ---
|
|
for filename in files:
|
|
# Get file extension (lowercase)
|
|
_, file_ext = os.path.splitext(filename)
|
|
file_ext_lower = file_ext.lower()
|
|
|
|
# Check exclusion rules (extension)
|
|
if file_ext_lower in excluded_extensions:
|
|
path_excluded = os.path.join(root, filename)
|
|
self.logger.debug(
|
|
f"Excluding file by extension: {path_excluded}"
|
|
)
|
|
files_excluded += 1
|
|
continue # Skip this file
|
|
|
|
# If not excluded, add file to ZIP
|
|
file_full_path = os.path.join(root, filename)
|
|
# Calculate relative path for storage inside ZIP
|
|
try:
|
|
archive_name = os.path.relpath(file_full_path, source_repo_path)
|
|
except ValueError: # Handle cases like different drives on Windows
|
|
archive_name = os.path.join(
|
|
os.path.basename(root), filename
|
|
) # Fallback
|
|
self.logger.warning(
|
|
f"Could not get relative path for {file_full_path}. Using fallback arcname: {archive_name}"
|
|
)
|
|
|
|
try:
|
|
# Write the file to the ZIP archive
|
|
zip_f.write(file_full_path, arcname=archive_name)
|
|
files_added += 1
|
|
# Log progress periodically
|
|
if files_added % 500 == 0 and files_added > 0:
|
|
self.logger.debug(f"Added {files_added} files...")
|
|
except FileNotFoundError:
|
|
self.logger.warning(
|
|
f"File disappeared before writing: {file_full_path}"
|
|
)
|
|
except PermissionError:
|
|
self.logger.warning(
|
|
f"Permission denied reading file: {file_full_path}"
|
|
)
|
|
except Exception as write_e:
|
|
# Log error writing a specific file but allow backup to continue
|
|
self.logger.error(
|
|
f"Error writing file '{file_full_path}' to ZIP: {write_e}",
|
|
exc_info=False,
|
|
) # Keep log clean
|
|
|
|
# Log final summary after successful walk and write attempts
|
|
self.logger.info(f"Backup ZIP creation process finished.")
|
|
self.logger.info(
|
|
f"Summary - Files Added: {files_added}, Excluded Files (ext): {files_excluded}, Excluded Dirs (name): {dirs_excluded_count}"
|
|
)
|
|
|
|
except (OSError, zipfile.BadZipFile, zipfile.LargeZipFile) as e:
|
|
# Handle OS errors (permissions, disk space) and critical ZIP format errors
|
|
self.logger.error(
|
|
f"Error creating/writing backup ZIP '{backup_full_path}': {e}",
|
|
exc_info=True,
|
|
)
|
|
# Re-raise as IOError for the caller
|
|
raise IOError(f"Failed to create/write backup ZIP: {e}") from e
|
|
except Exception as e:
|
|
# Catch any other unexpected error during the process
|
|
self.logger.exception(f"Unexpected error during ZIP backup: {e}")
|
|
raise # Re-raise the original exception
|
|
finally:
|
|
# Ensure the ZIP file is always closed
|
|
if zip_f:
|
|
try:
|
|
zip_f.close()
|
|
self.logger.debug(f"ZIP file '{backup_full_path}' closed.")
|
|
except Exception as close_e:
|
|
self.logger.error(
|
|
f"Error closing ZIP file '{backup_full_path}': {close_e}",
|
|
exc_info=True,
|
|
)
|
|
|
|
# --- Cleanup potentially empty ZIP file ---
|
|
final_path_to_return = None
|
|
try:
|
|
if os.path.exists(backup_full_path):
|
|
if files_added > 0:
|
|
# Success: file exists and has content
|
|
final_path_to_return = backup_full_path
|
|
else:
|
|
# File exists but is empty, remove it
|
|
self.logger.warning(
|
|
f"Backup ZIP is empty (0 files added): {backup_full_path}. Removing."
|
|
)
|
|
try:
|
|
os.remove(backup_full_path)
|
|
self.logger.info("Removed empty backup ZIP file.")
|
|
except OSError as rm_e:
|
|
self.logger.error(f"Failed remove empty backup ZIP: {rm_e}")
|
|
final_path_to_return = None # No valid backup created
|
|
else:
|
|
# File doesn't exist at the end
|
|
if files_added > 0:
|
|
self.logger.error(
|
|
f"Backup process reported adding files, but ZIP file missing: {backup_full_path}"
|
|
)
|
|
else:
|
|
self.logger.info(
|
|
"Backup process completed without creating a file (source empty or all excluded)."
|
|
)
|
|
final_path_to_return = None
|
|
except Exception as final_check_e:
|
|
self.logger.error(
|
|
f"Error during final check/cleanup of backup file: {final_check_e}",
|
|
exc_info=True,
|
|
)
|
|
final_path_to_return = None # Assume failure if check fails
|
|
|
|
return final_path_to_return # Return path or None
|