Flask 实现原始二进制流上传（客户端不能是表单上传、json上传，必须用代码实现上传）-马育民老师

# 介绍

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

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

### 关键

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

# 服务器端实现

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

### 1. 依赖与基础配置
首先初始化 Flask 应用，配置上传目录、允许的音频格式、合法 API-Key 列表：
```python
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` 实现流式处理**，避免大文件内存占用：

```python
@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. 安全与健壮性处理
1. **API 鉴权**：拦截非法请求，保护接口安全；
2. **格式白名单**：仅允许指定音频格式，防止恶意文件上传；
3. **唯一文件名**：UUID 生成文件名，彻底避免文件重名覆盖问题；
4. **空数据判断**：校验接收的字节数，拒绝空文件请求；
5. **自动创建目录**：避免因上传文件夹不存在导致写入失败。

# 接口调用

上面服务器端的写法，是原始二进制流上传，所以客户端不能是表单上传、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）
```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. 成功响应结果
```json
{
	'emotion': 'NEUTRAL',
	'itn': 'withitn',
	'lang': 'zh',
	'text': '嗯，sbo是s频MVC和sprint两大框架的呃集集合，并且呢可以减少很多的配置。然后呢还可以实现呃快速的加载依赖的功能。',
	'type': 'Speech'
}
```

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

原文出处：http://malaoshi.top/show_1GW3GJZ0TgwR.html