mirror of
https://github.com/msoedov/agentic_security.git
synced 2026-08-08 18:56:03 +02:00
Added error handling for subprocess.run calls by logging errors and raising AudioGenerationError.
Ensured cleanup of temporary files even if an error occurs.
This commit is contained in:
@@ -1,4 +1,3 @@
|
|||||||
|
|
||||||
from pyfiglet import Figlet, FontNotFound
|
from pyfiglet import Figlet, FontNotFound
|
||||||
from termcolor import colored
|
from termcolor import colored
|
||||||
|
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import base64
|
import base64
|
||||||
|
import logging
|
||||||
import os
|
import os
|
||||||
import platform
|
import platform
|
||||||
import subprocess
|
import subprocess
|
||||||
@@ -7,6 +8,16 @@ import uuid
|
|||||||
import httpx
|
import httpx
|
||||||
from cache_to_disk import cache_to_disk
|
from cache_to_disk import cache_to_disk
|
||||||
|
|
||||||
|
# Configure logging
|
||||||
|
logging.basicConfig(level=logging.INFO)
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class AudioGenerationError(Exception):
|
||||||
|
"""Custom exception for errors during audio generation."""
|
||||||
|
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
def encode(content: bytes) -> str:
|
def encode(content: bytes) -> str:
|
||||||
encoded_content = base64.b64encode(content).decode("utf-8")
|
encoded_content = base64.b64encode(content).decode("utf-8")
|
||||||
@@ -41,12 +52,25 @@ def generate_audio_mac_wav(prompt: str) -> bytes:
|
|||||||
# Read the WAV file into memory
|
# Read the WAV file into memory
|
||||||
with open(temp_wav_path, "rb") as f:
|
with open(temp_wav_path, "rb") as f:
|
||||||
audio_bytes = f.read()
|
audio_bytes = f.read()
|
||||||
|
|
||||||
|
except subprocess.CalledProcessError as e:
|
||||||
|
logger.error(f"Subprocess error: {e}")
|
||||||
|
raise AudioGenerationError("Failed to generate or convert audio.") from e
|
||||||
|
except FileNotFoundError as e:
|
||||||
|
logger.error(f"File not found: {e}")
|
||||||
|
raise AudioGenerationError("Required file not found.") from e
|
||||||
|
except Exception as e:
|
||||||
|
logger.exception("Unexpected error occurred.")
|
||||||
|
raise AudioGenerationError(
|
||||||
|
"An unexpected error occurred during audio generation."
|
||||||
|
) from e
|
||||||
finally:
|
finally:
|
||||||
# Clean up the temporary files
|
for path in (temp_aiff_path, temp_wav_path):
|
||||||
if os.path.exists(temp_aiff_path):
|
try:
|
||||||
os.remove(temp_aiff_path)
|
if os.path.exists(path):
|
||||||
if os.path.exists(temp_wav_path):
|
os.remove(path)
|
||||||
os.remove(temp_wav_path)
|
except Exception as e:
|
||||||
|
logger.warning(f"Failed to delete temporary file {path}: {e}")
|
||||||
|
|
||||||
# Return the audio bytes
|
# Return the audio bytes
|
||||||
return audio_bytes
|
return audio_bytes
|
||||||
|
|||||||
Reference in New Issue
Block a user