66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
"""入站语音解码:任意容器/编码(AMR/SILK 外的常见格式)→ 16k/16bit/mono PCM。
|
|
|
|
微信系渠道的语音格式:
|
|
- 企业微信 `media/get` 下发 AMR(AMR-NB 8kHz,`#!AMR\\n` 头)—— ffmpeg 原生解码器可解,
|
|
无需 opencore 编译选项;
|
|
- 个人微信 ClawBot 预计是 SILK v3(待真机探明,届时若 ffmpeg 解不了再引 pilk)。
|
|
|
|
实现:ffmpeg 子进程,stdin 喂原始字节(自动探测容器)→ stdout 收 s16le/16k/mono 裸 PCM,
|
|
不落临时文件。ffmpeg 是**系统依赖**(Linux `apt install ffmpeg`;Windows 开发机 winget 装
|
|
或设 FFMPEG_PATH 指向可执行文件),未安装时抛 AudioNotConfigured,上层给可读提示。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
|
|
_TIMEOUT_S = 30 # 60s 语音的解码远快于实时,30s 兜底防挂死
|
|
|
|
|
|
class AudioConvertError(RuntimeError):
|
|
"""音频解码失败(格式不识别 / ffmpeg 报错)。"""
|
|
|
|
|
|
class AudioNotConfigured(AudioConvertError):
|
|
"""ffmpeg 不可用 —— 部署缺系统依赖,提示运维安装。"""
|
|
|
|
|
|
def find_ffmpeg() -> str:
|
|
"""定位 ffmpeg 可执行文件:env FFMPEG_PATH 优先,其次 PATH。找不到抛 AudioNotConfigured。"""
|
|
cand = (os.getenv("FFMPEG_PATH") or "").strip() or shutil.which("ffmpeg")
|
|
if not cand or not (os.path.isfile(cand) or shutil.which(cand)):
|
|
raise AudioNotConfigured(
|
|
"语音解码不可用:未找到 ffmpeg(Linux 服务器 `apt install ffmpeg`;"
|
|
"或设 FFMPEG_PATH 指向 ffmpeg 可执行文件)"
|
|
)
|
|
return cand
|
|
|
|
|
|
def to_pcm16k(data: bytes) -> bytes:
|
|
"""音频字节(AMR/MP3/WAV/OGG 等 ffmpeg 认识的格式)→ 16kHz/16bit/单声道小端 PCM。
|
|
|
|
容器格式由 ffmpeg 从字节头自动探测,无需调用方声明。空输入返回 b""。
|
|
"""
|
|
if not data:
|
|
return b""
|
|
ffmpeg = find_ffmpeg()
|
|
cmd = [
|
|
ffmpeg, "-hide_banner", "-loglevel", "error",
|
|
"-i", "pipe:0",
|
|
"-f", "s16le", "-ar", "16000", "-ac", "1",
|
|
"pipe:1",
|
|
]
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd, input=data, capture_output=True, timeout=_TIMEOUT_S
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
raise AudioConvertError(f"音频解码超时(>{_TIMEOUT_S}s)")
|
|
except OSError as e:
|
|
raise AudioNotConfigured(f"ffmpeg 启动失败:{type(e).__name__}: {e}")
|
|
if proc.returncode != 0 or not proc.stdout:
|
|
err = (proc.stderr or b"").decode("utf-8", errors="replace").strip()
|
|
raise AudioConvertError(f"音频解码失败:{err[:300] or '无输出'}")
|
|
return proc.stdout
|