fix: prefer record url for audio downloads

astrbot/core/message/components.py

@@ -28,6 +28,8 @@
 import sys
 import uuid
 from enum import Enum
+from pathlib import Path
+from urllib.parse import unquote, urlparse
 
 if sys.version_info >= (3, 14):
     from pydantic import BaseModel
@@ -133,38 +135,60 @@ def fromFileSystem(path, **_):
     def fromURL(url: str, **_):
         if url.startswith("http://") or url.startswith("https://"):
             return Record(file=url, **_)
-        raise Exception("not a valid url")
+        raise ValueError("not a valid url")
 
     @staticmethod
     def fromBase64(bs64_data: str, **_):
         return Record(file=f"base64://{bs64_data}", **_)
 
+    @staticmethod
+    def _get_audio_suffix(url: str) -> str:
+        suffix = Path(unquote(urlparse(url).path)).suffix
+        return suffix or ".amr"
+
+    def _resolve_audio_source(self) -> str:
+        source = self.url or self.file
+        if not source:
+            raise ValueError("No valid file or URL provided")
+        return source
+
+    async def _download_audio_url(self, url: str) -> str:
+        temp_dir = Path(get_astrbot_temp_path())
+        temp_dir.mkdir(parents=True, exist_ok=True)
+        file_path = (
+            temp_dir / f"recordseg_{uuid.uuid4().hex}{self._get_audio_suffix(url)}"
+        )
+        await download_file(url, str(file_path))
+        if file_path.exists():
+            return str(file_path.resolve())
+        raise RuntimeError(f"download failed: {url}")
+
+    def _write_base64_audio_to_file(self, url: str) -> str:
+        bs64_data = url.removeprefix("base64://")
+        audio_bytes = base64.b64decode(bs64_data)
+        temp_dir = Path(get_astrbot_temp_path())
+        temp_dir.mkdir(parents=True, exist_ok=True)
+        file_path = temp_dir / f"recordseg_{uuid.uuid4().hex}.amr"
+        file_path.write_bytes(audio_bytes)
+        return str(file_path.resolve())
+
     async def convert_to_file_path(self) -> str:
         """将这个语音统一转换为本地文件路径。这个方法避免了手动判断语音数据类型，直接返回语音数据的本地路径（如果是网络 URL, 则会自动进行下载）。
 
         Returns:
             str: 语音的本地路径，以绝对路径表示。
 
         """
-        if not self.file:
-            raise Exception(f"not a valid file: {self.file}")
-        if self.file.startswith("file:///"):
-            return self.file[8:]
-        if self.file.startswith("http"):
-            file_path = await download_image_by_url(self.file)
-            return os.path.abspath(file_path)
-        if self.file.startswith("base64://"):
-            bs64_data = self.file.removeprefix("base64://")
-            image_bytes = base64.b64decode(bs64_data)
-            file_path = os.path.join(
-                get_astrbot_temp_path(), f"recordseg_{uuid.uuid4()}.jpg"
-            )
-            with open(file_path, "wb") as f:
-                f.write(image_bytes)
-            return os.path.abspath(file_path)
-        if os.path.exists(self.file):
-            return os.path.abspath(self.file)
-        raise Exception(f"not a valid file: {self.file}")
+        url = self._resolve_audio_source()
+        if url.startswith("file:///"):
+            return url[8:]
+        if url.startswith("http"):
+            return await self._download_audio_url(url)
+        if url.startswith("base64://"):
+            return self._write_base64_audio_to_file(url)
+        if os.path.exists(url):
+            return os.path.abspath(url)
+        raise FileNotFoundError(f"not a valid file: {url}")
 
     async def convert_to_base64(self) -> str:
         """将语音统一转换为 base64 编码。这个方法避免了手动判断语音数据类型，直接返回语音数据的 base64 编码。
@@ -174,19 +198,18 @@ async def convert_to_base64(self) -> str:
 
         """
         # convert to base64
