如何在Django中当Twilio通话结束时停止sounddevice音频录制

咱们先拆解下你当前代码的核心问题：你用的sounddevice.rec()是固定时长的阻塞式录制，它会硬等30秒才停止，完全感知不到Twilio通话何时结束。要实现通话结束就立刻停录，得改成事件驱动的录制逻辑，同时结合Twilio的通话状态回调来触发停止操作。

核心思路 #

1. 抛弃固定时长的阻塞录制，改用非阻塞的流录制（sounddevice.InputStream），让录制过程可以随时被中断
2. 利用Twilio的通话状态回调（Status Callback），当通话状态变为completed时，通知Django后端停止对应录制
3. 由于Django是请求响应模型，需要用共享状态（比如字典或数据库）来控制单条通话的录制启停（简单场景先用字典，多进程部署再换Redis）

代码实现步骤 #

第一步：重构录制逻辑为可启停的非阻塞流 #

把原来的固定时长录制改成能随时开关的流录制，用通话ID来区分不同的录制任务：

import sounddevice as sd
import numpy as np
from scipy.io.wavfile import write
from django.http import HttpResponse
import threading

# 用字典存储各通话的录制状态，key为Twilio的CallSid，value是录制控制参数
recording_tasks = {}
fs = 44100

def audio_callback(indata, frames, time, status, call_sid):
    """音频数据回调，实时保存帧数据"""
    task = recording_tasks.get(call_sid)
    if task and task['is_recording']:
        task['frames'].append(indata.copy())

def start_recording(call_sid):
    """启动指定通话的录制线程"""
    # 初始化该通话的录制状态
    recording_tasks[call_sid] = {
        'is_recording': True,
        'frames': []
    }
    print(f"开始录制通话 {call_sid}...")
    
    # 启动音频输入流，绑定回调函数
    with sd.InputStream(
        samplerate=fs,
        channels=2,
        callback=lambda indata, frames, time, status: audio_callback(indata, frames, time, status, call_sid)
    ):
        while recording_tasks[call_sid]['is_recording']:
            sd.sleep(100)  # 每隔100ms检查一次录制状态
    
    # 录制停止后保存音频文件
    task = recording_tasks.pop(call_sid)
    if task['frames']:
        audio_data = np.concatenate(task['frames'], axis=0)
        file_path = f"media/audio_call-{call_sid}.wav"
        write(file_path, fs, audio_data)
        print(f"通话 {call_sid} 录制停止，文件保存至：{file_path}")

def stop_recording(call_sid):
    """停止指定通话的录制"""
    task = recording_tasks.get(call_sid)
    if task:
        task['is_recording'] = False
        print(f"收到停止指令，正在结束通话 {call_sid} 的录制...")

第二步：创建Twilio通话状态回调视图 #

Twilio会在通话状态变化时向你指定的URL发送POST请求，我们用这个视图来监听通话结束事件：

from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
def twilio_status_callback(request):
    if request.method == 'POST':
        call_sid = request.POST.get('CallSid')
        call_status = request.POST.get('CallStatus')
        
        if call_status == 'completed' and call_sid in recording_tasks:
            # 通话结束，停止对应录制
            stop_recording(call_sid)
            return HttpResponse('Recording stopped successfully')
    return HttpResponse('OK')

第三步：修改原录制视图，异步启动录制 #

原来的阻塞式录制会卡住Django请求，所以用线程异步启动录制：

def recording_call(request):
    # 这里要从请求或Twilio的参数中拿到当前通话的CallSid
    # 示例：假设你从请求参数里获取CallSid
    call_sid = request.GET.get('call_sid')
    if not call_sid:
        return HttpResponse('Missing CallSid', status=400)
    
    # 启动独立线程录制，不阻塞Django响应
    threading.Thread(target=start_recording, args=(call_sid,)).start()
    return HttpResponse(f"Recording started for call {call_sid}")

第四步：配置Twilio发起呼叫时的回调参数 #

在你创建Twilio通话的代码里，指定status_callback为刚才的回调视图URL，只监听通话结束事件：

from twilio.rest import Client

# 替换成你的Twilio账号信息
TWILIO_ACCOUNT_SID = 'your_account_sid'
TWILIO_AUTH_TOKEN = 'your_auth_token'

client = Client(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)

call = client.calls.create(
    url='http://your-server-url/your-twiml-endpoint',  # 你的Twiml处理地址
    to='+1234567890',
    from_='+0987654321',
    status_callback='http://your-server-url/twilio-status-callback/',  # 回调视图的URL
    status_callback_event=['completed']  # 只监听通话结束事件
)
# 可以把call.sid传递给recording_call视图，关联录制任务

注意事项 #

• 多进程部署兼容：如果你的Django是多worker部署（比如Gunicorn多进程），内存里的recording_tasks字典无法跨进程共享，这时候建议用Redis来存储录制状态
• 文件命名规范：用Twilio的CallSid作为文件名，避免多个通话录制文件互相覆盖
• CSRF豁免：Twilio的回调请求是POST，必须给twilio_status_callback视图加上@csrf_exempt，否则Django会拒绝请求
• 异常处理：可以给录制逻辑加异常捕获，避免录制过程中崩溃导致内存泄漏

这样调整后，当Twilio通话结束时，会立即触发回调通知后端停止对应通话的录制，再也不会出现录满30秒才停的问题了。

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