refactor: Create admin enpoints and some renaming of api endpoints
This commit is contained in:
JSC
10
app/api/v1/admin/__init__.py
Normal file
10
app/api/v1/admin/__init__.py
Normal file
@@ -0,0 +1,10 @@
|
||||
"""Admin API endpoints."""
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from app.api.v1.admin import sounds
|
||||
|
||||
router = APIRouter(prefix="/admin")
|
||||
|
||||
# Include all admin sub-routers
|
||||
router.include_router(sounds.router)
|
||||
235
app/api/v1/admin/sounds.py
Normal file
235
app/api/v1/admin/sounds.py
Normal file
@@ -0,0 +1,235 @@
|
||||
"""Admin 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_admin_user
|
||||
from app.models.user import User
|
||||
from app.services.extraction_processor import extraction_processor
|
||||
from app.services.sound_normalizer import NormalizationResults, SoundNormalizerService
|
||||
from app.services.sound_scanner import ScanResults, SoundScannerService
|
||||
|
||||
router = APIRouter(prefix="/sounds", tags=["admin-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 ENDPOINTS
|
||||
@router.post("/scan")
|
||||
async def scan_sounds(
|
||||
current_user: Annotated[User, Depends(get_admin_user)],
|
||||
scanner_service: Annotated[SoundScannerService, Depends(get_sound_scanner_service)],
|
||||
) -> dict[str, ScanResults | str]:
|
||||
"""Sync the soundboard directory (add/update/delete sounds). Admin only."""
|
||||
try:
|
||||
results = await scanner_service.scan_soundboard_directory()
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to sync sounds: {e!s}",
|
||||
) from e
|
||||
else:
|
||||
return {
|
||||
"message": "Sound sync completed",
|
||||
"results": results,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/scan/custom")
|
||||
async def scan_custom_directory(
|
||||
directory: str,
|
||||
current_user: Annotated[User, Depends(get_admin_user)],
|
||||
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). Admin only."""
|
||||
try:
|
||||
results = await scanner_service.scan_directory(directory, sound_type)
|
||||
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
|
||||
else:
|
||||
return {
|
||||
"message": f"Sync of directory '{directory}' completed",
|
||||
"results": results,
|
||||
}
|
||||
|
||||
|
||||
# NORMALIZE ENDPOINTS
|
||||
@router.post("/normalize/all")
|
||||
async def normalize_all_sounds(
|
||||
current_user: Annotated[User, Depends(get_admin_user)],
|
||||
normalizer_service: Annotated[
|
||||
SoundNormalizerService,
|
||||
Depends(get_sound_normalizer_service),
|
||||
],
|
||||
force: Annotated[
|
||||
bool,
|
||||
Query( # noqa: FBT002
|
||||
description="Force normalization of already normalized sounds",
|
||||
),
|
||||
] = False,
|
||||
one_pass: Annotated[
|
||||
bool | None,
|
||||
Query(
|
||||
description="Use one-pass normalization (overrides config)",
|
||||
),
|
||||
] = None,
|
||||
) -> dict[str, NormalizationResults | str]:
|
||||
"""Normalize all unnormalized sounds. Admin only."""
|
||||
try:
|
||||
results = await normalizer_service.normalize_all_sounds(
|
||||
force=force,
|
||||
one_pass=one_pass,
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to normalize sounds: {e!s}",
|
||||
) from e
|
||||
else:
|
||||
return {
|
||||
"message": "Sound normalization completed",
|
||||
"results": results,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/normalize/type/{sound_type}")
|
||||
async def normalize_sounds_by_type(
|
||||
sound_type: str,
|
||||
current_user: Annotated[User, Depends(get_admin_user)],
|
||||
normalizer_service: Annotated[
|
||||
SoundNormalizerService,
|
||||
Depends(get_sound_normalizer_service),
|
||||
],
|
||||
force: Annotated[
|
||||
bool,
|
||||
Query( # noqa: FBT002
|
||||
description="Force normalization of already normalized sounds",
|
||||
),
|
||||
] = False,
|
||||
one_pass: Annotated[
|
||||
bool | None,
|
||||
Query(
|
||||
description="Use one-pass normalization (overrides config)",
|
||||
),
|
||||
] = None,
|
||||
) -> dict[str, NormalizationResults | str]:
|
||||
"""Normalize all sounds of a specific type (SDB, TTS, EXT). Admin only."""
|
||||
# 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,
|
||||
)
|
||||
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
|
||||
else:
|
||||
return {
|
||||
"message": f"Normalization of {sound_type} sounds completed",
|
||||
"results": results,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/normalize/{sound_id}")
|
||||
async def normalize_sound_by_id(
|
||||
sound_id: int,
|
||||
current_user: Annotated[User, Depends(get_admin_user)],
|
||||
normalizer_service: Annotated[
|
||||
SoundNormalizerService,
|
||||
Depends(get_sound_normalizer_service),
|
||||
],
|
||||
force: Annotated[
|
||||
bool,
|
||||
Query( # noqa: FBT002
|
||||
description="Force normalization of already normalized sound",
|
||||
),
|
||||
] = False,
|
||||
one_pass: Annotated[
|
||||
bool | None,
|
||||
Query(
|
||||
description="Use one-pass normalization (overrides config)",
|
||||
),
|
||||
] = None,
|
||||
) -> dict[str, str]:
|
||||
"""Normalize a specific sound by ID. Admin only."""
|
||||
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
|
||||
|
||||
|
||||
# EXTRACTION PROCESSOR STATUS
|
||||
@router.get("/extract/status")
|
||||
async def get_extraction_processor_status(
|
||||
current_user: Annotated[User, Depends(get_admin_user)], # noqa: ARG001
|
||||
) -> dict:
|
||||
"""Get the status of the extraction processor. Admin only."""
|
||||
return extraction_processor.get_status()
|
||||
Reference in New Issue
Block a user