5fb742a312
- Implement circuit breaker pattern for GPU health checks - Prevents repeated failures with configurable thresholds - Three states: CLOSED, OPEN, HALF_OPEN - Integrated into GPU health monitoring - Add comprehensive input validation and path sanitization - Path traversal attack prevention - Whitelist-based validation for models, devices, formats - Error message sanitization to prevent information leakage - File size limits and security checks - Centralize startup logic across servers - Extract common startup procedures to utils/startup.py - Deduplicate GPU health checks and initialization code - Simplify both MCP and API server startup sequences - Add proper Python package structure - Add __init__.py files to all modules - Improve package organization - Add circuit breaker status API endpoints - GET /health/circuit-breaker - View circuit breaker stats - POST /health/circuit-breaker/reset - Reset circuit breaker - Reorganize test files into tests/ directory - Rename and restructure test files for better organization
68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Audio Processing Module
|
|
Responsible for audio file validation and preprocessing
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
from typing import Union, Any
|
|
from pathlib import Path
|
|
from faster_whisper import decode_audio
|
|
|
|
from utils.input_validation import (
|
|
validate_audio_file as validate_audio_file_secure,
|
|
sanitize_error_message
|
|
)
|
|
|
|
# Log configuration
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def validate_audio_file(audio_path: str, allowed_dirs: list = None) -> None:
|
|
"""
|
|
Validate if an audio file is valid (with security checks).
|
|
|
|
Args:
|
|
audio_path: Path to the audio file
|
|
allowed_dirs: Optional list of allowed base directories
|
|
|
|
Raises:
|
|
FileNotFoundError: If audio file doesn't exist
|
|
ValueError: If audio file format is unsupported or file is empty
|
|
OSError: If file size cannot be checked
|
|
|
|
Returns:
|
|
None: If validation passes
|
|
"""
|
|
try:
|
|
# Use secure validation
|
|
validate_audio_file_secure(audio_path, allowed_dirs)
|
|
except Exception as e:
|
|
# Re-raise with sanitized error messages
|
|
error_msg = sanitize_error_message(str(e))
|
|
|
|
if "not found" in str(e).lower():
|
|
raise FileNotFoundError(error_msg)
|
|
elif "size" in str(e).lower():
|
|
raise OSError(error_msg)
|
|
else:
|
|
raise ValueError(error_msg)
|
|
|
|
def process_audio(audio_path: str) -> Union[str, Any]:
|
|
"""
|
|
Process audio file, perform decoding and preprocessing
|
|
|
|
Args:
|
|
audio_path: Path to the audio file
|
|
|
|
Returns:
|
|
Union[str, Any]: Processed audio data or original file path
|
|
"""
|
|
# Try to preprocess audio using decode_audio to handle more formats
|
|
try:
|
|
audio_data = decode_audio(audio_path)
|
|
logger.info(f"Successfully preprocessed audio: {os.path.basename(audio_path)}")
|
|
return audio_data
|
|
except Exception as audio_error:
|
|
logger.warning(f"Audio preprocessing failed, will use file path directly: {str(audio_error)}")
|
|
return audio_path |