Create utils/file_handler.py

utils/file_handler.py

@@ -0,0 +1,76 @@
+"""
+File handling utilities
+"""
+import os
+import shutil
+import tempfile
+from typing import Optional, List
+from pathlib import Path
+import logging
+
+logger = logging.getLogger(__name__)
+
+class FileHandler:
+    def __init__(self, base_dir: str = "./cache"):
+        self.base_dir = Path(base_dir)
+        self.base_dir.mkdir(exist_ok=True)
+        self.temp_dir = self.base_dir / "temp"
+        self.temp_dir.mkdir(exist_ok=True)
+    
+    def save_uploaded_file(self, uploaded_file, filename: str) -> str:
+        """Save uploaded file to cache"""
+        try:
+            file_path = self.temp_dir / filename
+            shutil.copy(uploaded_file, file_path)
+            logger.info(f"Saved uploaded file: {file_path}")
+            return str(file_path)
+        except Exception as e:
+            logger.error(f"Failed to save uploaded file: {str(e)}")
+            raise
+    
+    def create_temp_file(self, extension: str = ".wav") -> str:
+        """Create temporary file path"""
+        temp_file = tempfile.NamedTemporaryFile(
+            suffix=extension,
+            dir=self.temp_dir,
+            delete=False
+        )
+        return temp_file.name
+    
+    def cleanup_temp_files(self, max_age_hours: int = 24):
+        """Clean up old temporary files"""
+        try:
+            current_time = time.time()
+            max_age_seconds = max_age_hours * 3600
+            
+            for file_path in self.temp_dir.iterdir():
+                if file_path.is_file():
+                    file_age = current_time - file_path.stat().st_mtime
+                    if file_age > max_age_seconds:
+                        file_path.unlink()
+                        logger.info(f"Deleted old temp file: {file_path}")
+        except Exception as e:
+            logger.error(f"Cleanup failed: {str(e)}")
+    
+    def get_file_size(self, file_path: str) -> int:
+        """Get file size in bytes"""
+        return os.path.getsize(file_path)
+    
+    def validate_audio_file(self, file_path: str) -> bool:
+        """Validate if file is a supported audio format"""
+        supported_formats = ['.wav', '.mp3', '.flac', '.ogg', '.m4a']
+        file_ext = Path(file_path).suffix.lower()
+        return file_ext in supported_formats
+    
+    def get_storage_info(self) -> dict:
+        """Get storage usage information"""
+        total_size = sum(
+            f.stat().st_size for f in self.base_dir.rglob('*') if f.is_file()
+        )
+        return {
+            "total_size_mb": total_size / (1024 * 1024),
+            "file_count": len(list(self.base_dir.rglob('*'))),
+            "temp_dir_size_mb": sum(
+                f.stat().st_size for f in self.temp_dir.iterdir() if f.is_file()
+            ) / (1024 * 1024)
+        }