FastAPI WebSocket处理Media Recorder Blobs：后续视频片段损坏问题

核心问题原因 #

1. 文件句柄管理不当 #

你的代码中每次循环都会打开目标文件，但仅当缓冲达到阈值时才关闭文件。这会导致：

• 未关闭的文件无法将缓冲数据刷入磁盘，造成文件内容不完整
• 文件句柄泄漏，后续写入操作可能出现不可预料的错误
• 切换到新文件时，旧文件的句柄未释放，直接导致文件结构损坏

2. 缓冲逻辑存在漏洞 #

当bufferedSize达到阈值后，你直接写入当前数据并关闭文件，但未考虑当前数据的长度——如果单段Blob本身超过5MB，会导致单个文件大小超标；同时，关闭文件后未处理剩余数据的衔接，后续文件的起始数据不完整。

3. 缺少异常安全的文件操作 #

未使用Python的with语句管理文件上下文，一旦循环中出现异常，文件会处于未关闭状态，直接引发文件损坏。

修正后的后端代码 #

import os
from datetime import datetime
from fastapi import WebSocket, Query, Depends
# 替换为你项目中对应的导入路径
from your_auth_module import AuthJWT
from your_websocket_manager import manager

@app.websocket("/video")
async def websocket_endpoint(websocket: WebSocket, token: str = Query(...), videoId: str = Query(...), authorize: AuthJWT = Depends()):
    await manager.connect(websocket)

    temp_dir = os.path.join("temp", videoId)
    os.makedirs(temp_dir, exist_ok=True)

    queue_number = 1
    buffered_size = 0
    total_required = 5 * 1024 * 1024  # 5MB阈值
    current_file_path = os.path.join(temp_dir, f"{videoId}-video-{queue_number}.webm")
    
    print(f"{datetime.utcnow()}: Client connected: {websocket.client.host}")
    try:
        while True:
            data = await websocket.receive_bytes()
            data_len = len(data)
            
            # 计算当前文件剩余可写入的空间
            remaining_space = total_required - buffered_size
            
            if data_len <= remaining_space:
                # 当前数据可全部写入当前文件
                with open(current_file_path, 'ab') as f:
                    f.write(data)
                buffered_size += data_len
                print(f"缓冲大小: {buffered_size / (1024*1024):.2f}MB | 当前Blob大小: {data_len/(1024*1024):.2f}MB")
            else:
                # 拆分数据：先写满当前文件，剩余部分写入新文件
                if remaining_space > 0:
                    with open(current_file_path, 'ab') as f:
                        f.write(data[:remaining_space])
                    print(f"完成文件 {queue_number} | 最终大小: {total_required/(1024*1024):.2f}MB")
                
                # 处理剩余数据到新文件
                remaining_data = data[remaining_space:]
                queue_number += 1
                current_file_path = os.path.join(temp_dir, f"{videoId}-video-{queue_number}.webm")
                with open(current_file_path, 'ab') as f:
                    f.write(remaining_data)
                buffered_size = len(remaining_data)
                print(f"开始新文件 {queue_number} | 缓冲大小: {buffered_size/(1024*1024):.2f}MB")
            
            # 刚好达到阈值时，切换到新文件
            if buffered_size >= total_required:
                queue_number += 1
                current_file_path = os.path.join(temp_dir, f"{videoId}-video-{queue_number}.webm")
                buffered_size = 0
                print(f"完成文件 {queue_number-1} | 准备接收新文件")

    except WebSocketDisconnect:
        # 断开连接时保存未完成的文件
        if buffered_size > 0:
            print(f"客户端断开连接，保存未完成的文件 {queue_number}")
        print(f"{datetime.utcnow()}: Client disconnected: {websocket.client.host}")
        manager.disconnect(websocket)

额外注意事项 #

• 前端MediaRecorder配置：初始化MediaRecorder时指定统一的MIME类型（如video/webm;codecs=vp9），避免不同浏览器编码差异导致的文件兼容性问题。
• 文件完整性校验：可以在后端检查每个生成的WebM文件头是否包含WebM标识（\x1a\x45\xdf\xa3），快速排查无效文件。
• 异步优化：如果文件写入成为性能瓶颈，可使用aiofiles库进行异步文件操作，避免阻塞WebSocket事件循环。

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