diff --git a/astrbot/core/utils/media_utils.py b/astrbot/core/utils/media_utils.py index 1496206832..391ee3c770 100644 --- a/astrbot/core/utils/media_utils.py +++ b/astrbot/core/utils/media_utils.py @@ -1135,10 +1135,17 @@ async def convert_audio_format( Exception: Raised when ffmpeg is unavailable or conversion fails. """ source_path = Path(audio_path) - if source_path.suffix.lower() == f".{output_format}" and ( - not source_path.exists() or _get_audio_magic_type(audio_path) == output_format - ): - return audio_path + if source_path.suffix.lower() == f".{output_format}": + # The magic-byte probe performs synchronous file I/O. Keep it off the + # event loop because this helper is called while processing messages. + if not source_path.exists(): + return audio_path + + detected_format = await asyncio.to_thread(_get_audio_magic_type, audio_path) + if detected_format == output_format or ( + output_format == "ogg" and detected_format == "opus" + ): + return audio_path if output_path is None: temp_dir = Path(get_astrbot_temp_path()) diff --git a/tests/test_media_utils.py b/tests/test_media_utils.py index 976efc6db3..cba3f958a3 100644 --- a/tests/test_media_utils.py +++ b/tests/test_media_utils.py @@ -504,6 +504,51 @@ async def test_convert_audio_format_keeps_missing_target_path(): assert result == missing_path +@pytest.mark.asyncio +async def test_convert_audio_format_offloads_magic_byte_probe(tmp_path, monkeypatch): + source_path = tmp_path / "voice.wav" + source_path.write_bytes(b"RIFF\x24\x00\x00\x00WAVEfmt " + b"\x00" * 16) + probe_calls = [] + + async def fake_to_thread(func, *args): + probe_calls.append((func, args)) + return "wav" + + monkeypatch.setattr(media_utils.asyncio, "to_thread", fake_to_thread) + + result = await media_utils.convert_audio_format( + str(source_path), + output_format="wav", + ) + + assert result == str(source_path) + assert probe_calls == [(media_utils._get_audio_magic_type, (str(source_path),))] + + +@pytest.mark.asyncio +async def test_convert_audio_format_keeps_ogg_opus_without_reencoding( + tmp_path, monkeypatch +): + source_path = tmp_path / "voice.ogg" + source_path.write_bytes(b"OggS" + b"\x00" * 20 + b"OpusHead" + b"\x00" * 32) + + async def fail_create_subprocess_exec(*args, **kwargs): + raise AssertionError("an Ogg/Opus source should not be re-encoded") + + monkeypatch.setattr( + media_utils.asyncio, + "create_subprocess_exec", + fail_create_subprocess_exec, + ) + + result = await media_utils.convert_audio_format( + str(source_path), + output_format="ogg", + ) + + assert result == str(source_path) + + @pytest.mark.asyncio async def test_media_resolver_cleans_http_target_when_download_fails( tmp_path, monkeypatch