无法将Twilio媒体流保存为有效MP3文件的技术求助

核心问题 #

Twilio媒体流传输的是无文件头的8kHz单声道μ-law编码PCM音频，你的代码未正确匹配该格式，直接用默认参数处理导致音频文件无效。以下是基于pygame的修复方案：

修复后的完整代码 #

import base64
import json
import pygame
import wave
from flask import Flask, request
from flask_sock import Sock, ConnectionClosed
from twilio.twiml.voice_response import VoiceResponse, Start
from twilio.rest import Client
import io

app = Flask(__name__)
sock = Sock(app)
twilio_client = Client()

# Twilio音频流固定参数：8kHz采样率，单声道，μ-law编码
TWILIO_AUDIO_PARAMS = {
    "sample_rate": 8000,
    "channels": 1,
    "sample_width": 1,  # μ-law为8位样本，对应1字节
    "compression_type": b'ULAW'
}

@app.route('/call', methods=['POST'])
def call():
    """接收来电"""
    response = VoiceResponse()
    start = Start()
    start.stream(url=f'wss://{request.host}/stream')
    response.append(start)
    response.say('Please leave a message')
    response.pause(length=60)
    print(f'来电号码：{request.form["From"]}')
    return str(response), 200, {'Content-Type': 'text/xml'}

@sock.route('/stream')
def stream(ws):
    """接收并处理音频流"""
    audio_data = b''  # 存储原始μ-law音频数据

    while True:
        try:
            message = ws.receive()
        except ConnectionClosed:
            print("连接意外中断")
            break
            
        packet = json.loads(message)

        if packet['event'] == 'start':
            print('开始音频流传输')
        elif packet['event'] == 'media':
            audio_payload = base64.b64decode(packet['media']['payload'])
            audio_data += audio_payload
            play_audio_live(audio_payload)
        elif packet['event'] == 'stop':
            print('\n音频流传输结束')

            if audio_data:
                save_audio_as_mp3(audio_data, "audio.mp3")
                print("音频已保存为audio.mp3")
            else:
                print("未接收到有效音频数据")
            break

def play_audio_live(audio_payload):
    """用pygame实时播放μ-law音频"""
    # 仅初始化一次mixer，避免重复初始化报错
    if not pygame.mixer.get_init():
        pygame.mixer.pre_init(
            frequency=TWILIO_AUDIO_PARAMS["sample_rate"],
            size=8,  # 匹配μ-law的8位样本格式
            channels=TWILIO_AUDIO_PARAMS["channels"],
            allowedchanges=0
        )
        pygame.mixer.init()
    
    # 加载并播放音频块
    audio_chunk = pygame.mixer.Sound(buffer=audio_payload)
    audio_chunk.play()

def save_audio_as_mp3(audio_data, mp3_filename):
    """将μ-law音频转为MP3（无需ffmpeg，使用纯Python工具）"""
    # 第一步：给原始μ-law数据添加WAV文件头，生成标准可识别的音频文件
    wav_buffer = io.BytesIO()
    with wave.open(wav_buffer, 'wb') as wav_file:
        wav_file.setnchannels(TWILIO_AUDIO_PARAMS["channels"])
        wav_file.setsampwidth(TWILIO_AUDIO_PARAMS["sample_width"])
        wav_file.setframerate(TWILIO_AUDIO_PARAMS["sample_rate"])
        wav_file.setcomptype(TWILIO_AUDIO_PARAMS["compression_type"], "μ-law")
        wav_file.writeframes(audio_data)
    
    wav_buffer.seek(0)
    
    # 第二步：用pygame解码WAV为16位PCM，再用纯Python编码器转为MP3
    # 需要先安装lameenc：pip install lameenc
    import lameenc
    encoder = lameenc.Encoder()
    encoder.set_bit_rate(64)
    encoder.set_sample_rate(TWILIO_AUDIO_PARAMS["sample_rate"])
    encoder.set_channels(TWILIO_AUDIO_PARAMS["channels"])
    encoder.set_quality(2)  # 0=音质最好，9=编码最快
    
    pygame.mixer.init(
        frequency=TWILIO_AUDIO_PARAMS["sample_rate"],
        size=-16,
        channels=TWILIO_AUDIO_PARAMS["channels"]
    )
    sound = pygame.mixer.Sound(wav_buffer)
    pcm_data = sound.get_raw()
    
    # 编码并保存MP3
    mp3_data = encoder.encode(pcm_data)
    mp3_data += encoder.flush()
    with open(mp3_filename, 'wb') as f:
        f.write(mp3_data)

if __name__ == '__main__':
    from pyngrok import ngrok
    port = 5000
    public_url = ngrok.connect(port, bind_tls=True).public_url
    number = twilio_client.incoming_phone_numbers.list()[0]
    number.update(voice_url=public_url + '/call')
    print(f'等待来电：{number.phone_number}')

    app.run(port=port)

关键修复点 #

1. 匹配Twilio音频格式：明确指定pygame mixer使用8kHz采样率、8位样本、单声道，完全对应Twilio的μ-law编码标准
2. 添加WAV容器头：用wave模块给原始无格式的μ-law数据加上标准WAV头，让音频工具能正确识别
3. 纯Python转MP3：使用lameenc（无需ffmpeg的纯Python库）将解码后的PCM数据转为MP3，满足不能使用ffmpeg的限制
4. 代码错误修复：移除save_audio_as_mp3函数末尾的乱码，添加连接异常处理逻辑

注意事项 #

• 需提前安装lameenc：pip install lameenc，这是纯Python实现的MP3编码器，无外部依赖
• 建议使用pygame 2.x版本，避免版本兼容问题
• 实时播放时，pygame mixer仅初始化一次，避免重复初始化导致的音频异常

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