""" File serving and thumbnail generation API Provides endpoints for: - On-demand thumbnail generation for images and videos - File serving with proper caching headers """ from typing import Dict from fastapi import APIRouter, Depends, Query, Request from fastapi.responses import Response from PIL import Image from pathlib import Path import subprocess import io from slowapi import Limiter from slowapi.util import get_remote_address from ..core.dependencies import get_current_user from ..core.exceptions import handle_exceptions, NotFoundError, ValidationError from ..core.utils import validate_file_path from modules.universal_logger import get_logger router = APIRouter(prefix="/files", tags=["files"]) logger = get_logger('FilesRouter') limiter = Limiter(key_func=get_remote_address) @router.get("/thumbnail") @limiter.limit("300/minute") @handle_exceptions async def get_thumbnail( request: Request, path: str = Query(..., description="File path"), current_user: Dict = Depends(get_current_user) ): """ Generate and return thumbnail for image or video Args: path: Absolute path to file Returns: JPEG thumbnail (200x200px max, maintains aspect ratio) """ # Validate file is within allowed directories (prevents path traversal) file_path = validate_file_path(path, require_exists=True) file_ext = file_path.suffix.lower() # Generate thumbnail based on type if file_ext in ['.jpg', '.jpeg', '.png', '.webp', '.gif', '.heic']: # Image thumbnail with PIL img = Image.open(file_path) # Convert HEIC if needed if file_ext == '.heic': img = img.convert('RGB') # Create thumbnail (maintains aspect ratio) img.thumbnail((200, 200), Image.Resampling.LANCZOS) # Convert to JPEG buffer = io.BytesIO() if img.mode in ('RGBA', 'LA', 'P'): # Convert transparency to white background background = Image.new('RGB', img.size, (255, 255, 255)) if img.mode == 'P': img = img.convert('RGBA') background.paste(img, mask=img.split()[-1] if img.mode in ('RGBA', 'LA') else None) img = background img.convert('RGB').save(buffer, format='JPEG', quality=85, optimize=True) buffer.seek(0) return Response( content=buffer.read(), media_type="image/jpeg", headers={"Cache-Control": "public, max-age=3600"} ) elif file_ext in ['.mp4', '.webm', '.mov', '.avi', '.mkv']: # Video thumbnail with ffmpeg result = subprocess.run( [ 'ffmpeg', '-ss', '00:00:01', # Seek to 1 second '-i', str(file_path), '-vframes', '1', # Extract 1 frame '-vf', 'scale=200:-1', # Scale to 200px width, maintain aspect '-f', 'image2pipe', # Output to pipe '-vcodec', 'mjpeg', # JPEG codec 'pipe:1' ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False ) if result.returncode != 0: # Try without seeking (for very short videos) result = subprocess.run( [ 'ffmpeg', '-i', str(file_path), '-vframes', '1', '-vf', 'scale=200:-1', '-f', 'image2pipe', '-vcodec', 'mjpeg', 'pipe:1' ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=True ) return Response( content=result.stdout, media_type="image/jpeg", headers={"Cache-Control": "public, max-age=3600"} ) else: raise ValidationError("Unsupported file type")