@berilloda

Да, использовать очередь задач — это правильный подход для обработки аудиофайлов по очереди и предотвращения перегрузки вашего сервера.

Вот как вы можете улучшить работу с очередью и потоками в вашем коде:

Используйте ThreadPoolExecutor: Вместо ручного создания потоков можно использовать ThreadPoolExecutor из модуля concurrent.futures для управления потоками более эффективно.
Обработка каждого файла: Убедитесь, что у каждого аудиофайла есть уникальный audio_id и audio_name, чтобы предотвратить перезапись и конфликт данных.
Управляйте ресурсами: Можно добавить лимит на количество одновременно обрабатываемых аудиофайлов в зависимости от возможностей вашего оборудования.

Вот пример кода, который включает эти улучшения:

from flask import Flask, request, jsonify, send_file
import os
import subprocess
import time
from queue import Queue
from concurrent.futures import ThreadPoolExecutor

app = Flask(__name__)

# Директория для файла
directory = os.path.dirname(os.path.abspath(__file__))


# Создаем очередь задач с фиксированным количеством потоков
max_workers = 4  # Количество рабочих потоков в зависимости от возможностей ПК
task_queue = Queue()
executor = ThreadPoolExecutor(max_workers=max_workers)


def delete_file(file_path):
    time.sleep(300)  # 5 минут ожидания
    os.remove(file_path)


def process_audio_task(audio_file, audio_id, audio_name):
    try:
        start_time = time.time()

        # Сохраняем аудио
        audio_path = os.path.join(directory, 'program_audio/audio', f'{audio_name}.mp3')
        audio_file.save(audio_path)

        # Запуск обработки аудио
        subprocess.run(['starter.bat'], shell=True)

        # Ждем результата
        text_file_path = os.path.join(directory, 'program_audio/final_text', f'{audio_name}_final.txt')
        timeout = time.time() + 60
        while not os.path.exists(text_file_path):
            if time.time() > timeout:
                return jsonify({'error': 'Timeout'}), 500
            time.sleep(1)

        # Формируем ответ
        end_time = time.time()
        elapsed_time = end_time - start_time

        response_data = {
            'audio_id': audio_id,
            'elapsed_time': elapsed_time
        }

        # Отправляем результат
        response = send_file(text_file_path, as_attachment=True)

        # Запускаем удаление файла через 5 минут
        delete_thread = threading.Thread(target=delete_file, args=(text_file_path,))
        delete_thread.start()

        return response, response_data

    except Exception as e:
        return jsonify({'error': str(e)}), 500


def process_audio_worker():
    while True:
        audio_file, audio_id, audio_name = task_queue.get()
        executor.submit(process_audio_task, audio_file, audio_id, audio_name)
        task_queue.task_done()


@app.route('/api/process_audio', methods=['POST'])
def queue_process_audio():
    try:
        # Получаем данные
        audio_files = request.files.getlist('audio')
        audio_id = request.form['audio_id']
        audio_name = request.form['audio_name']

        # Добавляем задачи в очередь
        for audio_file in audio_files:
            task_queue.put((audio_file, audio_id, audio_name))

        return jsonify({'message': 'Audio files are added to processing queue.'}), 200

    except Exception as e:
        return jsonify({'error': str(e)}), 500


if __name__ == '__main__':
    # Запускаем поток для обработки задач
    worker_thread = threading.Thread(target=process_audio_worker)
    worker_thread.start()

    app.run(debug=True, port=39588, host='0.0.0.0')

Основные изменения и улучшения:

Оптимизация потоков: Вместо создания отдельных потоков для каждой задачи, использован ThreadPoolExecutor для управления количеством одновременно работающих задач.
Корректное использование subprocess.run: Переведено на список аргументов вместо строки для безопасности и универсальности, хотя использование shell=True все равно требует осторожности.
Удаление файлов: Как и ранее, удаление файлов производится через 5 минут с использованием потоков.

Этот код поможет стабилизировать работу сервера и предотвратит зависание при множественных параллельных запросах.

Как использовать функции высшего порядка в Haskell?

2 ответа

Похожие обсуждения: