import logging import os import shutil from fastapi import UploadFile, HTTPException, status from typing import BinaryIO, Tuple from app.core.config import settings from app.models.file import File logger = logging.getLogger(__name__) class FileService: """Service for handling file operations.""" @staticmethod async def save_file(upload_file: UploadFile) -> Tuple[str, str, int]: """ Save an uploaded file to disk. Args: upload_file: The uploaded file Returns: Tuple of (unique_filename, file_path, file_size) Raises: HTTPException: If file size exceeds the maximum allowed size """ try: # Generate a unique filename to prevent collisions original_filename = upload_file.filename or "unnamed_file" unique_filename = File.generate_unique_filename(original_filename) # Define the path where the file will be saved file_path = os.path.join(settings.FILE_STORAGE_DIR, unique_filename) # Check file size # First, we need to get the file size upload_file.file.seek(0, os.SEEK_END) file_size = upload_file.file.tell() upload_file.file.seek(0) # Reset file position if file_size > settings.MAX_FILE_SIZE: raise HTTPException( status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail=f"File size exceeds the maximum allowed size of {settings.MAX_FILE_SIZE} bytes", ) # Save the file with open(file_path, "wb") as buffer: shutil.copyfileobj(upload_file.file, buffer) logger.info(f"File saved successfully at {file_path}") return unique_filename, file_path, file_size except PermissionError as e: logger.error(f"Permission denied when saving file: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Permission denied when saving file. The server may not have write access to the storage directory.", ) except Exception as e: logger.error(f"Error saving file: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="An error occurred while saving the file", ) @staticmethod def get_file_path(filename: str) -> str: """ Get the full path for a file. Args: filename: The filename to retrieve Returns: String path to the file """ return os.path.join(settings.FILE_STORAGE_DIR, filename) @staticmethod def file_exists(filename: str) -> bool: """ Check if a file exists on disk. Args: filename: The filename to check Returns: True if the file exists, False otherwise """ file_path = FileService.get_file_path(filename) return os.path.exists(file_path) and os.path.isfile(file_path) @staticmethod def delete_file(filename: str) -> bool: """ Delete a file from disk. Args: filename: The filename to delete Returns: True if the file was deleted, False otherwise """ try: file_path = FileService.get_file_path(filename) if os.path.exists(file_path) and os.path.isfile(file_path): os.remove(file_path) logger.info(f"File deleted: {file_path}") return True return False except PermissionError as e: logger.error(f"Permission denied when deleting file: {e}") return False except Exception as e: logger.error(f"Error deleting file: {e}") return False @staticmethod def get_file_content(filename: str) -> BinaryIO: """ Get file content as a binary file object. Args: filename: The filename to retrieve Returns: File object opened in binary mode Raises: HTTPException: If the file does not exist """ try: file_path = FileService.get_file_path(filename) if not os.path.exists(file_path) or not os.path.isfile(file_path): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="File not found", ) return open(file_path, "rb") except PermissionError: logger.error(f"Permission denied when reading file: {filename}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Permission denied when reading file", ) except Exception as e: logger.error(f"Error reading file {filename}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="An error occurred while reading the file", )