如何在Python中将audio/webm;codecs=opus转为Amazon Lex兼容的audio/l16格式

针对Firefox下录制的audio/webm;codecs=opus格式音频，需要转换为Lex要求的audio/l16; rate=16000; channels=1（16位小端单声道PCM），以下是前端JavaScript和后端Python（Lambda）两种实现方案：

一、前端JavaScript转换（推荐） #

直接在前端完成格式转换后再上传，无需后端额外处理，核心利用Web Audio API实现：

代码实现 #

async function convertToLexAudioFormat(webmBlob) {
  // 初始化音频上下文，指定目标采样率16kHz
  const audioContext = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: 16000 });
  
  // 读取Blob并解码音频数据
  const arrayBuffer = await webmBlob.arrayBuffer();
  const audioBuffer = await audioContext.decodeAudioData(arrayBuffer);
  
  // 转换为单声道
  const monoBuffer = audioContext.createBuffer(1, audioBuffer.length, 16000);
  const sourceChannel = audioBuffer.getChannelData(0);
  const targetChannel = monoBuffer.getChannelData(0);
  targetChannel.set(sourceChannel);
  
  // 转为16位小端PCM数据
  const pcmInt16 = new Int16Array(monoBuffer.length);
  for (let i = 0; i < monoBuffer.length; i++) {
    // 限制采样值范围在[-1, 1]，再映射为16位整数
    const sample = Math.max(-1, Math.min(1, targetChannel[i]));
    pcmInt16[i] = sample < 0 ? sample * 0x8000 : sample * 0x7FFF;
  }
  
  // 转换为Base64字符串
  const uint8Array = new Uint8Array(pcmInt16.buffer);
  const base64 = btoa(String.fromCharCode(...uint8Array));
  return base64;
}

使用方式 #

录制完成获取到WebM格式的Blob后，调用上述函数得到符合要求的Base64字符串，直接发送给Lambda即可，Lambda无需再处理格式转换。

二、后端Python（Lambda）转换 #

若需在Lambda中处理格式转换，需使用纯Python依赖库（无需ffmpeg），核心步骤为解析WebM提取Opus帧、解码Opus为PCM、重采样至16kHz。

前置准备：创建Lambda层 #

Lambda需要依赖opuslib（解码Opus）和webm（解析WebM），需打包为Lambda层：

1. 在Linux环境或Docker中执行：

   mkdir python
   pip install opuslib webm -t python/
   zip -r audio-layers.zip python/
   

2. 将audio-layers.zip上传至Lambda层，并关联到目标函数。

代码实现 #

import base64
import uuid
import boto3
from opuslib import Decoder
import webm
import audioop

# 初始化Lex客户端
client = boto3.client('lexv2-runtime')

def decode_webm_to_opus_pcm(webm_bytes):
    """解析WebM文件，提取并解码Opus帧为48kHz 16位单声道PCM"""
    parser = webm.WebM()
    opus_frames = []
    for frame in parser.decode(webm_bytes):
        if frame['track_type'] == 'audio':
            opus_frames.append(frame['data'])
    
    # Opus解码器默认48kHz采样率，单声道
    decoder = Decoder(48000, 1)
    raw_pcm = b''
    for frame in opus_frames:
        raw_pcm += decoder.decode(frame)
    return raw_pcm

def resample_pcm_to_16k(pcm_data):
    """将48kHz PCM重采样为16kHz 16位单声道PCM"""
    # audioop.ratecv参数：数据、位宽、声道数、原采样率、目标采样率、状态
    resampled_pcm, _ = audioop.ratecv(pcm_data, 2, 1, 48000, 16000, None)
    return resampled_pcm

def get_lex(recording):
    try:
        response = client.recognize_utterance(
                botId='YOUR_BOT_ID',
                botAliasId='YOUR_BOT_ALIAS_ID',
                localeId='en_US',
                sessionId=str(uuid.uuid4()),
                requestContentType='audio/l16; rate=16000; channels=1',
                responseContentType='text/plain; charset=utf-8',
                inputStream=recording
            )
        # 替换为你已实现的interpretations解析逻辑
        return response['interpretations']
    except Exception as e:
        print(e)
        return {"error": str(e)}

def lambda_handler(event, context):
    # 解码前端传来的Base64音频
    decoded_webm = base64.b64decode(event['body'])
    
    # 1. 解析WebM并解码Opus为PCM
    pcm_48k = decode_webm_to_opus_pcm(decoded_webm)
    
    # 2. 重采样至16kHz
    pcm_16k = resample_pcm_to_16k(pcm_48k)
    
    # 3. 调用Lex服务
    lex_response = get_lex(pcm_16k)
    
    return {"statusCode": 200, "body": lex_response}

内容的提问来源于stack exchange，提问作者Leeroy Hannigan