| """AMR/任意音频 -> 8kHz 单声道 PCM s16le 转码。 |
| |
| 使用系统 ffmpeg（Dockerfile 已 apt-get install ffmpeg）。 |
| """ |
| from __future__ import annotations |
|
|
| import asyncio |
| import logging |
| from typing import Optional |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| async def decode_to_pcm_s16le( |
| audio_bytes: bytes, |
| *, |
| duration_sec: int = 3, |
| sample_rate: int = 8000, |
| channels: int = 1, |
| mime: str = "", |
| ) -> bytes: |
| """把任意音频（amr/mp3/wav/m4a 等）转成 8kHz 单声道 PCM s16le。 |
| |
| Args: |
| audio_bytes: 输入音频字节 |
| duration_sec: 截取时长（秒），默认 3 |
| sample_rate: 采样率，默认 8000（网易云 shazam_v2 要求） |
| channels: 声道数，默认 1 |
| mime: MIME 类型。若为 audio/pcm 则视为已是 PCM s16le 直接截断返回 |
| |
| Returns: |
| PCM s16le 字节流 |
| |
| Raises: |
| RuntimeError: ffmpeg 失败 |
| """ |
| |
| if mime in ("audio/pcm", "audio/x-pcm", "application/pcm"): |
| needed = sample_rate * channels * 2 * duration_sec |
| return audio_bytes[:needed] |
|
|
| args = [ |
| "ffmpeg", |
| "-i", "pipe:0", |
| "-f", "s16le", |
| "-acodec", "pcm_s16le", |
| "-ar", str(sample_rate), |
| "-ac", str(channels), |
| "-t", str(duration_sec), |
| "pipe:1", |
| ] |
| logger.info("[amr-decoder] ffmpeg start, input=%d bytes mime=%s", len(audio_bytes), mime) |
|
|
| proc = await asyncio.create_subprocess_exec( |
| *args, |
| stdin=asyncio.subprocess.PIPE, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.PIPE, |
| ) |
| stdout, stderr = await proc.communicate(input=audio_bytes) |
| if proc.returncode != 0: |
| err = stderr.decode("utf-8", errors="replace")[-500:] |
| raise RuntimeError(f"ffmpeg exit {proc.returncode}: {err}") |
| logger.info("[amr-decoder] ffmpeg ok, pcm=%d bytes", len(stdout)) |
| return stdout |
|
|
