File size: 1,903 Bytes
586d5d0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | import tempfile
import os
import subprocess
async def save_temp_file(upload_file):
# Get the original file extension from content type or filename
content_type = upload_file.content_type or ""
filename = upload_file.filename
# Determine file extension
if 'webm' in content_type or filename.endswith('.webm'):
original_suffix = ".webm"
elif 'ogg' in content_type or filename.endswith('.ogg'):
original_suffix = ".ogg"
elif 'mp3' in content_type or filename.endswith('.mp3'):
original_suffix = ".mp3"
elif 'wav' in content_type or filename.endswith('.wav'):
original_suffix = ".wav"
else:
original_suffix = ".wav" # default
# Save original file first
with tempfile.NamedTemporaryFile(delete=False, suffix=original_suffix) as tmp_original:
content = await upload_file.read()
tmp_original.write(content)
original_path = tmp_original.name
# If already WAV, return as is
if original_suffix == ".wav":
return original_path
# Convert to WAV using ffmpeg
try:
wav_path = original_path.replace(original_suffix, ".wav")
# Use ffmpeg for conversion
cmd = ['ffmpeg', '-i', original_path, '-acodec', 'pcm_s16le', '-ar', '16000', wav_path, '-y']
result = subprocess.run(cmd, capture_output=True, timeout=30)
if result.returncode != 0:
raise Exception(f"FFmpeg conversion failed: {result.stderr.decode()}")
# Delete original file
os.remove(original_path)
return wav_path
except Exception as e:
# If conversion fails, try to use original
print(f"Audio conversion error: {e}")
# Cleanup
if os.path.exists(original_path):
os.remove(original_path)
raise Exception(f"Failed to convert audio file: {e}") |