diff --git a/astrbot/core/utils/media_utils.py b/astrbot/core/utils/media_utils.py index 276a66b453..6cd58b92ec 100644 --- a/astrbot/core/utils/media_utils.py +++ b/astrbot/core/utils/media_utils.py @@ -1127,7 +1127,10 @@ async def convert_audio_format( Raises: Exception: Raised when ffmpeg is unavailable or conversion fails. """ - if audio_path.lower().endswith(f".{output_format}"): + 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 output_path is None: diff --git a/tests/test_media_utils.py b/tests/test_media_utils.py index 0cc3c85a12..efe5e65f02 100644 --- a/tests/test_media_utils.py +++ b/tests/test_media_utils.py @@ -419,6 +419,91 @@ async def fail_ensure_wav(*args, **kwargs): assert not list(tmp_path.iterdir()) +@pytest.mark.asyncio +async def test_convert_audio_format_transcodes_amr_bytes_misnamed_as_wav( + tmp_path, monkeypatch +): + source_path = tmp_path / "voice.wav" + source_path.write_bytes(b"#!AMR\n" + b"\x00" * 32) + converted_path = tmp_path / "converted.wav" + calls = [] + + class FakeProcess: + returncode = 0 + + async def communicate(self): + converted_path.write_bytes(b"RIFF\x24\x00\x00\x00WAVEfmt " + b"\x00" * 16) + return b"", b"" + + async def fake_create_subprocess_exec(*args, **kwargs): + calls.append((args, kwargs)) + return FakeProcess() + + monkeypatch.setattr( + media_utils.asyncio, + "create_subprocess_exec", + fake_create_subprocess_exec, + ) + + result = await media_utils.convert_audio_format( + str(source_path), + output_format="wav", + output_path=str(converted_path), + ) + + assert result == str(converted_path) + assert calls + assert calls[0][0][:4] == ("ffmpeg", "-y", "-i", str(source_path)) + + +@pytest.mark.asyncio +async def test_convert_audio_format_rewrites_misnamed_target_format( + tmp_path, monkeypatch +): + source_path = tmp_path / "voice.wav" + source_path.write_bytes(b"#!AMR\n" + b"\x00" * 32) + converted_path = tmp_path / "converted.amr" + calls = [] + + class FakeProcess: + returncode = 0 + + async def communicate(self): + converted_path.write_bytes(b"#!AMR\n" + b"\x00" * 32) + return b"", b"" + + async def fake_create_subprocess_exec(*args, **kwargs): + calls.append((args, kwargs)) + return FakeProcess() + + monkeypatch.setattr( + media_utils.asyncio, + "create_subprocess_exec", + fake_create_subprocess_exec, + ) + + result = await media_utils.convert_audio_format( + str(source_path), + output_format="amr", + output_path=str(converted_path), + ) + + assert result == str(converted_path) + assert calls + + +@pytest.mark.asyncio +async def test_convert_audio_format_keeps_missing_target_path(): + missing_path = "missing.wav" + + result = await media_utils.convert_audio_format( + missing_path, + output_format="wav", + ) + + assert result == missing_path + + @pytest.mark.asyncio async def test_media_resolver_cleans_http_target_when_download_fails( tmp_path, monkeypatch