Update app.py

app.py

@@ -9,14 +9,15 @@ import logging
 import torch
 import librosa
 from pathlib import Path
-import magic  # For MIME type detection
 from pydub import AudioSegment
+from moviepy.editor import VideoFileClip
 import traceback
 from logging.handlers import RotatingFileHandler
 import os
 import boto3
 from botocore.exceptions import NoCredentialsError
 import time
+import tempfile
 
 # Import functions from other modules
 from asr import transcribe, ASR_LANGUAGES
@@ -61,32 +62,62 @@ class TTSRequest(BaseModel):
     language: str
     speed: float
 
-def detect_mime_type(input_bytes):
-    mime = magic.Magic(mime=True)
-    return mime.from_buffer(input_bytes)
-
-def extract_audio(input_bytes):
-    mime_type = detect_mime_type(input_bytes)
-    
-    if mime_type.startswith('audio/'):
-        return sf.read(io.BytesIO(input_bytes))
-    elif mime_type.startswith('video/webm'):
-        audio = AudioSegment.from_file(io.BytesIO(input_bytes), format="webm")
-        audio_array = np.array(audio.get_array_of_samples())
-        sample_rate = audio.frame_rate
+def extract_audio_from_file(input_bytes):
+    with tempfile.NamedTemporaryFile(delete=False, suffix='.tmp') as temp_file:
+        temp_file.write(input_bytes)
+        temp_file_path = temp_file.name
+
+    try:
+        # First, try to read as a standard audio file
+        audio_array, sample_rate = sf.read(temp_file_path)
         return audio_array, sample_rate
-    else:
-        raise ValueError(f"Unsupported MIME type: {mime_type}")
+    except Exception:
+        try:
+            # Try to read as a video file
+            video = VideoFileClip(temp_file_path)
+            audio = video.audio
+            if audio is not None:
+                # Extract audio from video
+                audio_array = audio.to_soundarray()
+                sample_rate = audio.fps
+                
+                # Convert to mono if stereo
+                if len(audio_array.shape) > 1 and audio_array.shape[1] > 1:
+                    audio_array = audio_array.mean(axis=1)
+                
+                # Ensure audio is float32 and normalized
+                audio_array = audio_array.astype(np.float32)
+                audio_array /= np.max(np.abs(audio_array))
+                
+                video.close()
+                return audio_array, sample_rate
+            else:
+                raise ValueError("Video file contains no audio")
+        except Exception:
+            # If video reading fails, try as generic audio with pydub
+            try:
+                audio = AudioSegment.from_file(temp_file_path)
+                audio_array = np.array(audio.get_array_of_samples())
+                
+                # Convert to float32 and normalize
+                audio_array = audio_array.astype(np.float32) / (2**15 if audio.sample_width == 2 else 2**7)
+                
+                # Convert stereo to mono if necessary
+                if audio.channels == 2:
+                    audio_array = audio_array.reshape((-1, 2)).mean(axis=1)
+                
+                return audio_array, audio.frame_rate
+            except Exception as e:
+                raise ValueError(f"Unsupported file format: {str(e)}")
+    finally:
+        # Clean up the temporary file
+        os.unlink(temp_file_path)
 
 @app.post("/transcribe")
 async def transcribe_audio(request: AudioRequest):
     try:
         input_bytes = base64.b64decode(request.audio)
-        audio_array, sample_rate = extract_audio(input_bytes)
-
-        # Convert to mono if stereo
-        if len(audio_array.shape) > 1:
-            audio_array = audio_array.mean(axis=1)
+        audio_array, sample_rate = extract_audio_from_file(input_bytes)
 
         # Ensure audio_array is float32
         audio_array = audio_array.astype(np.float32)
@@ -206,7 +237,7 @@ async def synthesize_speech(request: TTSRequest):
 async def identify_language(request: AudioRequest):
     try:
         input_bytes = base64.b64decode(request.audio)
-        audio_array, sample_rate = extract_audio(input_bytes)
+        audio_array, sample_rate = extract_audio_from_file(input_bytes)
         result = identify(audio_array)
         return JSONResponse(content={"language_identification": result})
     except Exception as e: