Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 15 additions & 2 deletions astrbot/core/utils/media_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

from astrbot import logger
from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
from astrbot.core.utils.tencent_record_helper import tencent_silk_to_wav

IMAGE_COMPRESS_DEFAULT_MAX_SIZE = 1280
IMAGE_COMPRESS_DEFAULT_QUALITY = 95
Expand Down Expand Up @@ -300,9 +301,17 @@ async def ensure_wav(audio_path: str, output_path: str | None = None) -> str:
if not audio_path:
return audio_path

if _get_audio_magic_type(audio_path) == "wav":
audio_type = _get_audio_magic_type(audio_path)
if audio_type == "wav":
return audio_path

if audio_type == "silk":
if output_path is None:
temp_dir = get_astrbot_temp_path()
os.makedirs(temp_dir, exist_ok=True)
output_path = os.path.join(temp_dir, f"media_audio_{uuid.uuid4().hex}.wav")
return await tencent_silk_to_wav(audio_path, output_path)

return await convert_audio_to_wav(audio_path, output_path)
Comment on lines +308 to 315
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The logic for generating a temporary output path is duplicated here and in convert_audio_format. Using pathlib.Path for consistency and refactoring the path generation to the top of the conversion logic makes the function cleaner and ensures output_path is available for both silk and other formats, adhering to the rule of avoiding code duplication. Furthermore, please ensure this new attachment handling functionality is covered by unit tests.

Suggested change
if audio_type == "silk":
if output_path is None:
temp_dir = get_astrbot_temp_path()
os.makedirs(temp_dir, exist_ok=True)
output_path = os.path.join(temp_dir, f"media_audio_{uuid.uuid4().hex}.wav")
return await tencent_silk_to_wav(audio_path, output_path)
return await convert_audio_to_wav(audio_path, output_path)
if output_path is None:
temp_dir = Path(get_astrbot_temp_path())
temp_dir.mkdir(parents=True, exist_ok=True)
output_path = str(temp_dir / f"media_audio_{uuid.uuid4().hex}.wav")
if audio_type == "silk":
return await tencent_silk_to_wav(audio_path, output_path)
return await convert_audio_to_wav(audio_path, output_path)
References
  1. When implementing similar functionality for different cases (e.g., direct vs. quoted attachments), refactor the logic into a shared helper function to avoid code duplication.
  2. New functionality, such as handling attachments, should be accompanied by corresponding unit tests.



Expand Down Expand Up @@ -341,7 +350,11 @@ def _get_audio_magic_type(audio_path: str) -> str:
if header[:4] == b"ftyp" and b"mp4" in header[:8]:
return "mp4"

if header[:8] == b"#!SILK_V3":
if header.startswith(b"#!SILK_V3"):
return "silk"

# Tencent SILK: leading \x02 byte before #!SILK_V3
if header.startswith(b"\x02#!SILK_V3"):
return "silk"

return ""
Expand Down