| import tempfile |
| import os |
| import subprocess |
|
|
| async def save_temp_file(upload_file): |
| |
| content_type = upload_file.content_type or "" |
| filename = upload_file.filename |
| |
| |
| 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" |
| |
| |
| 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 original_suffix == ".wav": |
| return original_path |
| |
| |
| try: |
| wav_path = original_path.replace(original_suffix, ".wav") |
| |
| |
| 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()}") |
| |
| |
| os.remove(original_path) |
| |
| return wav_path |
| except Exception as e: |
| |
| print(f"Audio conversion error: {e}") |
| |
| if os.path.exists(original_path): |
| os.remove(original_path) |
| raise Exception(f"Failed to convert audio file: {e}") |
