34e6289f92
- Implemented endpoints for normalizing all sounds, normalizing sounds by type, and normalizing a specific sound by ID in the sounds API. - Added dependency injection for the SoundNormalizerService. - Included role-based access control to restrict normalization actions to admin users. - Created comprehensive test cases for the new normalization endpoints, covering success scenarios, parameter handling, and error cases. - Removed redundant test file for sound normalizer endpoints as tests are now integrated into the main sound endpoints test suite.
236 lines
8.0 KiB
Python
236 lines
8.0 KiB
Python
"""Sound management API endpoints."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, 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_normalizer import NormalizationResults, SoundNormalizerService
|
|
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)
|
|
|
|
|
|
async def get_sound_normalizer_service(
|
|
session: Annotated[AsyncSession, Depends(get_db)],
|
|
) -> SoundNormalizerService:
|
|
"""Get the sound normalizer service."""
|
|
return SoundNormalizerService(session)
|
|
|
|
|
|
# SCAN
|
|
@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
|
|
|
|
|
|
# NORMALIZE
|
|
@router.post("/normalize/all")
|
|
async def normalize_all_sounds(
|
|
current_user: Annotated[User, Depends(get_current_active_user_flexible)],
|
|
normalizer_service: Annotated[
|
|
SoundNormalizerService, Depends(get_sound_normalizer_service)
|
|
],
|
|
force: bool = Query(
|
|
False, description="Force normalization of already normalized sounds"
|
|
),
|
|
one_pass: bool | None = Query(
|
|
None, description="Use one-pass normalization (overrides config)"
|
|
),
|
|
) -> dict[str, NormalizationResults | str]:
|
|
"""Normalize all unnormalized sounds."""
|
|
# Only allow admins to normalize sounds
|
|
if current_user.role not in ["admin", "superadmin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only administrators can normalize sounds",
|
|
)
|
|
|
|
try:
|
|
results = await normalizer_service.normalize_all_sounds(
|
|
force=force,
|
|
one_pass=one_pass,
|
|
)
|
|
return {
|
|
"message": "Sound normalization completed",
|
|
"results": results,
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to normalize sounds: {e!s}",
|
|
) from e
|
|
|
|
|
|
@router.post("/normalize/type/{sound_type}")
|
|
async def normalize_sounds_by_type(
|
|
sound_type: str,
|
|
current_user: Annotated[User, Depends(get_current_active_user_flexible)],
|
|
normalizer_service: Annotated[
|
|
SoundNormalizerService, Depends(get_sound_normalizer_service)
|
|
],
|
|
force: bool = Query(
|
|
False, description="Force normalization of already normalized sounds"
|
|
),
|
|
one_pass: bool | None = Query(
|
|
None, description="Use one-pass normalization (overrides config)"
|
|
),
|
|
) -> dict[str, NormalizationResults | str]:
|
|
"""Normalize all sounds of a specific type (SDB, TTS, EXT)."""
|
|
# Only allow admins to normalize sounds
|
|
if current_user.role not in ["admin", "superadmin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only administrators can normalize sounds",
|
|
)
|
|
|
|
# Validate sound type
|
|
valid_types = ["SDB", "TTS", "EXT"]
|
|
if sound_type not in valid_types:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid sound type. Must be one of: {', '.join(valid_types)}",
|
|
)
|
|
|
|
try:
|
|
results = await normalizer_service.normalize_sounds_by_type(
|
|
sound_type=sound_type,
|
|
force=force,
|
|
one_pass=one_pass,
|
|
)
|
|
return {
|
|
"message": f"Normalization of {sound_type} sounds completed",
|
|
"results": results,
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to normalize {sound_type} sounds: {e!s}",
|
|
) from e
|
|
|
|
|
|
@router.post("/normalize/{sound_id}")
|
|
async def normalize_sound_by_id(
|
|
sound_id: int,
|
|
current_user: Annotated[User, Depends(get_current_active_user_flexible)],
|
|
normalizer_service: Annotated[
|
|
SoundNormalizerService, Depends(get_sound_normalizer_service)
|
|
],
|
|
force: bool = Query(
|
|
False, description="Force normalization of already normalized sound"
|
|
),
|
|
one_pass: bool | None = Query(
|
|
None, description="Use one-pass normalization (overrides config)"
|
|
),
|
|
) -> dict[str, str]:
|
|
"""Normalize a specific sound by ID."""
|
|
# Only allow admins to normalize sounds
|
|
if current_user.role not in ["admin", "superadmin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only administrators can normalize sounds",
|
|
)
|
|
|
|
try:
|
|
# Get the sound
|
|
sound = await normalizer_service.sound_repo.get_by_id(sound_id)
|
|
if not sound:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Sound with ID {sound_id} not found",
|
|
)
|
|
|
|
# Normalize the sound
|
|
result = await normalizer_service.normalize_sound(
|
|
sound=sound,
|
|
force=force,
|
|
one_pass=one_pass,
|
|
)
|
|
|
|
# Check result status
|
|
if result["status"] == "error":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to normalize sound: {result['error']}",
|
|
)
|
|
|
|
return {
|
|
"message": f"Sound normalization {result['status']}: {sound.filename}",
|
|
"status": result["status"],
|
|
"reason": result["reason"] or "",
|
|
"normalized_filename": result["normalized_filename"] or "",
|
|
}
|
|
|
|
except HTTPException:
|
|
# Re-raise HTTPExceptions without wrapping them
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to normalize sound: {e!s}",
|
|
) from e
|