API для работы с файлами, индексация файлов и результатов распощнавания

apiApp/services/audio_service.py

@@ -0,0 +1,79 @@
+from sqlalchemy.orm import Session
+from typing import List, Optional
+import os
+from apiApp.database import Audio
+from apiApp.schemas import AudioCreate
+from apiApp.config import UPLOAD_FOLDER
+import datetime
+import uuid
+
+
+class AudioCRUD:
+
+    @staticmethod
+    def get_all(db: Session) -> List[Audio]:
+        """Получить все аудиофайлы"""
+        return db.query(Audio).all()
+
+    @staticmethod
+    def get_by_id(db: Session, audio_id: uuid.UUID) -> Optional[Audio]:
+        """Получить аудиофайл по ID"""
+        return db.query(Audio).filter(Audio.id == audio_id).first()
+
+    @staticmethod
+    def get_by_filename(db: Session, filename: str) -> Optional[Audio]:
+        """Получить аудиофайл по имени файла"""
+        return db.query(Audio).filter(Audio.filename == filename).first()
+
+    @staticmethod
+    def create(db: Session, audio_data: AudioCreate, file_path: str, file_size: int = None) -> Audio:
+        """Создать новую запись аудиофайла"""
+        db_audio = Audio(
+            filename=audio_data.filename,
+            file_path=file_path,
+            index_date=datetime.datetime.utcnow(),
+            file_size=file_size
+        )
+        db.add(db_audio)
+        db.commit()
+        db.refresh(db_audio)
+        return db_audio
+
+    @staticmethod
+    def update(db: Session, audio_id: uuid.UUID, audio_data: dict) -> Optional[Audio]:
+        """Обновить запись аудиофайла"""
+        db_audio = db.query(Audio).filter(Audio.id == audio_id).first()
+        if not db_audio:
+            return None
+
+        for key, value in audio_data.items():
+            if hasattr(db_audio, key):
+                setattr(db_audio, key, value)
+
+        db.commit()
+        db.refresh(db_audio)
+        return db_audio
+
+    @staticmethod
+    def delete(db: Session, audio_id: uuid.UUID) -> Optional[Audio]:
+        """Удалить аудиофайл"""
+        db_audio = db.query(Audio).filter(Audio.id == audio_id).first()
+        if not db_audio:
+            return None
+
+        # Удаление файла с диска
+        if db_audio.file_path and os.path.exists(db_audio.file_path):
+            try:
+                os.remove(db_audio.file_path)
+            except Exception as e:
+                print(f"Error deleting file: {e}")
+
+        db.delete(db_audio)
+        db.commit()
+        return db_audio
+
+    @staticmethod
+    def update_recognition_result(db: Session, audio_id: uuid.UUID, result: dict) -> Optional[Audio]:
+        """Обновить результат распознавания"""
+        return AudioCRUD.update(db, audio_id, {"recognition_result": result})
+