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
51 changes: 49 additions & 2 deletions astrbot/core/utils/media_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -295,14 +295,57 @@ async def ensure_wav(audio_path: str, output_path: str | None = None) -> str:
"""Ensure the audio path points to wav format by extension/guess and convert when needed.

If the file appears to already be wav, return it directly to avoid extra conversion.
Handles Tencent QQ special formats (silk / amr) that ffmpeg cannot decode.
"""

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 in ("silk",):
# Tencent Silk format (commonly used by QQ). ffmpeg cannot decode it.
from astrbot.core.utils.tencent_record_helper import tencent_silk_to_wav

if not output_path:
from pathlib import Path
from uuid import uuid4

from astrbot.core.utils.astrbot_path import get_astrbot_temp_path

temp_dir = Path(get_astrbot_temp_path())
temp_dir.mkdir(parents=True, exist_ok=True)
output_path = str(temp_dir / f"media_audio_{uuid4().hex}.wav")
Comment on lines +313 to +321
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.

suggestion: Consider extracting the repeated temp WAV output_path creation into a small helper to avoid duplication.

This temp WAV path construction appears in both the Silk and AMR branches. A small helper (e.g. _make_temp_wav_path()) would remove duplication and keep future changes to naming or directory logic centralized.

Suggested implementation:

    def _make_temp_wav_path() -> str:
        from pathlib import Path
        from uuid import uuid4

        from astrbot.core.utils.astrbot_path import get_astrbot_temp_path

        temp_dir = Path(get_astrbot_temp_path())
        temp_dir.mkdir(parents=True, exist_ok=True)
        return str(temp_dir / f"media_audio_{uuid4().hex}.wav")

    audio_type = _get_audio_magic_type(audio_path)
        if not output_path:
            output_path = _make_temp_wav_path()

In the AMR branch (the if audio_type in ("amr",): block), replace any duplicated logic that constructs a temporary WAV output_path (likely the same Path(get_astrbot_temp_path()) + uuid4 pattern) with:

        if not output_path:
            output_path = _make_temp_wav_path()

This will ensure both Silk and AMR conversions share the same centralized temp WAV path creation logic.


logger.info(f"Detected Silk audio format, converting to wav: {audio_path}")
return await tencent_silk_to_wav(audio_path, output_path)

if audio_type in ("amr",):
# AMR from Tencent platforms may also be a variant that ffmpeg misdetects.
# Try ffmpeg first as it handles standard AMR correctly.
try:
return await convert_audio_to_wav(audio_path, output_path)
except Exception as e:
Comment on lines +329 to +331
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.

issue (bug_risk): Catching a broad Exception here may hide unexpected problems; consider narrowing the exception type or re-raising.

This bare Exception catch will also hide programmer errors (e.g. TypeError, ValueError) from convert_audio_to_wav. Prefer catching the specific error type that represents ffmpeg failures and let other exceptions propagate, potentially re-raising after logging.

logger.warning(
f"ffmpeg failed to convert amr file, trying pyffmpeg fallback: {e}"
)
from astrbot.core.utils.tencent_record_helper import convert_to_pcm_wav

if not output_path:
from pathlib import Path
from uuid import uuid4

from astrbot.core.utils.astrbot_path import get_astrbot_temp_path

temp_dir = Path(get_astrbot_temp_path())
temp_dir.mkdir(parents=True, exist_ok=True)
output_path = str(temp_dir / f"media_audio_{uuid4().hex}.wav")

return await convert_to_pcm_wav(audio_path, output_path)

return await convert_audio_to_wav(audio_path, output_path)
Comment on lines +304 to 349
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 in both the silk and amr blocks. Additionally, several imports (Path, uuid, get_astrbot_temp_path) are redundant as they are already available at the module level. Refactoring the path generation to the top of the conversion logic simplifies the function and avoids code duplication. Furthermore, as this logic handles media attachments, please ensure it is accompanied by corresponding unit tests.

    audio_type = _get_audio_magic_type(audio_path)

    if audio_type == "wav":
        return audio_path

    # If not wav, we will need to convert. Ensure output_path is set.
    if not output_path:
        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":
        # Tencent Silk format (commonly used by QQ). ffmpeg cannot decode it.
        from astrbot.core.utils.tencent_record_helper import tencent_silk_to_wav
        logger.info(f"Detected Silk audio format, converting to wav: {audio_path}")
        return await tencent_silk_to_wav(audio_path, output_path)

    if audio_type == "amr":
        # AMR from Tencent platforms may also be a variant that ffmpeg misdetects.
        # Try ffmpeg first as it handles standard AMR correctly.
        try:
            return await convert_audio_to_wav(audio_path, output_path)
        except Exception as e:
            logger.warning(
                f"ffmpeg failed to convert amr file, trying pyffmpeg fallback: {e}"
            )
            from astrbot.core.utils.tencent_record_helper import convert_to_pcm_wav
            return await convert_to_pcm_wav(audio_path, output_path)

    return await convert_audio_to_wav(audio_path, output_path)
References
  1. When implementing similar functionality for different cases, 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 +384,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[:8] == b"#!SILK_V3"
or header[1:9] == b"#!SILK_V3"
or b"SILK" in header[:16]
):
return "silk"

return ""
Expand Down
Loading