36949a1f1c
- Implemented SoundRepository for database operations related to sounds, including methods for retrieving, creating, updating, and deleting sound records. - Developed SoundScannerService to scan directories for audio files, calculate their metadata, and synchronize with the database. - Added support for various audio file formats and integrated ffmpeg for audio duration extraction. - Created comprehensive tests for sound API endpoints and sound scanner service to ensure functionality and error handling. - Updated dependencies to include ffmpeg-python for audio processing.
80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
"""Sound management API endpoints."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.core.dependencies import get_current_active_user_flexible
|
|
from app.models.user import User
|
|
from app.services.sound_scanner import ScanResults, SoundScannerService
|
|
|
|
router = APIRouter(prefix="/sounds", tags=["sounds"])
|
|
|
|
|
|
async def get_sound_scanner_service(
|
|
session: Annotated[AsyncSession, Depends(get_db)],
|
|
) -> SoundScannerService:
|
|
"""Get the sound scanner service."""
|
|
return SoundScannerService(session)
|
|
|
|
|
|
@router.post("/scan")
|
|
async def scan_sounds(
|
|
current_user: Annotated[User, Depends(get_current_active_user_flexible)],
|
|
scanner_service: Annotated[SoundScannerService, Depends(get_sound_scanner_service)],
|
|
) -> dict[str, ScanResults | str]:
|
|
"""Sync the soundboard directory (add/update/delete sounds)."""
|
|
# Only allow admins to scan sounds
|
|
if current_user.role not in ["admin", "superadmin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only administrators can sync sounds",
|
|
)
|
|
|
|
try:
|
|
results = await scanner_service.scan_soundboard_directory()
|
|
return {
|
|
"message": "Sound sync completed",
|
|
"results": results,
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to sync sounds: {e!s}",
|
|
) from e
|
|
|
|
|
|
@router.post("/scan/custom")
|
|
async def scan_custom_directory(
|
|
directory: str,
|
|
current_user: Annotated[User, Depends(get_current_active_user_flexible)],
|
|
scanner_service: Annotated[SoundScannerService, Depends(get_sound_scanner_service)],
|
|
sound_type: str = "SDB",
|
|
) -> dict[str, ScanResults | str]:
|
|
"""Sync a custom directory with the database (add/update/delete sounds)."""
|
|
# Only allow admins to sync sounds
|
|
if current_user.role not in ["admin", "superadmin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only administrators can sync sounds",
|
|
)
|
|
|
|
try:
|
|
results = await scanner_service.scan_directory(directory, sound_type)
|
|
return {
|
|
"message": f"Sync of directory '{directory}' completed",
|
|
"results": results,
|
|
}
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e),
|
|
) from e
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to sync directory: {e!s}",
|
|
) from e
|