""" File handling utilities for VFX Project Management System. Provides secure file upload, validation, and serving functionality. """ import os import shutil import mimetypes from datetime import datetime from typing import Optional, List, Tuple from pathlib import Path from PIL import Image from fastapi import HTTPException, UploadFile from sqlalchemy.orm import Session import hashlib class FileHandler: """Handles file operations for the VFX system.""" # Supported VFX media formats SUPPORTED_FORMATS = { # Video formats '.mov', '.mp4', '.avi', '.mkv', '.webm', # Image formats '.exr', '.jpg', '.jpeg', '.png', '.tiff', '.tif', '.dpx', '.hdr', # Document formats '.pdf', '.txt', '.doc', '.docx', # Archive formats '.zip', '.rar', '.7z' } # File size limits (in bytes) MAX_ATTACHMENT_SIZE = 10 * 1024 * 1024 # 10MB for attachments MAX_SUBMISSION_SIZE = 500 * 1024 * 1024 # 500MB for submissions (fallback) # Movie file extensions that should use global upload limit MOVIE_EXTENSIONS = {'.mov', '.mp4', '.avi', '.mkv', '.webm'} # Thumbnail settings THUMBNAIL_SIZE = (200, 200) THUMBNAIL_QUALITY = 85 def __init__(self, base_upload_dir: str = "uploads"): # Use absolute path relative to this file's location # This ensures it works whether run from workspace root or backend directory current_file = Path(__file__).resolve() self.backend_dir = current_file.parent.parent # Go up from utils/ to backend/ self.base_upload_dir = self.backend_dir / base_upload_dir self.base_upload_dir.mkdir(parents=True, exist_ok=True) # Create subdirectories self.attachments_dir = self.base_upload_dir / "attachments" self.submissions_dir = self.base_upload_dir / "submissions" self.thumbnails_dir = self.base_upload_dir / "thumbnails" self.project_thumbnails_dir = self.base_upload_dir / "project_thumbnails" for directory in [self.attachments_dir, self.submissions_dir, self.thumbnails_dir, self.project_thumbnails_dir]: directory.mkdir(parents=True, exist_ok=True) def validate_file(self, file: UploadFile, max_size: int, db: Optional[Session] = None) -> None: """Validate uploaded file format and size.""" if not file.filename: raise HTTPException(status_code=400, detail="No filename provided") # Check file extension file_extension = Path(file.filename).suffix.lower() if file_extension not in self.SUPPORTED_FORMATS: raise HTTPException( status_code=400, detail=f"Unsupported file format. Supported formats: {', '.join(sorted(self.SUPPORTED_FORMATS))}" ) # For movie files, use global upload limit if available effective_max_size = max_size if file_extension in self.MOVIE_EXTENSIONS and db: global_limit = self.get_global_upload_limit(db) if global_limit: effective_max_size = global_limit # Check file size (we'll read content later, so this is a preliminary check) if hasattr(file, 'size') and file.size and file.size > effective_max_size: size_mb = effective_max_size // (1024*1024) raise HTTPException( status_code=413, detail=f"File too large. Maximum size is {size_mb}MB" ) def generate_unique_filename(self, original_filename: str, prefix: str = "") -> str: """Generate a unique filename with timestamp and hash.""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") file_stem = Path(original_filename).stem file_extension = Path(original_filename).suffix # Create a short hash for uniqueness hash_input = f"{original_filename}{timestamp}".encode() short_hash = hashlib.md5(hash_input).hexdigest()[:8] if prefix: return f"{prefix}_{file_stem}_{timestamp}_{short_hash}{file_extension}" else: return f"{file_stem}_{timestamp}_{short_hash}{file_extension}" def create_directory_structure(self, task_id: int, file_type: str) -> Path: """Create organized directory structure for file storage.""" if file_type == "attachment": base_dir = self.attachments_dir elif file_type == "submission": base_dir = self.submissions_dir else: raise ValueError(f"Unknown file type: {file_type}") # Create task-specific directory task_dir = base_dir / str(task_id) task_dir.mkdir(exist_ok=True) return task_dir def get_global_upload_limit(self, db: Session) -> Optional[int]: """Get the global upload limit in bytes from database.""" try: from models.global_settings import GlobalSettings setting = db.query(GlobalSettings).filter( GlobalSettings.setting_key == "global_upload_limit_mb" ).first() if setting: limit_mb = int(setting.setting_value) return limit_mb * 1024 * 1024 # Convert MB to bytes except Exception: pass return None def store_relative_path(self, absolute_path: str) -> str: """Convert absolute path to relative path for database storage.""" absolute_path_obj = Path(absolute_path).resolve() try: # Convert to relative path from backend directory relative_path = absolute_path_obj.relative_to(self.backend_dir) return str(relative_path).replace('\\', '/') # Use forward slashes for consistency except ValueError: # If path is not under backend directory, return as-is # This handles edge cases during migration return str(absolute_path_obj).replace('\\', '/') def resolve_absolute_path(self, stored_path: str) -> str: """Resolve stored path (relative or absolute) to absolute path.""" stored_path_obj = Path(stored_path) # If already absolute, return as-is if stored_path_obj.is_absolute(): return str(stored_path_obj) # Otherwise, resolve relative to backend directory absolute_path = self.backend_dir / stored_path_obj return str(absolute_path.resolve()) def is_relative_path(self, path: str) -> bool: """Check if a path is relative to backend directory.""" path_obj = Path(path) return not path_obj.is_absolute() def migrate_path_to_relative(self, absolute_path: str) -> str: """Convert legacy absolute path to new relative format.""" return self.store_relative_path(absolute_path) async def save_file(self, file: UploadFile, task_id: int, file_type: str, version_number: Optional[int] = None, db: Optional[Session] = None) -> Tuple[str, int]: """Save uploaded file and return relative file path and size.""" # Read file content file_content = await file.read() file_size = len(file_content) # Determine max size based on file type and global settings max_size = self.MAX_SUBMISSION_SIZE if file_type == "submission" else self.MAX_ATTACHMENT_SIZE # For movie files, use global upload limit if available file_extension = Path(file.filename).suffix.lower() if file_extension in self.MOVIE_EXTENSIONS and db: global_limit = self.get_global_upload_limit(db) if global_limit: max_size = global_limit # Validate size after reading if file_size > max_size: size_mb = max_size // (1024*1024) raise HTTPException( status_code=413, detail=f"File too large. Maximum size is {size_mb}MB" ) # Create directory structure task_dir = self.create_directory_structure(task_id, file_type) # Generate unique filename prefix = f"v{version_number:03d}" if version_number else "" unique_filename = self.generate_unique_filename(file.filename, prefix) file_path = task_dir / unique_filename # Save file with open(file_path, "wb") as buffer: buffer.write(file_content) # Return relative path for database storage relative_path = self.store_relative_path(str(file_path)) return relative_path, file_size def delete_file(self, file_path: str) -> bool: """Delete a file from the filesystem.""" try: # Resolve to absolute path for filesystem operations absolute_path = self.resolve_absolute_path(file_path) if os.path.exists(absolute_path): os.remove(absolute_path) # Also delete thumbnail if it exists thumbnail_path = self.get_thumbnail_path(file_path) absolute_thumbnail_path = self.resolve_absolute_path(thumbnail_path) if os.path.exists(absolute_thumbnail_path): os.remove(absolute_thumbnail_path) return True except Exception: pass return False def get_thumbnail_path(self, file_path: str) -> str: """Get the thumbnail path for a given file (returns relative path).""" # Resolve to absolute path to get the filename absolute_path = self.resolve_absolute_path(file_path) file_path_obj = Path(absolute_path) thumbnail_name = f"{file_path_obj.stem}_thumb.jpg" # Return relative path for consistency absolute_thumbnail_path = self.thumbnails_dir / thumbnail_name return self.store_relative_path(str(absolute_thumbnail_path)) def create_thumbnail(self, file_path: str) -> Optional[str]: """Create a thumbnail for image files (returns relative path).""" try: # Resolve to absolute path for filesystem operations absolute_file_path = self.resolve_absolute_path(file_path) file_extension = Path(absolute_file_path).suffix.lower() # Only create thumbnails for image formats (excluding EXR for now due to complexity) image_formats = {'.jpg', '.jpeg', '.png', '.tiff', '.tif'} if file_extension not in image_formats: return None # Open and resize image with Image.open(absolute_file_path) as img: # Convert to RGB if necessary if img.mode in ('RGBA', 'LA', 'P'): img = img.convert('RGB') # Create thumbnail img.thumbnail(self.THUMBNAIL_SIZE, Image.Resampling.LANCZOS) # Get thumbnail path (relative) thumbnail_path = self.get_thumbnail_path(file_path) absolute_thumbnail_path = self.resolve_absolute_path(thumbnail_path) # Save thumbnail img.save(absolute_thumbnail_path, 'JPEG', quality=self.THUMBNAIL_QUALITY) return thumbnail_path # Return relative path except Exception as e: # Log error but don't fail the upload print(f"Failed to create thumbnail for {file_path}: {e}") return None def get_file_info(self, file_path: str) -> dict: """Get file information including size, type, and modification time.""" try: # Resolve to absolute path for filesystem operations absolute_path = self.resolve_absolute_path(file_path) file_path_obj = Path(absolute_path) if not file_path_obj.exists(): return {'exists': False} stat = file_path_obj.stat() mime_type, _ = mimetypes.guess_type(str(file_path_obj)) return { 'size': stat.st_size, 'mime_type': mime_type or 'application/octet-stream', 'modified_at': datetime.fromtimestamp(stat.st_mtime), 'exists': True } except Exception: return {'exists': False} def is_image_file(self, file_path: str) -> bool: """Check if file is an image.""" # Works with both relative and absolute paths file_extension = Path(file_path).suffix.lower() image_formats = {'.jpg', '.jpeg', '.png', '.tiff', '.tif', '.exr', '.hdr', '.dpx'} return file_extension in image_formats def is_video_file(self, file_path: str) -> bool: """Check if file is a video.""" # Works with both relative and absolute paths file_extension = Path(file_path).suffix.lower() video_formats = {'.mov', '.mp4', '.avi', '.mkv', '.webm'} return file_extension in video_formats # Global file handler instance file_handler = FileHandler()