314 lines
9.9 KiB
Python
314 lines
9.9 KiB
Python
"""Main file service for handling file uploads"""
|
|
|
|
from io import BytesIO
|
|
from typing import Any, Dict, Optional
|
|
|
|
from flask import current_app
|
|
|
|
from app import db
|
|
from app.models import FileAttachment
|
|
from app.services.storage.base import StorageAdapter
|
|
from app.services.storage.file_processor import ProcessorFactory
|
|
|
|
|
|
class FileService:
|
|
"""Service for handling file upload and storage operations"""
|
|
|
|
def __init__(self, storage_adapter: Optional[StorageAdapter] = None):
|
|
"""
|
|
Initialize FileService with dependency injection
|
|
|
|
Args:
|
|
storage_adapter: Storage adapter
|
|
(optional, defaults to factory-provided adapter)
|
|
"""
|
|
if storage_adapter:
|
|
self.storage = storage_adapter
|
|
else:
|
|
from app.services.storage.storage_factory import StorageFactory
|
|
|
|
self.storage = StorageFactory.get_storage()
|
|
|
|
def upload_file(
|
|
self,
|
|
file_data: BytesIO,
|
|
filename: str,
|
|
content_type: str,
|
|
file_size: int,
|
|
uploaded_by: int,
|
|
attachable_type: str,
|
|
attachable_id: int,
|
|
) -> FileAttachment:
|
|
"""
|
|
Upload a file and create FileAttachment record
|
|
|
|
Args:
|
|
file_data: File data as BytesIO
|
|
filename: Original filename
|
|
content_type: MIME type
|
|
file_size: File size in bytes
|
|
uploaded_by: User ID who uploaded the file
|
|
attachable_type: Type of entity (Card, Comment, etc.)
|
|
attachable_id: ID of the entity
|
|
|
|
Returns:
|
|
FileAttachment: Created attachment record
|
|
"""
|
|
# Get appropriate processor
|
|
processor = ProcessorFactory.get_processor(content_type)
|
|
|
|
# Validate file
|
|
if not processor.validate(file_data, file_size, content_type):
|
|
raise ValueError("File validation failed")
|
|
|
|
# Process file and get metadata (this also generates thumbnail)
|
|
file_data.seek(0) # Ensure file is at start
|
|
process_result = processor.process(file_data, content_type)
|
|
metadata = process_result["metadata"]
|
|
thumbnail_data = process_result["thumbnail_data"]
|
|
|
|
# Seek back to start for main file upload
|
|
file_data.seek(0)
|
|
|
|
# Determine bucket
|
|
bucket_name = self._get_bucket_for_type(content_type)
|
|
|
|
# Generate unique object name
|
|
object_name = self.storage.generate_unique_object_name(filename)
|
|
|
|
# Upload main file
|
|
self.storage.upload_file(
|
|
file_data, bucket_name, object_name, content_type, file_size
|
|
)
|
|
|
|
# Upload thumbnail if available
|
|
thumbnail_info = {}
|
|
if thumbnail_data:
|
|
try:
|
|
thumb_bucket = current_app.config["MINIO_THUMBNAILS_BUCKET"]
|
|
thumb_object_name = f"thumb_{object_name}"
|
|
|
|
# Get thumbnail length
|
|
thumbnail_data.seek(0, 2)
|
|
thumb_length = thumbnail_data.tell()
|
|
thumbnail_data.seek(0)
|
|
|
|
# Upload thumbnail
|
|
self.storage.upload_file(
|
|
thumbnail_data,
|
|
thumb_bucket,
|
|
thumb_object_name,
|
|
"image/jpeg",
|
|
length=thumb_length,
|
|
)
|
|
|
|
thumbnail_info = {
|
|
"thumbnail_minio_object_name": thumb_object_name,
|
|
"thumbnail_minio_bucket": thumb_bucket,
|
|
}
|
|
|
|
current_app.logger.info(
|
|
f"Thumbnail uploaded: {thumb_bucket}/{thumb_object_name}"
|
|
)
|
|
except Exception as e:
|
|
current_app.logger.error(f"Failed to upload thumbnail: {e}")
|
|
# Continue without thumbnail
|
|
|
|
# Create database record
|
|
attachment = FileAttachment(
|
|
filename=object_name,
|
|
original_name=filename,
|
|
file_type=metadata.get("file_type", "document"),
|
|
mime_type=content_type,
|
|
file_size=file_size,
|
|
minio_bucket=bucket_name,
|
|
minio_object_name=object_name,
|
|
uploaded_by=uploaded_by,
|
|
attachable_type=attachable_type,
|
|
attachable_id=attachable_id,
|
|
**thumbnail_info,
|
|
)
|
|
|
|
db.session.add(attachment)
|
|
db.session.commit()
|
|
|
|
current_app.logger.info(
|
|
f"File uploaded: {filename} -> {bucket_name}/{object_name}"
|
|
)
|
|
|
|
return attachment
|
|
|
|
def delete_file(self, attachment_id: int) -> bool:
|
|
"""
|
|
Delete a file and its attachment record
|
|
|
|
Args:
|
|
attachment_id: ID of the FileAttachment
|
|
|
|
Returns:
|
|
bool: True if successful
|
|
"""
|
|
attachment = db.session.get(FileAttachment, attachment_id)
|
|
if not attachment:
|
|
return False
|
|
|
|
# Delete from MinIO
|
|
try:
|
|
self.storage.delete_file(
|
|
attachment.minio_bucket, attachment.minio_object_name
|
|
)
|
|
|
|
# Delete thumbnail if exists
|
|
if (
|
|
attachment.thumbnail_minio_bucket
|
|
and attachment.thumbnail_minio_object_name
|
|
):
|
|
self.storage.delete_file(
|
|
attachment.thumbnail_minio_bucket,
|
|
attachment.thumbnail_minio_object_name,
|
|
)
|
|
except Exception as e:
|
|
current_app.logger.error(f"Error deleting file from MinIO: {e}")
|
|
|
|
# Delete from database
|
|
attachment.soft_delete()
|
|
|
|
current_app.logger.info(f"File deleted: {attachment.original_name}")
|
|
return True
|
|
|
|
def get_file_url(self, attachment_id: int, expires_in: int = 3600) -> str:
|
|
"""
|
|
Generate a presigned URL for file access
|
|
|
|
Args:
|
|
attachment_id: ID of the FileAttachment
|
|
expires_in: URL expiration time in seconds
|
|
|
|
Returns:
|
|
str: Presigned URL
|
|
"""
|
|
attachment = db.session.get(FileAttachment, attachment_id)
|
|
if not attachment:
|
|
raise ValueError("Attachment not found")
|
|
|
|
return self.storage.generate_presigned_url(
|
|
attachment.minio_bucket,
|
|
attachment.minio_object_name,
|
|
expires_in,
|
|
)
|
|
|
|
def get_thumbnail_url(
|
|
self, attachment_id: int, expires_in: int = 3600
|
|
) -> Optional[str]:
|
|
"""
|
|
Generate a presigned URL for thumbnail access
|
|
|
|
Args:
|
|
attachment_id: ID of the FileAttachment
|
|
expires_in: URL expiration time in seconds
|
|
|
|
Returns:
|
|
str: Presigned URL or None if no thumbnail
|
|
"""
|
|
attachment = db.session.get(FileAttachment, attachment_id)
|
|
if not attachment or not attachment.thumbnail_minio_bucket:
|
|
return None
|
|
|
|
return self.storage.generate_presigned_url(
|
|
attachment.thumbnail_minio_bucket,
|
|
attachment.thumbnail_minio_object_name,
|
|
expires_in,
|
|
)
|
|
|
|
def _get_bucket_for_type(self, mime_type: str) -> str:
|
|
"""
|
|
Get appropriate bucket for MIME type
|
|
|
|
Args:
|
|
mime_type: MIME type of the file
|
|
|
|
Returns:
|
|
str: Bucket name
|
|
"""
|
|
if mime_type.startswith("image/"):
|
|
return current_app.config["MINIO_IMAGES_BUCKET"]
|
|
elif mime_type == "application/pdf":
|
|
return current_app.config["MINIO_DOCUMENTS_BUCKET"]
|
|
else:
|
|
return current_app.config["MINIO_DOCUMENTS_BUCKET"]
|
|
|
|
def get_file_by_id(self, attachment_id: int) -> Optional[FileAttachment]:
|
|
"""
|
|
Get file attachment by ID
|
|
|
|
Args:
|
|
attachment_id: ID of the FileAttachment
|
|
|
|
Returns:
|
|
FileAttachment or None
|
|
"""
|
|
return db.session.get(FileAttachment, attachment_id)
|
|
|
|
def get_files_for_entity(
|
|
self, attachable_type: str, attachable_id: int
|
|
) -> list[FileAttachment]:
|
|
"""
|
|
Get all files attached to an entity
|
|
|
|
Args:
|
|
attachable_type: Type of entity (Card, Comment, etc.)
|
|
attachable_id: ID of the entity
|
|
|
|
Returns:
|
|
list: List of FileAttachment objects
|
|
"""
|
|
return FileAttachment.query.filter_by(
|
|
attachable_type=attachable_type, attachable_id=attachable_id
|
|
).all()
|
|
|
|
def get_file_info(self, attachment_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get file information including UUID-based secure URLs
|
|
|
|
Security: Only exposes public-facing information.
|
|
Internal details (MinIO paths, IDs, entity references) are hidden.
|
|
|
|
Uses UUID for secure file access:
|
|
- UUIDs are non-sequential and hard to guess
|
|
- All endpoints require JWT authentication
|
|
- MinIO server stays hidden from public
|
|
- Frontend uses blob URLs for secure image display
|
|
|
|
Args:
|
|
attachment_id: ID of the FileAttachment
|
|
|
|
Returns:
|
|
dict: File information with UUID-based URLs
|
|
"""
|
|
attachment = self.get_file_by_id(attachment_id)
|
|
if not attachment:
|
|
return None
|
|
|
|
# Use UUID-based secure URLs (all require JWT authentication)
|
|
download_url = f"/files/{attachment.uuid}/download"
|
|
thumbnail_url = (
|
|
f"/files/{attachment.uuid}/thumbnail"
|
|
if attachment.thumbnail_minio_object_name
|
|
else None
|
|
)
|
|
view_url = f"/files/{attachment.uuid}/view"
|
|
|
|
# Only return public-facing fields (security: hide internal implementation)
|
|
return {
|
|
"uuid": attachment.uuid,
|
|
"original_name": attachment.original_name,
|
|
"file_type": attachment.file_type,
|
|
"file_size": attachment.file_size,
|
|
"mime_type": attachment.mime_type,
|
|
"created_at": attachment.created_at.isoformat()
|
|
if attachment.created_at
|
|
else None,
|
|
"download_url": download_url,
|
|
"thumbnail_url": thumbnail_url,
|
|
"view_url": view_url,
|
|
}
|