175 lines
6.0 KiB
Python
175 lines
6.0 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"""
|
|
|
|
_CHUNK_SIZE_BYTES = 1024 * 1024 # 1MB
|
|
|
|
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_extension(self, filename: str) -> str:
|
|
"""Validar extensión del archivo y retornarla."""
|
|
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}"
|
|
)
|
|
|
|
return extension
|
|
|
|
def _validate_magic_bytes(self, extension: str, first_bytes: bytes) -> None:
|
|
"""Validación básica por firma (magic bytes) para tipos comunes."""
|
|
|
|
signatures = {
|
|
# PDFs start with %PDF-
|
|
"pdf": [b"%PDF-"],
|
|
# PNG signature
|
|
"png": [b"\x89PNG\r\n\x1a\n"],
|
|
# JPEG starts with FF D8 FF
|
|
"jpg": [b"\xff\xd8\xff"],
|
|
"jpeg": [b"\xff\xd8\xff"],
|
|
# Legacy MS Office (OLE Compound File)
|
|
"doc": [b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"],
|
|
"xls": [b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"],
|
|
# OOXML (zip-based)
|
|
"docx": [b"PK\x03\x04", b"PK\x05\x06", b"PK\x07\x08"],
|
|
"xlsx": [b"PK\x03\x04", b"PK\x05\x06", b"PK\x07\x08"],
|
|
}
|
|
|
|
# For plain text, we can't reliably validate via magic bytes.
|
|
if extension == "txt":
|
|
return
|
|
|
|
allowed = signatures.get(extension)
|
|
if not allowed:
|
|
return
|
|
|
|
if not any(first_bytes.startswith(sig) for sig in allowed):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Contenido de archivo no coincide con la extensión declarada",
|
|
)
|
|
|
|
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")
|
|
|
|
extension = self._validate_extension(file.filename)
|
|
|
|
# Nombre único
|
|
original_extension = Path(file.filename).suffix.lower()
|
|
safe_filename = f"{uuid.uuid4().hex}{original_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 (streaming) + checksums incrementales
|
|
md5 = hashlib.md5()
|
|
sha256 = hashlib.sha256()
|
|
file_size = 0
|
|
validated_magic = False
|
|
first_bytes: bytes = b""
|
|
|
|
try:
|
|
with open(file_path, "wb") as f:
|
|
while True:
|
|
chunk = await file.read(self._CHUNK_SIZE_BYTES)
|
|
if not chunk:
|
|
break
|
|
|
|
if not validated_magic:
|
|
first_bytes = chunk[:16]
|
|
self._validate_magic_bytes(extension, first_bytes)
|
|
validated_magic = True
|
|
|
|
file_size += len(chunk)
|
|
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",
|
|
)
|
|
|
|
md5.update(chunk)
|
|
sha256.update(chunk)
|
|
f.write(chunk)
|
|
|
|
if file_size == 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Archivo vacío",
|
|
)
|
|
|
|
except HTTPException:
|
|
# Eliminar archivo parcial si existe
|
|
try:
|
|
if file_path.exists():
|
|
file_path.unlink()
|
|
except Exception:
|
|
pass
|
|
raise
|
|
except Exception as exc:
|
|
try:
|
|
if file_path.exists():
|
|
file_path.unlink()
|
|
except Exception:
|
|
pass
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error guardando archivo: {exc}",
|
|
)
|
|
|
|
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.hexdigest(),
|
|
"sha256_hash": sha256.hexdigest(),
|
|
}
|
|
|
|
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()
|