File size: 1,987 Bytes
d731d8e 6d61ecf d731d8e 6d61ecf d731d8e 6d61ecf d731d8e 6d61ecf d731d8e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """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 失败
"""
# 裸 PCM 直通（便于测试 / 手表若直接录 PCM 也支持）
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
|