file_manager.py

"""
Audio Volume Normalizer - File Management Module

This module handles file operations including:
- Scanning directories for audio files (MP3 and WAV)
- Creating backups of files before processing
- Managing file operations with error handling
"""

import logging
import os
import shutil
import tempfile
from typing import List, Dict, Tuple, Optional, Callable

logger = logging.getLogger(__name__)


class FileManager:
    """Manages file operations for the Audio Volume Normalizer."""
    
    def __init__(self):
        """Initialize the FileManager."""
        self.temp_dir: Optional[str] = None
        self.backup_files: Dict[str, str] = {}  # Maps original path to backup path
    
    def scan_directory(self, directory: str) -> List[str]:
        """
        Scan a directory for audio files (MP3 and WAV).
        
        Args:
            directory: Path to the directory to scan
            
        Returns:
            List of full paths to audio files
            
        Raises:
            FileNotFoundError: If the directory doesn't exist
            PermissionError: If the directory can't be accessed
        """
        logger.info(f"Scanning directory: {directory}")
        
        if not os.path.isdir(directory):
            logger.error(f"Directory not found: {directory}")
            raise FileNotFoundError(f"Directory not found: {directory}")
        
        audio_files = []
        
        try:
            for root, _, files in os.walk(directory):
                for file in files:
                    # Skip macOS metadata files (files starting with "._")
                    if not file.startswith('._') and (file.lower().endswith('.mp3') or file.lower().endswith('.wav')):
                        full_path = os.path.join(root, file)
                        audio_files.append(full_path)
                        logger.debug(f"Found audio file: {full_path}")
        except PermissionError as e:
            logger.error(f"Permission error scanning directory: {e}")
            raise
        
        mp3_count = sum(1 for f in audio_files if f.lower().endswith('.mp3'))
        wav_count = sum(1 for f in audio_files if f.lower().endswith('.wav'))
        logger.info(f"Found {len(audio_files)} audio files ({mp3_count} MP3, {wav_count} WAV)")
        return audio_files
    
    def create_backup(self, file_paths: List[str]) -> bool:
        """
        Create backups of the specified files in a temporary directory.
        
        Args:
            file_paths: List of file paths to back up
            
        Returns:
            True if backup was successful, False otherwise
            
        Raises:
            IOError: If backup creation fails
        """
        logger.info(f"Creating backups for {len(file_paths)} files")
        
        # Create a temporary directory for backups
        try:
            self.temp_dir = tempfile.mkdtemp(prefix="audio_normalizer_backup_")
            logger.debug(f"Created temporary backup directory: {self.temp_dir}")
        except IOError as e:
            logger.error(f"Failed to create temporary directory: {e}")
            return False
        
        # Copy files to the backup directory
        for file_path in file_paths:
            try:
                file_name = os.path.basename(file_path)
                backup_path = os.path.join(self.temp_dir, file_name)
                
                # Handle duplicate filenames by adding a suffix
                if os.path.exists(backup_path):
                    base, ext = os.path.splitext(file_name)
                    count = 1
                    while os.path.exists(backup_path):
                        backup_path = os.path.join(
                            self.temp_dir, f"{base}_{count}{ext}"
                        )
                        count += 1
                
                shutil.copy2(file_path, backup_path)
                self.backup_files[file_path] = backup_path
                logger.debug(f"Backed up: {file_path} -> {backup_path}")
            except (IOError, OSError) as e:
                logger.error(f"Failed to back up {file_path}: {e}")
                self.cleanup_backups()
                raise IOError(f"Backup failed for {file_path}: {e}")
        
        logger.info(f"Successfully backed up {len(file_paths)} files")
        return True
    
    def restore_from_backup(self, file_path: str) -> bool:
        """
        Restore a single file from its backup.
        
        Args:
            file_path: Path to the original file to restore
            
        Returns:
            True if restore was successful, False otherwise
        """
        if not self.temp_dir or file_path not in self.backup_files:
            logger.warning(f"No backup found for {file_path}")
            return False
        
        backup_path = self.backup_files[file_path]
        
        try:
            shutil.copy2(backup_path, file_path)
            logger.info(f"Restored {file_path} from backup")
            return True
        except (IOError, OSError) as e:
            logger.error(f"Failed to restore {file_path}: {e}")
            return False
    
    def restore_all_backups(self) -> Tuple[int, int]:
        """
        Restore all files from their backups.
        
        Returns:
            Tuple of (number of successful restores, number of failed restores)
        """
        if not self.temp_dir:
            logger.warning("No backups to restore")
            return (0, 0)
        
        success_count = 0
        fail_count = 0
        
        for original_path in self.backup_files:
            if self.restore_from_backup(original_path):
                success_count += 1
            else:
                fail_count += 1
        
        logger.info(f"Restored {success_count} files, {fail_count} failed")
        return (success_count, fail_count)
    
    def cleanup_backups(self) -> None:
        """Remove the temporary backup directory and all its contents."""
        if self.temp_dir and os.path.exists(self.temp_dir):
            try:
                shutil.rmtree(self.temp_dir)
                logger.info(f"Removed backup directory: {self.temp_dir}")
                self.temp_dir = None
                self.backup_files = {}
            except (IOError, OSError) as e:
                logger.error(f"Failed to remove backup directory: {e}")
    
    def save_normalized_file(self, original_path: str, audio_data: any, reference_encoding: Optional[Dict[str, any]] = None) -> bool:
        """
        Save normalized audio data to the original file path.
        
        Args:
            original_path: Path to the original file
            audio_data: Normalized audio data (pydub AudioSegment)
            reference_encoding: Optional encoding parameters to apply
            
        Returns:
            True if save was successful, False otherwise
        """
        try:
            # Determine the format based on file extension
            if original_path.lower().endswith('.mp3'):
                format_type = "mp3"
            elif original_path.lower().endswith('.wav'):
                format_type = "wav"
            else:
                logger.error(f"Unsupported file format for {original_path}")
                return False
            
            # Export options
            export_kwargs = {"format": format_type}
            
            # Apply reference encoding if provided
            if reference_encoding:
                # Set sample rate
                if "sample_rate" in reference_encoding:
                    audio_data = audio_data.set_frame_rate(reference_encoding["sample_rate"])
                    logger.debug(f"Setting sample rate to {reference_encoding['sample_rate']} Hz")
                
                # Set channels
                if "channels" in reference_encoding:
                    if reference_encoding["channels"] == 1 and audio_data.channels == 2:
                        audio_data = audio_data.set_channels(1)
                        logger.debug("Converting stereo to mono")
                    elif reference_encoding["channels"] == 2 and audio_data.channels == 1:
                        audio_data = audio_data.set_channels(2)
                        logger.debug("Converting mono to stereo")
                
                # For MP3 files, set bitrate based on reference
                if format_type == "mp3":
                    # Use a default high quality bitrate
                    export_kwargs["bitrate"] = "320k"
            
            # Export the audio to the original file path
            audio_data.export(original_path, **export_kwargs)
            logger.info(f"Saved normalized file: {original_path} with reference encoding applied")
            return True
        except Exception as e:
            logger.error(f"Failed to save normalized file {original_path}: {e}")
            return False
    
    def get_file_info(self, file_path: str) -> Dict[str, any]:
        """
        Get information about a file.
        
        Args:
            file_path: Path to the file
            
        Returns:
            Dictionary with file information
        """
        try:
            stat = os.stat(file_path)
            return {
                'path': file_path,
                'name': os.path.basename(file_path),
                'size': stat.st_size,
                'modified': stat.st_mtime
            }
        except (IOError, OSError) as e:
            logger.error(f"Failed to get file info for {file_path}: {e}")
            return {
                'path': file_path,
                'name': os.path.basename(file_path),
                'size': 0,
                'modified': 0,
                'error': str(e)
            }