-        if not self.file:
-            raise Exception(f"not a valid file: {self.file}")
-        if self.file.startswith("file:///"):
-            bs64_data = file_to_base64(self.file[8:])
-        elif self.file.startswith("http"):
-            file_path = await download_image_by_url(self.file)
+        url = self._resolve_audio_source()
+        if url.startswith("file:///"):
+            bs64_data = file_to_base64(url[8:])
+        elif url.startswith("http"):
+            file_path = await self._download_audio_url(url)
             bs64_data = file_to_base64(file_path)
-        elif self.file.startswith("base64://"):
-            bs64_data = self.file
-        elif os.path.exists(self.file):
-            bs64_data = file_to_base64(self.file)
+        elif url.startswith("base64://"):
+            bs64_data = url
+        elif os.path.exists(url):
+            bs64_data = file_to_base64(url)
         else:
-            raise Exception(f"not a valid file: {self.file}")
+            raise FileNotFoundError(f"not a valid file: {url}")
         bs64_data = bs64_data.removeprefix("base64://")
         return bs64_data
 

tests/unit/test_message_record_component.py

@@ -0,0 +1,81 @@
+import base64
+from pathlib import Path
+
+import pytest
+
+import astrbot.core.message.components as components
+from astrbot.core.message.components import Record
+
+
+@pytest.mark.asyncio
+async def test_record_convert_to_file_path_prefers_url_when_file_is_name(
+    monkeypatch,
+    tmp_path,
+):
+    calls: list[tuple[str, Path]] = []
+    audio_bytes = b"audio-content"
+    audio_url = "http://napcat.local/nt_data/Ptt/2026-04/Ori/voice.amr"
+
+    async def fake_download_file(url: str, path: str, *_, **__) -> None:
+        target = Path(path)
+        calls.append((url, target))
+        target.write_bytes(audio_bytes)
+
+    monkeypatch.setattr(components, "download_file", fake_download_file)
+    monkeypatch.setattr(components, "get_astrbot_temp_path", lambda: str(tmp_path))
+
+    record = Record(file="voice.amr", url=audio_url)
+
+    file_path = Path(await record.convert_to_file_path())
+
+    assert file_path.read_bytes() == audio_bytes
+    assert file_path.suffix == ".amr"
+    assert calls == [(audio_url, file_path)]
+
+
+@pytest.mark.asyncio
+async def test_record_convert_to_base64_prefers_url_when_file_is_name(
+    monkeypatch,
+    tmp_path,
+):
+    audio_bytes = b"audio-content"
+    audio_url = "http://napcat.local/nt_data/Ptt/2026-04/Ori/voice.amr"
+
+    async def fake_download_file(url: str, path: str, *_, **__) -> None:
+        assert url == audio_url
+        Path(path).write_bytes(audio_bytes)
+
+    monkeypatch.setattr(components, "download_file", fake_download_file)
+    monkeypatch.setattr(components, "get_astrbot_temp_path", lambda: str(tmp_path))
+
+    record = Record(file="voice.amr", url=audio_url)
+
+    assert await record.convert_to_base64() == base64.b64encode(audio_bytes).decode()
+
+
+@pytest.mark.asyncio
+async def test_record_convert_to_file_path_prefers_base64_url_when_file_is_name(
+    monkeypatch,
+    tmp_path,
+):
+    audio_bytes = b"audio-content"
+    audio_url = f"base64://{base64.b64encode(audio_bytes).decode()}"
+
+    monkeypatch.setattr(components, "get_astrbot_temp_path", lambda: str(tmp_path))
+
+    record = Record(file="voice.amr", url=audio_url)
+
+    file_path = Path(await record.convert_to_file_path())
+
+    assert file_path.read_bytes() == audio_bytes
+    assert file_path.suffix == ".amr"
+
+
+@pytest.mark.asyncio
+async def test_record_convert_to_base64_prefers_base64_url_when_file_is_name():
+    audio_bytes = b"audio-content"
+    audio_base64 = base64.b64encode(audio_bytes).decode()
+
+    record = Record(file="voice.amr", url=f"base64://{audio_base64}")
+
+    assert await record.convert_to_base64() == audio_base64