介绍

在后端开发中，音频/大文件上传是非常常见的需求，直接将文件全部加载到内存会导致服务器内存溢出，尤其针对大体积音频文件，流式读取写入是最优解决方案。

本文基于 Flask 框架，实现一个安全、高效、支持大文件流式上传的音频上传接口，包含 API-Key 鉴权、格式校验、文件流式存储、唯一文件名生成等核心功能，可直接复用在语音识别、音频处理等业务场景中。

关键

服务器端的写法，是 原始二进制流上传，所以客户端不能是表单上传、json上传，必须用代码实现上传

服务器端实现

功能梳理

接收客户端 POST 请求，上传音频文件（支持 WAV/MP3 等格式）
接口鉴权：通过 API-Key 验证请求合法性
格式校验：仅允许指定格式的音频文件
大文件支持：流式读取写入，不占用过量内存
文件名防冲突：使用 UUID 生成唯一文件名
标准化响应：统一返回 JSON 格式结果

1. 依赖与基础配置

首先初始化 Flask 应用，配置上传目录、允许的音频格式、合法 API-Key 列表：

import os
import uuid
from flask import Flask, request, jsonify

# 初始化Flask应用
app = Flask(__name__)

# 配置：文件上传存储目录
app.config['UPLOAD_FOLDER'] = "uploads"
'''
指定要上传的文件的最大大小是1M， 超出大小会抛出RequestEntityTooLarge 异常
浏览器显示413错误
'''
app.config['MAX_CONTENT_PATH']=1024*1024

# 自动创建上传目录（避免路径不存在报错）
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

# 配置：允许上传的音频格式
ALLOWED_EXTENSIONS = {'WAV', 'MP3', 'FLAC', 'M4A'}
# 配置：合法的API-Key鉴权列表
apikeys = {"test_token_123", "prod_token_456"}

2. 核心上传接口实现

这是接口的核心逻辑，重点使用 request.stream 实现流式处理，避免大文件内存占用：

@app.route('/api/sense_voice_small', methods=['POST'])
def sense_voice_small():
    # 获取音频的二进制数据
    # request.data 直接读取 HTTP Body 中的原始字节流
    # 这里不要用，否则下面无法流式写入数据
    # audio_binary = request.data

    # 获取单个参数 (推荐方式，如果不存在返回 None)
    audio_format:str = request.args.get('format')
    apikey = request.args.get('apikey')
    # print("request.args:",request.args)
    # 校验apikey
    if apikey not in apikeys:
        return jsonify({
            "err_no":500,
            "err_msg":"apikey无效"
        })

    if audio_format.upper() not in ALLOWED_EXTENSIONS:
        return jsonify({
            "code":500,
            "msg":"只能上传 "+ ",".join(ALLOWED_EXTENSIONS) +" 格式的文件"
        })
    # 获取音频格式信息
    # 客户端请求头中设置了 "Content-Type": "audio/wav; rate=16000"
    # 通过 request.content_type 获取这个字符串
    content_type = request.content_type
    # 输出示例: audio/wav; rate=16000
    print("content_type:",content_type)

    _id = str(uuid.uuid1()).replace('-', '')
    filename = _id + "." + audio_format
    path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
    print("path:",path)
    chunk_total = 0
    # 使用 input_stream 进行流式读取，适合大文件
    # 这样可以避免一次性把几个 G 的视频音频加载到内存
    with open(path, 'wb') as f:
        # f.write(audio_binary)
        for chunk in request.stream:
            f.write(chunk)
            chunk_total += len(chunk)
    print("chunk_total:",chunk_total)
    if not chunk_total:
        return jsonify({
            "err_no":100,
            "err_msg": "未接收到音频数据"
        })
    # print("写入数据完成")
    result = generate(path)
    # print("result:",result)
    return jsonify({
        "err_no": 0,
        "err_msg": "success",
        "result": result
    })

