Spaces:
Paused
Paused
| import os | |
| import shutil | |
| from fastapi import UploadFile | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Optional | |
| from ..core.config import settings | |
| from .logger import logger | |
| class FileStorage: | |
| def __init__(self): | |
| self.upload_dir = Path("uploads") | |
| self.upload_dir.mkdir(exist_ok=True) | |
| # Create subdirectories for different file types | |
| self.image_dir = self.upload_dir / "images" | |
| self.document_dir = self.upload_dir / "documents" | |
| self.image_dir.mkdir(exist_ok=True) | |
| self.document_dir.mkdir(exist_ok=True) | |
| async def save_file( | |
| self, | |
| file: UploadFile, | |
| category: str = "documents", | |
| max_size: int = 10 * 1024 * 1024 # 10MB default | |
| ) -> Optional[str]: | |
| try: | |
| # Validate file size | |
| file.file.seek(0, os.SEEK_END) | |
| size = file.file.tell() | |
| file.file.seek(0) | |
| if size > max_size: | |
| raise ValueError(f"File size exceeds maximum limit of {max_size/1024/1024}MB") | |
| # Generate unique filename | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| filename = f"{timestamp}_{file.filename}" | |
| # Determine storage directory based on category | |
| if category == "images": | |
| save_dir = self.image_dir | |
| allowed_types = {".jpg", ".jpeg", ".png", ".gif"} | |
| else: | |
| save_dir = self.document_dir | |
| allowed_types = {".pdf", ".doc", ".docx", ".txt"} | |
| # Validate file type | |
| file_ext = Path(file.filename).suffix.lower() | |
| if file_ext not in allowed_types: | |
| raise ValueError(f"File type {file_ext} not allowed") | |
| # Save file | |
| file_path = save_dir / filename | |
| with file_path.open("wb") as buffer: | |
| shutil.copyfileobj(file.file, buffer) | |
| return str(file_path.relative_to(self.upload_dir)) | |
| except Exception as e: | |
| logger.error(f"File upload error: {str(e)}") | |
| return None | |
| async def delete_file(self, file_path: str) -> bool: | |
| try: | |
| full_path = self.upload_dir / file_path | |
| if full_path.exists(): | |
| full_path.unlink() | |
| return True | |
| return False | |
| except Exception as e: | |
| logger.error(f"File deletion error: {str(e)}") | |
| return False | |
| def get_file_url(self, file_path: str) -> str: | |
| """Generate URL for accessing the file""" | |
| return f"/uploads/{file_path}" | |
| file_storage = FileStorage() |