Flask API에서 파일을 디스크 대신 메모리로 처리하기

from flask import Flask, request, jsonify
import torch
import logging
import module_model
import os
from werkzeug.utils import secure_filename
import numpy as np
from flask_cors import CORS
from threading import Lock
 
app = Flask(__name__)
 
CORS(app, resources={r"/*": {"origins": "*"}})
 
app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['ALLOWED_EXTENSIONS'] = {'wav', 'mp3', 'mp4', 'flac'}  # 허용 파일 형식
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 최대 16MB 파일
 
job_lock = Lock()
 
def get_device(device_type: str) -> torch.device:
    return torch.device("cuda" if device_type in ['gpu', 'cuda'] and torch.cuda.is_available() else "cpu")
 
def allowed_file(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS']
 
def scoreto(score):
    score = (1 / (1 + np.exp(2 * (score + 6))))
    return score
 
@app.route('/analyze', methods=['POST'])
def analyze():
    if 'file' not in request.files:
        return jsonify({'error': 'No file part'}), 400
 
    file = request.files['file']
 
    if file and allowed_file(file.filename):
        filename = secure_filename(file.filename)
        filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
        file.save(filepath)
 
        device_type = request.form.get('device', 'gpu')
        device = get_device(device_type)
 
        logging.info("Received File: %s", filename)
        logging.info("Device Type: %s", device)
 
        with job_lock:
            logging.info("Job started for file: %s", filename)
            score = module_model.test_model(filepath, device)
            score_values = [scoreto(s.item()) for s in score]
            stt = module_model.STT(filepath, device)
            logging.info("STT Result: %s", stt)
            logging.info("Analyzed Score: %s", score_values)
            logging.info("Job finished for file: %s", filename)
 
        return jsonify({'score': score_values, 'text': stt})
    else:
        return jsonify({'error': 'Invalid file type'}), 400
 
@app.route('/')
def user():
    return 'Hello, World!'
 
if __name__ == '__main__':
    if not os.path.exists(app.config['UPLOAD_FOLDER']):
        os.makedirs(app.config['UPLOAD_FOLDER'])
 
    app.run('0.0.0.0', port=2580, debug=True)

@app.route('/analyze', methods=['POST'])
def analyze():
    if 'file' not in request.files:
        return jsonify({'error': 'No file part'}), 400
 
    file = request.files['file']
    filename = file.filename
 
    if not filename or '.' not in filename:
        return jsonify({'error': 'Invalid filename'}), 400
    ext = filename.rsplit('.', 1)[1].lower()
    if ext not in {'wav', 'mp3', 'mp4', 'flac'}:
        return jsonify({'error': 'Invalid file type'}), 400
 
    file_bytes = file.read()
 
    device_type = request.form.get('device', 'gpu')
    device = get_device(device_type)
 
    with job_lock:
        score = module_model.test_model(file_bytes, device)
        score_values = [scoreto(s.item()) for s in score]
        stt = module_model.STT(file_bytes, device)
 
    return jsonify({'score': score_values, 'text': stt})

# 1. 업로드된 파일을 메모리로 읽는다
file = request.files['file']
file_bytes = file.read()
 
# 2. 필요하면 메모리 버퍼로 감싼다
buffer = io.BytesIO(file_bytes)
 
# 3. 바로 처리한다
score = module_model.test_model(file_bytes, device)