icamarillo
d4ff32dac7
- Fixed client-profile GET endpoint to prevent 500 errors - Made ClientProfileResponse fields optional (id, created_at, updated_at) - Returns empty profile data instead of creating DB entry on GET - Added new attachment model and schemas for file handling - Added file handler core utility for upload management
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
"""
|
|
File Handler - ServiceManagerWeb
|
|
Gestión simple de archivos adjuntos
|
|
"""
|
|
import os
|
|
import uuid
|
|
import hashlib
|
|
from pathlib import Path
|
|
from typing import Tuple
|
|
from fastapi import UploadFile, HTTPException, status
|
|
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
class FileHandler:
|
|
"""Handler simple para archivos adjuntos"""
|
|
|
|
def __init__(self):
|
|
self.upload_path = Path(settings.UPLOAD_PATH)
|
|
self.max_size_bytes = settings.MAX_UPLOAD_SIZE_MB * 1024 * 1024
|
|
self.allowed_extensions = settings.ALLOWED_FILE_EXTENSIONS
|
|
# Crear directorio si no existe
|
|
self.upload_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _validate_file(self, filename: str, file_size: int) -> None:
|
|
"""Validar archivo"""
|
|
extension = Path(filename).suffix.lower().lstrip('.')
|
|
|
|
if extension not in self.allowed_extensions:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Extensión no permitida: {extension}"
|
|
)
|
|
|
|
if file_size > self.max_size_bytes:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail=f"Archivo muy grande. Máximo: {settings.MAX_UPLOAD_SIZE_MB}MB"
|
|
)
|
|
|
|
def _calculate_checksums(self, content: bytes) -> Tuple[str, str]:
|
|
"""Calcular MD5 y SHA256"""
|
|
return hashlib.md5(content).hexdigest(), hashlib.sha256(content).hexdigest()
|
|
|
|
async def save_upload(self, file: UploadFile, tenant_id: uuid.UUID, ticket_id: uuid.UUID) -> dict:
|
|
"""Guardar archivo y retornar metadata"""
|
|
if not file.filename:
|
|
raise HTTPException(status_code=400, detail="Filename requerido")
|
|
|
|
content = await file.read()
|
|
file_size = len(content)
|
|
|
|
self._validate_file(file.filename, file_size)
|
|
|
|
md5_hash, sha256_hash = self._calculate_checksums(content)
|
|
|
|
# Nombre único
|
|
extension = Path(file.filename).suffix.lower()
|
|
safe_filename = f"{uuid.uuid4().hex}{extension}"
|
|
|
|
# Estructura: uploads/tenant_id/tickets/ticket_id/
|
|
file_directory = self.upload_path / str(tenant_id) / "tickets" / str(ticket_id)
|
|
file_directory.mkdir(parents=True, exist_ok=True)
|
|
|
|
file_path = file_directory / safe_filename
|
|
relative_path = str(file_path.relative_to(self.upload_path))
|
|
|
|
# Guardar archivo
|
|
with open(file_path, "wb") as f:
|
|
f.write(content)
|
|
|
|
import mimetypes
|
|
mime_type = mimetypes.guess_type(file.filename)[0] or "application/octet-stream"
|
|
|
|
return {
|
|
"filename": safe_filename,
|
|
"original_filename": file.filename,
|
|
"file_path": relative_path,
|
|
"file_size": file_size,
|
|
"mime_type": mime_type,
|
|
"md5_hash": md5_hash,
|
|
"sha256_hash": sha256_hash
|
|
}
|
|
|
|
def get_file_path(self, relative_path: str) -> Path:
|
|
"""Obtener path absoluto del archivo"""
|
|
file_path = (self.upload_path / relative_path).resolve()
|
|
|
|
# Verificar que no escape del directorio de uploads
|
|
if not str(file_path).startswith(str(self.upload_path.resolve())):
|
|
raise HTTPException(status_code=403, detail="Acceso denegado")
|
|
|
|
if not file_path.exists():
|
|
raise HTTPException(status_code=404, detail="Archivo no encontrado")
|
|
|
|
return file_path
|
|
|
|
|
|
file_handler = FileHandler()
|