|
""" |
|
Created By: ishwor subedi |
|
Date: 2024-07-31 |
|
""" |
|
import os |
|
import tempfile |
|
from fastapi import Form |
|
from fastapi import UploadFile, HTTPException, status |
|
from src.models.apis_models import TextToSpeechRequest |
|
from fastapi.routing import APIRouter |
|
from src.pipeline.speech_transcription_pipeline import SpeechTranscriptionPipeline |
|
from src import logging as logger |
|
from src.utils.error_handling import create_success_response, raise_http_exception |
|
|
|
speech_translator_router = APIRouter(tags=["SpeechTranscription"]) |
|
pipeline = SpeechTranscriptionPipeline() |
|
|
|
|
|
@speech_translator_router.post( |
|
"/text_to_speech", |
|
) |
|
async def text_to_speech(request: TextToSpeechRequest): |
|
logger.info(f">>>text_to_speech API Triggered <<<") |
|
try: |
|
audio_bytes = pipeline.text_to_speech(request.text, request.lang, request.tld) |
|
if not audio_bytes: |
|
raise ValueError("Audio generation failed.") |
|
response = create_success_response(code=200, data={"audio": audio_bytes}) |
|
logger.info(f">>>text_to_speech API success <<<") |
|
return response |
|
except ValueError as ve: |
|
logger.info(f">>>text_to_speech API failed {ve}<<<") |
|
raise_http_exception(code=400, message="Text to speech failed") |
|
|
|
except Exception as e: |
|
logger.error(f">>> Error processing text-to-speech {e}<<<") |
|
raise_http_exception(code=500, message="Internal server error") |
|
|
|
|
|
@speech_translator_router.post( |
|
"/speech_to_text", |
|
|
|
) |
|
async def speech_to_text(audio: UploadFile, lang: str = Form(...)): |
|
logger.info(f">>>speech_to_text API Triggered <<<") |
|
try: |
|
audio_bytes = await audio.read() |
|
if not audio_bytes: |
|
logger.error(f">>> Empty audio file <<<") |
|
raise ValueError("Empty audio file") |
|
except Exception as e: |
|
logger.error(f">>> Invalid audio file {e}<<<") |
|
raise HTTPException( |
|
status_code=status.HTTP_400_BAD_REQUEST, |
|
detail="Invalid audio file" |
|
) |
|
|
|
try: |
|
with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_audio_file: |
|
temp_audio_file.write(audio_bytes) |
|
temp_audio_file_path = temp_audio_file.name |
|
except Exception as e: |
|
logger.error(f">>> Error creating temporary file{e} <<<") |
|
raise_http_exception(code=500, message="Internal server error") |
|
|
|
try: |
|
transcript = pipeline.speech_to_text(temp_audio_file_path, lang) |
|
response = create_success_response(code=200, data={"transcript": transcript}) |
|
logger.info(f">>>speech_to_text API success <<<") |
|
|
|
return response |
|
|
|
except FileNotFoundError: |
|
logger.error(f">>> Temporary file not found <<<") |
|
raise HTTPException( |
|
status_code=status.HTTP_404_NOT_FOUND, |
|
detail="Temporary file not found" |
|
) |
|
except Exception as e: |
|
logger.error(f">>> Error processing speech-to-text {e}<<<") |
|
raise_http_exception(code=500, message="Error processing speech-to-text") |
|
|
|
finally: |
|
if os.path.exists(temp_audio_file_path): |
|
os.remove(temp_audio_file_path) |
|
|
