indigo
23740af816
Coordinators can configure, per task type, accepted file extensions, a
naming convention, and (for video files) required resolution/format/
codec/frame rate. All checking happens entirely client-side in the
browser before upload - the backend only stores and validates the
configuration itself, never inspects submitted files.
- backend: submission_config_by_task_type JSON column on Project,
schemas/submission_config.py for validation, GET/PUT
/projects/{id}/submission-config endpoints. Added .mxf/.ma/.usd(a/c)
to the supported file formats.
- frontend: new Submissions tab in Project Settings
(SubmissionConfigManager.vue) to configure rules per task type.
TaskSubmissions.vue enforces them before upload: extension/naming
checks via regex, and real video resolution/codec/frame rate
detection via mediainfo.js (parses container metadata directly,
so it works for formats browsers can't play natively like MXF,
ProRes, or DNxHD).
317 lines
13 KiB
Python
317 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', '.mxf',
|
|
# Image formats
|
|
'.exr', '.jpg', '.jpeg', '.png', '.tiff', '.tif', '.dpx', '.hdr',
|
|
# Document formats
|
|
'.pdf', '.txt', '.doc', '.docx',
|
|
# Archive formats
|
|
'.zip', '.rar', '.7z',
|
|
# Scene formats
|
|
'.ma', '.usd', '.usda', '.usdc'
|
|
}
|
|
|
|
# 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', '.mxf'}
|
|
|
|
# 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() |