315 lines
13 KiB
Python
315 lines
13 KiB
Python
"""
|
|
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() |