if __name__ == '__main__':
    # 生产环境请关闭debug
    app.run(host='0.0.0.0', port=5000, debug=True)

详解

1. 为什么用 `request.stream` 而不是 `request.data`？

request.data：会一次性读取整个请求体到内存，上传几GB的大音频文件会直接导致服务器内存溢出；
request.stream：流式逐块读取，每次只加载一小部分数据到内存，写入文件后立即释放，完美支持大文件上传。

代码中特意注释了 request.data，一旦调用会耗尽请求流，导致后续流式读取失败，这是新手最容易踩的坑！

2. 请求参数与请求体设计

URL 参数：format（音频格式）、token（鉴权），适合传递简单参数；
请求体：直接传递音频原始二进制数据，客户端无需封装为表单，适配绝大多数音频采集客户端；
可通过 request.content_type 获取客户端请求头（如 audio/wav; rate=16000），用于校验音频采样率等信息。

3. 安全与健壮性处理

API 鉴权：拦截非法请求，保护接口安全；
格式白名单：仅允许指定音频格式，防止恶意文件上传；
唯一文件名：UUID 生成文件名，彻底避免文件重名覆盖问题；
空数据判断：校验接收的字节数，拒绝空文件请求；
自动创建目录：避免因上传文件夹不存在导致写入失败。

接口调用

上面服务器端的写法，是原始二进制流上传，所以客户端不能是表单上传、json上传，必须用代码实现上传

1. 客户端封装库

import os
import requests
import logging

logger = logging.getLogger(__name__)

url = "http://localhost:8090/api/sense_voice_small"
apikey = os.getenv("API_KEY", "sk-5879235234059")

def speech_to_text(audio_path):

    if not apikey:
        return None

    wav_path = audio_path

    try:
        with open(wav_path, "rb") as f:
            audio_data = f.read()

        print("读取数据完毕")

        params = {
            "format": "mp3",
            "rate": 16000,
            "channel": 1,
            "cuid": "ai_interview_system",
            "apikey": apikey
        }

        headers = {
            "Content-Type": "audio/wav; rate=16000"
        }
        print("url:", url)
        print("params:", params)
        response = requests.post(url, params=params, headers=headers, data=audio_data)
        result = response.json()
        logger.info(f"ASR返回结果: {result}")

        if result.get("err_no") == 0:
            # API返回格式可能是 {"err_no":0, "err_msg":"success", "result":["识别结果"]}
            if isinstance(result.get("result"), list):
                text = "".join(result["result"])
            else:
                text = str(result.get("result", ""))
            logger.info(f"语音识别成功: {text[:50]}...")
            return text
        else:
            logger.error(f"语音识别失败: {result}")
            return None
    except Exception as e:
        logger.error(f"语音识别异常: {str(e)}")
        return None
    finally:
        if os.path.exists(wav_path):
            # os.remove(wav_path)
            pass

2. 客户端请求示例（Python）

from speech_to_text import speech_to_text
from dotenv import load_dotenv

load_dotenv()

audio_path = r"D:\培训作业提交\0426作业提交\backend\uploads\audio\audio_20260429085405.mp3"

asr_text = speech_to_text(audio_path)
print(asr_text)
if asr_text:
    audio_text = asr_text
    print(f"语音转文字成功:",audio_text)
else:
    print(f"语音转文字返回空结果")

3. 成功响应结果

{
    'emotion': 'NEUTRAL',
    'itn': 'withitn',
    'lang': 'zh',
    'text': '嗯，sbo是s频MVC和sprint两大框架的呃集集合，并且呢可以减少很多的配置。然后呢还可以实现呃快速的加载依赖的功能。',
    'type': 'Speech'
}

总结

这套 Flask 音频上传方案，核心优势是流式处理大文件，同时兼顾了接口安全、参数校验、文件管理等核心需求，不仅适用于音频，也可以无缝适配视频、压缩包等所有大文件上传场景。

原文出处：/show_1GW3GJZ0TgwR.html

介绍

关键