Flask 实现原始二进制流上传(客户端不能是表单上传、json上传,必须用代码实现上传) 作者:马育民 • 2026-05-06 16:33 • 阅读:10000 # 介绍 在后端开发中,**音频/大文件上传**是非常常见的需求,直接将文件全部加载到内存会导致服务器内存溢出,尤其针对大体积音频文件,**流式读取写入**是最优解决方案。 本文基于 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