Python音频流（ByteArray）分割异常：后续WAV文件无声音

核心原因 #

第二个及后续文件无声音的问题，大概率是因为直接写入原始音频裸数据，缺少WAV文件必需的文件头。第一个文件能正常播放，可能是客户端发送的初始数据中包含了完整的WAV头，而后续的音频流只有裸PCM数据，没有头信息，播放器无法解析音频参数（采样率、位深度、声道数等），导致无法播放。

另外，代码中chunk_size = 16000 * 2的计算假设是16kHz采样率、16位单声道（2字节/采样）的2秒数据，但如果实际音频参数不符，也会导致长度计算错误，出现无声或杂音。

解决方案 #

1. 统一添加WAV文件头 #

每次保存音频块时，为裸PCM数据添加正确的WAV文件头，确保播放器能识别音频参数。

2. 匹配实际音频参数 #

确认客户端发送的音频参数（采样率、位深度、声道数）与代码中的计算一致，避免chunk_size计算错误。

修改后的代码 #

import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from dotenv import load_dotenv

load_dotenv()
app = FastAPI()

def add_wav_header(pcm_data, sample_rate=16000, bits_per_sample=16, channels=1):
    """为裸PCM数据添加WAV文件头"""
    data_size = len(pcm_data)
    header = bytearray()
    # RIFF标识
    header.extend(b'RIFF')
    # 文件总大小（RIFF头 + fmt块 + data块）
    header.extend((36 + data_size).to_bytes(4, byteorder='little'))
    # WAVE标识
    header.extend(b'WAVE')
    # fmt块标识
    header.extend(b'fmt ')
    # fmt块大小（PCM格式固定为16）
    header.extend((16).to_bytes(4, byteorder='little'))
    # 音频格式（PCM为1）
    header.extend((1).to_bytes(2, byteorder='little'))
    # 声道数
    header.extend((channels).to_bytes(2, byteorder='little'))
    # 采样率
    header.extend((sample_rate).to_bytes(4, byteorder='little'))
    # 字节率 = 采样率 * 声道数 * 位深度/8
    byte_rate = sample_rate * channels * (bits_per_sample // 8)
    header.extend((byte_rate).to_bytes(4, byteorder='little'))
    # 块对齐 = 声道数 * 位深度/8
    block_align = channels * (bits_per_sample // 8)
    header.extend((block_align).to_bytes(2, byteorder='little'))
    # 位深度
    header.extend((bits_per_sample).to_bytes(2, byteorder='little'))
    # data块标识
    header.extend(b'data')
    # data块大小
    header.extend((data_size).to_bytes(4, byteorder='little'))
    return header + pcm_data

async def transcribe_chunk(audio_chunk, filename: str):
    print("filename:", filename)
    wav_data = add_wav_header(audio_chunk)
    with open(f'audio_chunk{filename}.wav', 'wb') as f:
        print("audio_chunk size:", len(audio_chunk))
        f.write(wav_data)

@app.websocket("/audio")
async def websocket_endpoint(websocket: WebSocket, token: str):
    await websocket.accept()
    audio_buffer = bytearray()
    # 根据实际音频参数计算chunk_size：16kHz、16位单声道，2秒数据量=16000*1*2*2=64000字节
    sample_rate = 16000
    bits_per_sample = 16
    channels = 1
    chunk_duration = 2  # 秒
    chunk_size = sample_rate * channels * (bits_per_sample // 8) * chunk_duration
    tasks = []
    i = 0
    try:
        while True:
            data = await websocket.receive_bytes()
            print("received data")
            audio_buffer += data

            # 循环处理缓冲区中所有完整的chunk
            while len(audio_buffer) >= chunk_size:
                print("buffer has enough data for a chunk")
                i += 1
                audio_chunk = audio_buffer[:chunk_size]
                task = asyncio.create_task(transcribe_chunk(audio_chunk, f"{token}{i}"))
                tasks.append(task)
                audio_buffer = audio_buffer[chunk_size:]

    except WebSocketDisconnect:
        print("Client disconnected")
        if audio_buffer:
            i += 1
            task = asyncio.create_task(transcribe_chunk(audio_buffer, f"{token}{i}_remaining"))
            tasks.append(task)

    await asyncio.gather(*tasks)

额外注意事项 #

• 若客户端使用其他音频参数（比如48kHz采样率、立体声），需同步修改add_wav_header和chunk_size的对应参数。
• 如果客户端发送的是完整的WAV文件流（每次包含头），则需要在接收时识别每个WAV文件的边界，而非按固定大小分割。

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