138 lines
4.0 KiB
Python
138 lines
4.0 KiB
Python
"""Sound management service for admin operations."""
|
|
|
|
import os
|
|
|
|
from app.database import db
|
|
from app.models.sound import Sound
|
|
from app.services.sound_normalizer_service import SoundNormalizerService
|
|
|
|
|
|
class SoundManagementService:
|
|
"""Service for managing sound files and database operations."""
|
|
|
|
@staticmethod
|
|
def get_sounds_with_file_status(
|
|
sound_type: str = "SDB",
|
|
page: int = 1,
|
|
per_page: int = 50,
|
|
) -> dict:
|
|
"""Get paginated sounds with file existence status."""
|
|
# Validate sound type
|
|
if sound_type not in ["SDB", "SAY", "STR"]:
|
|
raise ValueError("Invalid sound type")
|
|
|
|
# Get paginated results
|
|
sounds_query = Sound.query.filter_by(type=sound_type)
|
|
total = sounds_query.count()
|
|
|
|
sounds = (
|
|
sounds_query.offset((page - 1) * per_page).limit(per_page).all()
|
|
)
|
|
|
|
# Convert to detailed dict format with file status
|
|
sounds_data = []
|
|
for sound in sounds:
|
|
sound_dict = sound.to_dict()
|
|
sound_dict.update(
|
|
SoundManagementService._get_file_status(sound),
|
|
)
|
|
sounds_data.append(sound_dict)
|
|
|
|
return {
|
|
"sounds": sounds_data,
|
|
"pagination": {
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"total": total,
|
|
"pages": (total + per_page - 1) // per_page,
|
|
},
|
|
"type": sound_type,
|
|
}
|
|
|
|
@staticmethod
|
|
def _get_file_status(sound: Sound) -> dict:
|
|
"""Get file existence status for a sound."""
|
|
original_path = os.path.join(
|
|
"sounds",
|
|
sound.type.lower(),
|
|
sound.filename,
|
|
)
|
|
status = {"original_exists": os.path.exists(original_path)}
|
|
|
|
if sound.is_normalized and sound.normalized_filename:
|
|
normalized_path = os.path.join(
|
|
"sounds",
|
|
"normalized",
|
|
sound.type.lower(),
|
|
sound.normalized_filename,
|
|
)
|
|
status["normalized_exists"] = os.path.exists(normalized_path)
|
|
else:
|
|
status["normalized_exists"] = False
|
|
|
|
return status
|
|
|
|
@staticmethod
|
|
def delete_sound_with_files(sound_id: int) -> dict:
|
|
"""Delete a sound and its associated files."""
|
|
sound = Sound.query.get(sound_id)
|
|
if not sound:
|
|
raise ValueError("Sound not found")
|
|
|
|
if not sound.is_deletable:
|
|
raise ValueError("Sound is not deletable")
|
|
|
|
errors = []
|
|
|
|
# Delete normalized file if exists
|
|
if sound.is_normalized and sound.normalized_filename:
|
|
normalized_path = os.path.join(
|
|
"sounds",
|
|
"normalized",
|
|
sound.type.lower(),
|
|
sound.normalized_filename,
|
|
)
|
|
if os.path.exists(normalized_path):
|
|
try:
|
|
os.remove(normalized_path)
|
|
except Exception as e:
|
|
errors.append(f"Failed to delete normalized file: {e}")
|
|
|
|
# Delete original file
|
|
original_path = os.path.join(
|
|
"sounds",
|
|
sound.type.lower(),
|
|
sound.filename,
|
|
)
|
|
if os.path.exists(original_path):
|
|
try:
|
|
os.remove(original_path)
|
|
except Exception as e:
|
|
errors.append(f"Failed to delete original file: {e}")
|
|
|
|
if errors:
|
|
raise Exception("; ".join(errors))
|
|
|
|
# Delete database record
|
|
sound_name = sound.name
|
|
db.session.delete(sound)
|
|
db.session.commit()
|
|
|
|
return {
|
|
"message": f"Sound '{sound_name}' deleted successfully",
|
|
"sound_id": sound_id,
|
|
}
|
|
|
|
@staticmethod
|
|
def normalize_sound(
|
|
sound_id: int,
|
|
overwrite: bool = False,
|
|
two_pass: bool = True,
|
|
) -> dict:
|
|
"""Normalize a specific sound."""
|
|
return SoundNormalizerService.normalize_sound(
|
|
sound_id,
|
|
overwrite,
|
|
two_pass,
|
|
)
|