""" File Manager for AegisLM Reporting System Production-grade file storage and management with cleanup and monitoring. """ import os import shutil import time from datetime import datetime, timedelta from typing import Dict, Any, List, Optional from pathlib import Path import sys # Add parent directory to path for imports current_dir = Path(__file__).parent backend_dir = current_dir.parent if str(backend_dir) not in sys.path: sys.path.insert(0, str(backend_dir)) class ReportFileManager: """ Production-grade file manager for report storage and lifecycle management. Handles file storage, cleanup, monitoring, and access control. """ def __init__(self, reports_dir: Optional[str] = None, max_storage_mb: float = 1000.0): """ Initialize file manager. Args: reports_dir: Directory for storing reports max_storage_mb: Maximum storage limit in MB """ if reports_dir is None: # Default to backend/reports relative to this file current_file = Path(__file__) self.reports_dir = current_file.parent.parent / "reports" else: self.reports_dir = Path(reports_dir) self.max_storage_mb = max_storage_mb # Ensure reports directory exists self.reports_dir.mkdir(parents=True, exist_ok=True) # Create subdirectories (self.reports_dir / "json").mkdir(exist_ok=True) (self.reports_dir / "csv").mkdir(exist_ok=True) (self.reports_dir / "archive").mkdir(exist_ok=True) (self.reports_dir / "temp").mkdir(exist_ok=True) def get_unique_filename(self, run_id: str, report_type: str, format: str) -> str: """ Generate unique filename for report. Args: run_id: Run ID report_type: Type of report format: File format Returns: Unique filename """ timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") base_filename = f"run_{run_id}_{report_type}_{timestamp}.{format}" # Check for conflicts counter = 1 filename = base_filename while (self.reports_dir / filename).exists(): stem = Path(base_filename).stem suffix = Path(base_filename).suffix filename = f"{stem}_{counter}{suffix}" counter += 1 return filename def get_file_path(self, filename: str, format: str = "json") -> Path: """ Get full file path for a report. Args: filename: Filename format: File format Returns: Full file path """ if format.lower() == "json": return self.reports_dir / "json" / filename elif format.lower() == "csv": return self.reports_dir / "csv" / filename else: return self.reports_dir / filename def store_file(self, content: str, filename: str, format: str = "json") -> Path: """ Store file content with proper management. Args: content: File content filename: Filename format: File format Returns: File path Raises: ValueError: If storage fails """ try: file_path = self.get_file_path(filename, format) # Write file with open(file_path, 'w', encoding='utf-8') as f: f.write(content) return file_path except Exception as e: raise ValueError(f"Failed to store file: {str(e)}") def get_file_info(self, file_path: str) -> Dict[str, Any]: """ Get detailed file information. Args: file_path: File path Returns: File information """ try: file_path = Path(file_path) if not file_path.exists(): return {"error": "File not found"} stat = file_path.stat() return { "file_path": str(file_path), "file_name": file_path.name, "file_size_bytes": stat.st_size, "file_size_mb": round(stat.st_size / (1024 * 1024), 2), "created_at": datetime.fromtimestamp(stat.st_ctime).isoformat(), "modified_at": datetime.fromtimestamp(stat.st_mtime).isoformat(), "accessed_at": datetime.fromtimestamp(stat.st_atime).isoformat(), "file_extension": file_path.suffix.lower(), "is_readable": os.access(file_path, os.R_OK), "is_writable": os.access(file_path, os.W_OK) } except Exception as e: return {"error": f"Failed to get file info: {str(e)}"} def list_files(self, format: Optional[str] = None, run_id: Optional[str] = None) -> List[Dict[str, Any]]: """ List files with optional filtering. Args: format: Filter by format run_id: Filter by run ID Returns: List of file information """ files = [] try: # Determine search directories search_dirs = [self.reports_dir] if format: if format.lower() == "json": search_dirs = [self.reports_dir / "json"] elif format.lower() == "csv": search_dirs = [self.reports_dir / "csv"] for search_dir in search_dirs: if not search_dir.exists(): continue for file_path in search_dir.iterdir(): if not file_path.is_file(): continue # Apply filters if format and file_path.suffix.lower() != f".{format.lower()}": continue if run_id and run_id not in file_path.name: continue file_info = self.get_file_info(file_path) if "error" not in file_info: files.append(file_info) # Sort by creation date (newest first) files.sort(key=lambda x: x['created_at'], reverse=True) return files except Exception as e: return [{"error": f"Failed to list files: {str(e)}"}] def delete_file(self, file_path: str) -> bool: """ Delete a file safely. Args: file_path: File path Returns: True if deleted, False otherwise """ try: file_path = Path(file_path) if file_path.exists(): file_path.unlink() return True return False except Exception: return False def archive_file(self, file_path: str) -> bool: """ Archive a file to the archive directory. Args: file_path: File path Returns: True if archived, False otherwise """ try: source_path = Path(file_path) if not source_path.exists(): return False # Create archive filename with timestamp timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") archive_filename = f"{source_path.stem}_{timestamp}{source_path.suffix}" archive_path = self.reports_dir / "archive" / archive_filename # Move file to archive shutil.move(str(source_path), str(archive_path)) return True except Exception: return False def cleanup_old_files(self, days_old: int = 30, dry_run: bool = False) -> Dict[str, Any]: """ Clean up old files. Args: days_old: Age threshold in days dry_run: If True, only report what would be deleted Returns: Cleanup results """ cutoff_date = datetime.utcnow() - timedelta(days=days_old) files_to_delete = [] files_to_archive = [] total_size_freed = 0 try: # Find old files for file_info in self.list_files(): created_at = datetime.fromisoformat(file_info['created_at']) if created_at < cutoff_date: if days_old > 60: # Very old files, archive files_to_archive.append(file_info) else: # Moderately old files, delete files_to_delete.append(file_info) total_size_freed += file_info['file_size_bytes'] results = { "cutoff_date": cutoff_date.isoformat(), "files_to_delete": len(files_to_delete), "files_to_archive": len(files_to_archive), "total_size_freed_mb": round(total_size_freed / (1024 * 1024), 2), "dry_run": dry_run } if not dry_run: # Actually perform cleanup deleted_count = 0 archived_count = 0 for file_info in files_to_delete: if self.delete_file(file_info['file_path']): deleted_count += 1 for file_info in files_to_archive: if self.archive_file(file_info['file_path']): archived_count += 1 results["deleted_count"] = deleted_count results["archived_count"] = archived_count else: results["files_to_delete_details"] = files_to_delete[:10] # First 10 for preview results["files_to_archive_details"] = files_to_archive[:10] # First 10 for preview return results except Exception as e: return {"error": f"Cleanup failed: {str(e)}"} def get_storage_stats(self) -> Dict[str, Any]: """ Get storage statistics. Returns: Storage statistics """ try: total_size = 0 file_count = 0 format_counts = {} for file_info in self.list_files(): file_count += 1 total_size += file_info['file_size_bytes'] format_name = file_info['file_extension'].replace('.', '') format_counts[format_name] = format_counts.get(format_name, 0) + 1 # Calculate directory sizes json_dir_size = self._get_directory_size(self.reports_dir / "json") csv_dir_size = self._get_directory_size(self.reports_dir / "csv") archive_dir_size = self._get_directory_size(self.reports_dir / "archive") temp_dir_size = self._get_directory_size(self.reports_dir / "temp") return { "total_files": file_count, "total_size_bytes": total_size, "total_size_mb": round(total_size / (1024 * 1024), 2), "max_storage_mb": self.max_storage_mb, "storage_usage_percent": round((total_size / (1024 * 1024)) / self.max_storage_mb * 100, 2), "format_distribution": format_counts, "directory_sizes_mb": { "json": round(json_dir_size / (1024 * 1024), 2), "csv": round(csv_dir_size / (1024 * 1024), 2), "archive": round(archive_dir_size / (1024 * 1024), 2), "temp": round(temp_dir_size / (1024 * 1024), 2) }, "reports_directory": str(self.reports_dir), "stats_timestamp": datetime.utcnow().isoformat() } except Exception as e: return {"error": f"Failed to get storage stats: {str(e)}"} def check_storage_health(self) -> Dict[str, Any]: """ Check storage health and recommendations. Returns: Storage health assessment """ try: stats = self.get_storage_stats() if "error" in stats: return {"status": "error", "message": stats["error"]} health_status = "healthy" issues = [] recommendations = [] # Check storage usage usage_percent = stats["storage_usage_percent"] if usage_percent > 90: health_status = "critical" issues.append("Storage usage is critically high") recommendations.append("Immediately clean up old files") recommendations.append("Consider increasing storage limit") elif usage_percent > 75: health_status = "warning" issues.append("Storage usage is high") recommendations.append("Schedule cleanup of old files") elif usage_percent > 50: recommendations.append("Monitor storage usage") # Check file count if stats["total_files"] > 10000: if health_status == "healthy": health_status = "warning" issues.append("Large number of files may impact performance") recommendations.append("Consider archiving old files") # Check directory distribution archive_size = stats["directory_sizes_mb"]["archive"] if archive_size > stats["total_size_mb"] * 0.5: recommendations.append("Archive directory is large, consider cleanup") # Check permissions if not os.access(self.reports_dir, os.W_OK): health_status = "critical" issues.append("Reports directory is not writable") recommendations.append("Check directory permissions") return { "status": health_status, "issues": issues, "recommendations": recommendations, "storage_stats": stats, "health_timestamp": datetime.utcnow().isoformat() } except Exception as e: return { "status": "error", "message": f"Health check failed: {str(e)}", "health_timestamp": datetime.utcnow().isoformat() } def optimize_storage(self) -> Dict[str, Any]: """ Optimize storage by cleaning up and organizing files. Returns: Optimization results """ try: results = { "started_at": datetime.utcnow().isoformat(), "actions_performed": [] } # Clean up temp files temp_files = list((self.reports_dir / "temp").glob("*")) temp_files_deleted = 0 temp_size_freed = 0 for temp_file in temp_files: try: size = temp_file.stat().st_size temp_file.unlink() temp_files_deleted += 1 temp_size_freed += size except Exception: pass if temp_files_deleted > 0: results["actions_performed"].append({ "action": "cleaned_temp_files", "files_deleted": temp_files_deleted, "size_freed_mb": round(temp_size_freed / (1024 * 1024), 2) }) # Archive very old files (over 60 days) cleanup_result = self.cleanup_old_files(days_old=60, dry_run=False) if "archived_count" in cleanup_result and cleanup_result["archived_count"] > 0: results["actions_performed"].append({ "action": "archived_old_files", "files_archived": cleanup_result["archived_count"], "days_threshold": 60 }) # Delete moderately old files (over 30 days) cleanup_result = self.cleanup_old_files(days_old=30, dry_run=False) if "deleted_count" in cleanup_result and cleanup_result["deleted_count"] > 0: results["actions_performed"].append({ "action": "deleted_old_files", "files_deleted": cleanup_result["deleted_count"], "size_freed_mb": cleanup_result["total_size_freed_mb"], "days_threshold": 30 }) results["completed_at"] = datetime.utcnow().isoformat() results["total_actions"] = len(results["actions_performed"]) return results except Exception as e: return { "error": f"Storage optimization failed: {str(e)}", "completed_at": datetime.utcnow().isoformat() } def _get_directory_size(self, directory: Path) -> int: """Calculate total size of directory.""" total_size = 0 try: for file_path in directory.rglob("*"): if file_path.is_file(): total_size += file_path.stat().st_size except Exception: pass return total_size def create_backup(self, backup_dir: Optional[str] = None) -> Dict[str, Any]: """ Create backup of reports directory. Args: backup_dir: Backup directory path Returns: Backup results """ try: if backup_dir is None: timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") backup_dir = self.reports_dir.parent / f"reports_backup_{timestamp}" backup_path = Path(backup_dir) if backup_path.exists(): return {"error": "Backup directory already exists"} # Create backup shutil.copytree(self.reports_dir, backup_path) # Calculate backup size backup_size = self._get_directory_size(backup_path) return { "backup_path": str(backup_path), "backup_size_mb": round(backup_size / (1024 * 1024), 2), "created_at": datetime.utcnow().isoformat(), "source_directory": str(self.reports_dir) } except Exception as e: return {"error": f"Backup failed: {str(e)}"} def restore_backup(self, backup_dir: str, restore_to_temp: bool = True) -> Dict[str, Any]: """ Restore reports from backup. Args: backup_dir: Backup directory path restore_to_temp: If True, restore to temp directory first Returns: Restore results """ try: backup_path = Path(backup_dir) if not backup_path.exists(): return {"error": "Backup directory not found"} if restore_to_temp: # Restore to temp directory first temp_restore_path = self.reports_dir / "temp" / "restore" if temp_restore_path.exists(): shutil.rmtree(temp_restore_path) shutil.copytree(backup_path, temp_restore_path) return { "restored_to": str(temp_restore_path), "restore_type": "temp", "restored_at": datetime.utcnow().isoformat(), "message": "Backup restored to temp directory. Review before moving to main directory." } else: # Direct restore (backup current first) current_backup = self.create_backup() if "error" in current_backup: return {"error": f"Failed to backup current directory: {current_backup['error']}"} # Remove current directory and restore shutil.rmtree(self.reports_dir) shutil.copytree(backup_path, self.reports_dir) return { "restored_to": str(self.reports_dir), "restore_type": "direct", "current_backup": current_backup["backup_path"], "restored_at": datetime.utcnow().isoformat() } except Exception as e: return {"error": f"Restore failed: {str(e)}"} # Global instance for easy access _file_manager = None def get_file_manager() -> ReportFileManager: """ Get the global file manager instance. Returns: Global file manager instance """ global _file_manager if _file_manager is None: _file_manager = ReportFileManager() return _file_manager # Convenience functions for direct usage def get_unique_filename(run_id: str, report_type: str, format: str) -> str: """ Get unique filename for report. Args: run_id: Run ID report_type: Type of report format: File format Returns: Unique filename """ manager = get_file_manager() return manager.get_unique_filename(run_id, report_type, format) def store_report_file(content: str, filename: str, format: str = "json") -> str: """ Store report file. Args: content: File content filename: Filename format: File format Returns: File path """ manager = get_file_manager() return str(manager.store_file(content, filename, format))