poturaevpetr
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
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,
|
|
sourse: str = "internal"
|
|
) -> Audio:
|
|
"""Создать новую запись аудиофайла"""
|
|
db_audio = Audio(
|
|
filename=audio_data.filename,
|
|
file_path=file_path,
|
|
index_date=datetime.datetime.utcnow(),
|
|
file_size=file_size,
|
|
sourse=sourse
|
|
)
|
|
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})
|
|
|