visus-plus/api/app/infrastructure/appointment_files_service.py

import os
import uuid
from typing import Optional

import aiofiles
from fastapi import UploadFile, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.responses import FileResponse
from werkzeug.utils import secure_filename

from app.application.appointment_files_repository import AppointmentFilesRepository
from app.application.appointments_repository import AppointmentsRepository
from app.domain.entities.appointment_file import AppointmentFileEntity
from app.domain.models import AppointmentFile, User


class AppointmentFilesService:
    def __init__(self, db: AsyncSession):
        self.appointment_files_repository = AppointmentFilesRepository(db)
        self.appointments_repository = AppointmentsRepository(db)

    async def get_file_by_id(self, file_id: int, user: User) -> FileResponse:
        appointment_file = await self.appointment_files_repository.get_by_id(file_id)

        if not appointment_file:
            raise HTTPException(404, "Файл с таким ID не найден")

        appointment = await self.appointments_repository.get_by_id(appointment_file.appointment_id)

        if not appointment:
            raise HTTPException(404, "Прием с таким ID не найден")

        if appointment.doctor_id != user.id and user.role.title != 'Администратор':
            raise HTTPException(403, 'Доступ запрещен')

        if not os.path.exists(appointment_file.file_path):
            raise HTTPException(404, "Файл не найден на диске")

        return FileResponse(
            appointment_file.file_path,
            media_type=self.get_media_type(appointment_file.file_path),
            filename=os.path.basename(appointment_file.file_title),
        )

    async def get_files_by_appointment_id(self, appointment_id: int, user: User) -> Optional[
        list[AppointmentFileEntity]
    ]:
        appointment = await self.appointments_repository.get_by_id(appointment_id)

        if not appointment:
            raise HTTPException(404, "Прием с таким ID не найден")

        if appointment.doctor_id != user.id and user.role.title != 'Администратор':
            raise HTTPException(403, 'Доступ запрещен')

        appointment_files = await self.appointment_files_repository.get_by_appointment_id(appointment_id)

        return [
            self.model_to_entity(appointment_file)
            for appointment_file in appointment_files
        ]

    async def upload_file(self, appointment_id: int, file: UploadFile, user: User) -> AppointmentFileEntity:
        appointment = await self.appointments_repository.get_by_id(appointment_id)

        if not appointment:
            raise HTTPException(404, "Прием с таким ID не найден")

        if appointment.doctor_id != user.id and user.role.title != 'Администратор':
            raise HTTPException(403, 'Доступ запрещен')

        file_path = await self.save_file(file, f'uploads/appointment_files/{appointment.id}')

        appointment_file = AppointmentFile(
            file_title=file.filename,
            file_path=file_path,
            appointment_id=appointment_id,
        )

        return self.model_to_entity(
            await self.appointment_files_repository.create(appointment_file)
        )

    async def delete_file(self, file_id: int, user: User) -> AppointmentFileEntity:
        appointment_file = await self.appointment_files_repository.get_by_id(file_id)

        if not appointment_file:
            raise HTTPException(404, "Файл с таким ID не найден")

        appointment = await self.appointments_repository.get_by_id(appointment_file.appointment_id)

        if not appointment:
            raise HTTPException(404, "Прием с таким ID не найден")

        if appointment.doctor_id != user.id and user.role.title != 'Администратор':
            raise HTTPException(403, 'Доступ запрещен')

        if not os.path.exists(appointment_file.file_path):
            raise HTTPException(404, "Файл не найден на диске")

        if os.path.exists(appointment_file.file_path):
            os.remove(appointment_file.file_path)

        return self.model_to_entity(
            await self.appointment_files_repository.delete(appointment_file)
        )

    async def save_file(self, file: UploadFile, upload_dir: str = 'uploads/appointment_files') -> str:
        os.makedirs(upload_dir, exist_ok=True)
        filename = self.generate_filename(file)
        file_path = os.path.join(upload_dir, filename)

        async with aiofiles.open(file_path, 'wb') as out_file:
            content = await file.read()
            await out_file.write(content)
        return file_path

    @staticmethod
    def generate_filename(file: UploadFile) -> str:
        return secure_filename(f"{uuid.uuid4()}_{file.filename}")

    @staticmethod
    def model_to_entity(appointment_file_model: AppointmentFile) -> AppointmentFileEntity:
        return AppointmentFileEntity(
            id=appointment_file_model.id,
            file_path=appointment_file_model.file_path,
            file_title=appointment_file_model.file_title,
            appointment_id=appointment_file_model.appointment_id,
        )

    @staticmethod
    def get_media_type(filename: str) -> str:
        extension = filename.split('.')[-1].lower()
        if extension in ['jpeg', 'jpg', 'png']:
            return f"image/{extension}"
        if extension == 'pdf':
            return "application/pdf"
        if extension in ['zip']:
            return "application/zip"
        if extension in ['doc', 'docx']:
            return "application/msword"
        if extension in ['xls', 'xlsx']:
            return "application/vnd.ms-excel"
        if extension in ['ppt', 'pptx']:
            return "application/vnd.ms-powerpoint"
        if extension in ['txt']:
            return "text/plain"
        return "application/octet-stream"