import whisper
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
import io
import torch
import numpy as np
from pydantic import BaseModel
import os
import tempfile# 加载 Whisper 模型
model = whisper.load_model("medium")# FastAPI 应用
app = FastAPI()def pad_or_trim(array, length: int = 16000 * 30, *, axis: int = -1):"""Split the audio array into multiple segments of length N_SAMPLES if it exceeds N_SAMPLES.If it is shorter, pad the array to N_SAMPLES."""if torch.is_tensor(array):arrays = []for i in range(0, array.shape[axis], length):segment = array.index_select(dim=axis, index=torch.arange(i, min(i + length, array.shape[axis]), device=array.device))if segment.shape[axis] < length:pad_widths = [(0, 0)] * segment.ndimpad_widths[axis] = (0, length - segment.shape[axis])segment = torch.nn.functional.pad(segment, [pad for sizes in pad_widths[::-1] for pad in sizes])arrays.append(segment)array = torch.stack(arrays)else:arrays = []for i in range(0, array.shape[axis], length):segment = array.take(indices=range(i, min(i + length, array.shape[axis])), axis=axis)if segment.shape[axis] < length:pad_widths = [(0, 0)] * segment.ndimpad_widths[axis] = (0, length - segment.shape[axis])segment = np.pad(segment, pad_widths)arrays.append(segment)array = np.stack(arrays)return array@app.post("/transcribe/")
async def transcribe_audio(file: UploadFile = File(...)):# 将上传的文件保存到临时文件with tempfile.NamedTemporaryFile(delete=False) as tmp:tmp.write(file.file.read())tmp_path = tmp.nameaudio = whisper.load_audio(tmp_path)# Pad or trim the audio to 30 secondsaudio = pad_or_trim(audio)async def generate():# Process each audio segment# for segment in audio:# Compute the log-Mel spectrogram and move to the same device as the modelmel = whisper.log_mel_spectrogram(audio).to(model.device)# Detect the spoken language_, probs = model.detect_language(mel)for probs in probs:detected_language = max(probs, key=probs.get)yield f"data: Detected language: {detected_language}\n\n"# Decode the audiooptions = whisper.DecodingOptions()result = whisper.decode(model, mel, options)# Yield the transcribed textfor r in result:yield f"data: {r.text}\n\n"# 删除临时文件os.remove(tmp_path)return StreamingResponse(generate(), media_type="text/event-stream")# 运行 FastAPI 应用
if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=6006)

测试方法

1. 使用 curl

curl -X POST "http://127.0.0.1:8000/transcribe/" -H "accept: text/event-stream" -H "Content-Type: multipart/form-data" -F "file=@path_to_your_audio_file"

2. 使用 requests Python 脚本

import requestsurl = "http://localhost:6006/transcribe/"
file_path = "瓦解－－史塔克.mp3"  # 替换为你的实际音频文件路径with open(file_path, "rb") as f:files = {"file": f}response = requests.post(url, files=files, headers={"accept": "text/event-stream"})print(response)for line in response.iter_lines():if line:print(line.decode("utf-8"))

3. 使用 Postman

• 打开 Postman，创建一个新的 POST 请求。
• 输入 URL：http://127.0.0.1:8000/transcribe/
• 设置请求头：
  • accept：text/event-stream
  • Content-Type：multipart/form-data
• 在 Body 选项卡中选择 form-data，添加一个新的键值对：
  • Key：file
  • Type：File
  • Value：选择你的音频文件
• 发送请求，并查看服务器返回的流数据。

注意事项

• 确保你的 Whisper 模型已经正确安装，并且你的 Python 环境中包含所有必要的依赖项（如 torch, numpy, fastapi, uvicorn 等）。
• 确保 FastAPI 服务正在运行，使用 uvicorn 启动该服务。