diff --git a/.env.example b/.env.example new file mode 100644 index 00000000..25f0d6bc --- /dev/null +++ b/.env.example @@ -0,0 +1,5 @@ +RABBITMQ_DEFAULT_USER=wst_user +RABBITMQ_DEFAULT_PASS=wst_pass + +FLOWER_USER=${RABBITMQ_DEFAULT_USER} +FLOWER_PASSWORD=${RABBITMQ_DEFAULT_PASS} \ No newline at end of file diff --git a/app/celery_app.py b/app/celery_app.py new file mode 100644 index 00000000..50b7eba3 --- /dev/null +++ b/app/celery_app.py @@ -0,0 +1,178 @@ +from celery import Celery, Task +from kombu import Exchange, Queue +from celery import bootsteps +from abc import ABC +import os +from app.config import Config + + +config_path = os.environ.get("APP_CONF") +if not config_path: + raise RuntimeError("APP_CONF environment variable is not set") +Config.init_config(config_path) + +broker_url = Config.c.celery.broker_url +result_backend = Config.c.celery.result_backend + +DLX_NAME = "dlx" +DLQ_NAME = "dlq" +DLQ_ROUTING_KEY = "dlq" + + +class DLQTask(Task, ABC): + """ + Отправляет сообщение в DLQ только после исчерпания всех ретраев. + """ + + abstract = True + + def on_failure(self, exc, task_id, args, kwargs, einfo): + self._send_to_dlq(args, kwargs) + super().on_failure(exc, task_id, args, kwargs, einfo) + + def _send_to_dlq(self, args, kwargs): + self.apply_async( + args=args, + kwargs=kwargs, + queue=DLQ_NAME, + exchange=DLX_NAME, + routing_key=DLQ_ROUTING_KEY + ) + + +dlx_exchange = Exchange(DLX_NAME, type="direct", durable=True) + +dead_letter_args = { + "x-dead-letter-exchange": DLX_NAME, + "x-dead-letter-routing-key": DLQ_ROUTING_KEY, +} + +task_queues = [ + Queue( + "default", + Exchange("default", type="direct", durable=True), + routing_key="default", + durable=True, + queue_arguments=dead_letter_args, + ), + Queue( + "audio_recognition", + Exchange("audio_recognition", type="direct", durable=True), + routing_key="audio_recognition", + durable=True, + queue_arguments=dead_letter_args, + ), + Queue( + "audio_processing", + Exchange("audio_processing", type="direct", durable=True), + routing_key="audio_processing", + durable=True, + queue_arguments=dead_letter_args, + ), + Queue( + "presentation_recognition", + Exchange("presentation_recognition", type="direct", durable=True), + routing_key="presentation_recognition", + durable=True, + queue_arguments=dead_letter_args, + ), + Queue( + "presentation_processing", + Exchange("presentation_processing", type="direct", durable=True), + routing_key="presentation_processing", + durable=True, + queue_arguments=dead_letter_args, + ), + Queue( + "training", + Exchange("training", type="direct", durable=True), + routing_key="training", + durable=True, + queue_arguments=dead_letter_args, + ), + Queue( + "passback", + Exchange("passback", type="direct", durable=True), + routing_key="passback", + durable=True, + queue_arguments=dead_letter_args, + ), + Queue( + DLQ_NAME, + dlx_exchange, + routing_key=DLQ_ROUTING_KEY, + durable=True, + ), +] + +celery_app = Celery( + "web_speech_trainer", + broker=broker_url, + backend=result_backend, + include=[ + "app.tasks.audio_recognition", + "app.tasks.audio_processing", + "app.tasks.presentation_recognition", + "app.tasks.presentation_processing", + "app.tasks.training_processing", + "app.tasks.passback_processing", + ], +) + +celery_app.conf.update( + task_serializer="json", + accept_content=["json"], + result_serializer="json", + broker_connection_retry_on_startup = True, + result_extended=True, + result_expires=3600, + timezone="UTC", + enable_utc=True, + task_acks_late=True, + task_reject_on_worker_lost=True, + task_track_started=True, + task_time_limit=30 * 60, + task_soft_time_limit=25 * 60, + worker_prefetch_multiplier=1 +) + +celery_app.conf.task_default_queue = "default" +celery_app.conf.task_default_exchange = "default" +celery_app.conf.task_default_routing_key = "default" +celery_app.conf.task_queues = task_queues + + +celery_app.conf.task_routes = { + "app.tasks.audio_recognition.recognize_audio_task": { + "queue": "audio_recognition", + "exchange": "audio_recognition", + "routing_key": "audio_recognition", + }, + "app.tasks.audio_processing.process_recognized_audio_task": { + "queue": "audio_processing", + "exchange": "audio_processing", + "routing_key": "audio_processing", + }, + "app.tasks.presentation_recognition.recognize_presentation_task": { + "queue": "presentation_recognition", + "exchange": "presentation_recognition", + "routing_key": "presentation_recognition", + }, + "app.tasks.presentation_processing.process_recognized_presentation_task": { + "queue": "presentation_processing", + "exchange": "presentation_processing", + "routing_key": "presentation_processing", + }, + "app.tasks.training_processing.process_training_task": { + "queue": "training", + "exchange": "training", + "routing_key": "training", + }, + "app.tasks.passback_processing.send_score_to_lms_task": { + "queue": "passback", + "exchange": "passback", + "routing_key": "passback", + }, +} + +celery = celery_app diff --git a/app/download_nltk.py b/app/download_nltk.py new file mode 100644 index 00000000..d206a893 --- /dev/null +++ b/app/download_nltk.py @@ -0,0 +1,4 @@ +import nltk + +nltk.download('punkt') +nltk.download('stopwords') \ No newline at end of file diff --git a/app/mongo_odm.py b/app/mongo_odm.py index 63b6e790..05d63d62 100644 --- a/app/mongo_odm.py +++ b/app/mongo_odm.py @@ -315,7 +315,7 @@ def change_audio_status(self, training_id, status): update={'$set': {'audio_status': status, 'audio_status_last_update': datetime.now()}}, ) self.check_failed_training_audio(training_id, status) - self.check_training_ready_for_processing(training_id, status) + # self.check_training_ready_for_processing(training_id, status) def change_presentation_status(self, training_id, status): Trainings.objects.model._mongometa.collection.find_one_and_update( @@ -323,7 +323,7 @@ def change_presentation_status(self, training_id, status): update={'$set': {'presentation_status': status, 'presentation_status_last_update': datetime.now()}}, ) self.check_failed_training_presentation(training_id, status) - self.check_training_ready_for_processing(training_id, status) + # self.check_training_ready_for_processing(training_id, status) def add_recognized_presentation_id(self, training_id, recognized_presentation_id): training = self.get_training(training_id) @@ -543,7 +543,7 @@ def update_scores(self, task_attempt_id, training_id, score): if task_attempt_db is None: return task_attempt_db.training_scores[str(training_id)] = score - self.submit_scores_for_passback(task_attempt_db, training_id) + # self.submit_scores_for_passback(task_attempt_db, training_id) return task_attempt_db.save() def submit_scores_for_passback(self, task_attempt, training_id): diff --git a/app/tasks/audio_processing.py b/app/tasks/audio_processing.py new file mode 100644 index 00000000..340d30b8 --- /dev/null +++ b/app/tasks/audio_processing.py @@ -0,0 +1,85 @@ +from app.celery_app import celery, DLQTask +from app.audio import Audio +from app.recognized_audio import RecognizedAudio +from app.mongo_odm import DBManager, TrainingsDBManager +from app.status import AudioStatus +from celery.exceptions import SoftTimeLimitExceeded +from app.root_logger import get_root_logger + +logger = get_root_logger("audio_processing_task") + + +@celery.task(bind=True, max_retries=3, base=DLQTask) +def process_recognized_audio_task(self, result): + """ + Задача обработки распознанного аудио. + """ + try: + training_id = None + recognized_audio_id = None + + training_id = result["training_id"] + recognized_audio_id = result["recognized_audio_id"] + + logger.info( + f"Starting process_recognized_audio_task for training_id={training_id}, recognized_audio_id={recognized_audio_id}" + ) + + # Обновление статуса + TrainingsDBManager().change_audio_status(training_id, AudioStatus.PROCESSING) + + json_file = DBManager().get_file(recognized_audio_id) + if json_file is None: + raise Exception(f"Recognized audio file {recognized_audio_id} not found") + + # Обработка + recognized_audio = RecognizedAudio.from_json_file(json_file) + json_file.close() + + slide_switch_timestamps = TrainingsDBManager().get_slide_switch_timestamps( + training_id + ) + + audio = Audio(recognized_audio, slide_switch_timestamps) + + # Сохранение результата + audio_id = DBManager().add_file(repr(audio)) + TrainingsDBManager().add_audio_id(training_id, audio_id) + TrainingsDBManager().change_audio_status(training_id, AudioStatus.PROCESSED) + + logger.info( + f"Finished process_recognized_audio_task for training_id={training_id}" + ) + + return { + "status": "success", + "training_id": str(training_id), + "audio_id": str(audio_id), + "type": "audio", + } + + except Exception as exc: + if training_id is None: + logger.error(f"Error in process_recognized_audio_task") + raise + + logger.error( + f"Error in process_recognized_audio_task for training_id={training_id}: {exc}" + ) + if self.request.retries < self.max_retries and not isinstance( + exc, SoftTimeLimitExceeded + ): + logger.info( + f"Retrying process_recognized_audio_task for training_id={training_id}, attempt={self.request.retries + 1}" + ) + raise self.retry(exc=exc, countdown=60) + + TrainingsDBManager().change_audio_status( + training_id, AudioStatus.PROCESSING_FAILED + ) + TrainingsDBManager().append_verdict( + training_id, f"Audio processing failed after all retries: {exc}" + ) + TrainingsDBManager().set_score(training_id, 0) + + raise diff --git a/app/tasks/audio_recognition.py b/app/tasks/audio_recognition.py new file mode 100644 index 00000000..5dc82894 --- /dev/null +++ b/app/tasks/audio_recognition.py @@ -0,0 +1,75 @@ +from app.celery_app import celery, DLQTask +from app.audio_recognizer import WhisperAudioRecognizer +from app.config import Config +from app.mongo_odm import DBManager, TrainingsDBManager +from app.status import AudioStatus +from celery.exceptions import SoftTimeLimitExceeded +from app.root_logger import get_root_logger + +logger = get_root_logger("audio_recognition_task") + + +@celery.task(bind=True, max_retries=3, base=DLQTask) +def recognize_audio_task(self, training_id, presentation_record_file_id): + """ + Задача распознавания аудио. + """ + try: + logger.info( + f"Starting recognize_audio_task for training_id={training_id}, presentation_record_file_id={presentation_record_file_id}" + ) + + # Обновление статуса + TrainingsDBManager().change_audio_status(training_id, AudioStatus.RECOGNIZING) + + presentation_record_file = DBManager().get_file(presentation_record_file_id) + if presentation_record_file is None: + raise Exception( + f"Presentation record file {presentation_record_file_id} not found" + ) + + # Распознавание + recognizer = WhisperAudioRecognizer(url=Config.c.whisper.url) + recognized_audio = recognizer.recognize(presentation_record_file) + + # Сохранение результата + recognized_audio_id = DBManager().add_file(repr(recognized_audio)) + TrainingsDBManager().add_recognized_audio_id(training_id, recognized_audio_id) + TrainingsDBManager().change_audio_status(training_id, AudioStatus.RECOGNIZED) + + TrainingsDBManager().change_audio_status( + training_id, AudioStatus.SENT_FOR_PROCESSING + ) + + logger.info(f"Finished recognize_audio_task for training_id={training_id}") + return { + "status": "success", + "training_id": str(training_id), + "recognized_audio_id": str(recognized_audio_id), + } + + except Exception as exc: + if training_id is None: + logger.error(f"Error in recognize_audio_task") + raise + + logger.error( + f"Error in recognize_audio_task for training_id={training_id}: {exc}" + ) + if self.request.retries < self.max_retries and not isinstance( + exc, SoftTimeLimitExceeded + ): + logger.info( + f"Retrying recognize_audio_task for training_id={training_id}, attempt={self.request.retries + 1}" + ) + raise self.retry(exc=exc, countdown=60) + + TrainingsDBManager().change_audio_status( + training_id, AudioStatus.RECOGNITION_FAILED + ) + TrainingsDBManager().append_verdict( + training_id, f"Recognition failed after all retries: {exc}" + ) + TrainingsDBManager().set_score(training_id, 0) + + raise diff --git a/app/tasks/passback_processing.py b/app/tasks/passback_processing.py new file mode 100644 index 00000000..1c450304 --- /dev/null +++ b/app/tasks/passback_processing.py @@ -0,0 +1,119 @@ +from app.celery_app import celery, DLQTask +from app.mongo_odm import TaskAttemptsDBManager, ConsumersDBManager +from app.status import PassBackStatus +from celery.exceptions import SoftTimeLimitExceeded +from app.root_logger import get_root_logger +from app.utils import is_testing_active +from lti import ToolProvider + +logger = get_root_logger("passback_processing_task") + + +@celery.task(bind=True, max_retries=3, base=DLQTask) +def send_score_to_lms_task(self, training_result): + """ + Отправка оценки в LMS после успешной обработки тренировки. + + training_result: результат от process_training_task + Пример: + { + 'status': 'success', + 'training_id': '...', + 'task_attempt_id': '...', + 'score': 0.85 + } + """ + try: + logger.info(f"Starting send_score_to_lms_task with: {training_result}") + + # Извлечение пришедших данных + training_id = training_result.get("training_id") + task_attempt_id = training_result.get("task_attempt_id") + score = training_result.get("score") + + if not training_id: + error_msg = "No training_id in training_result" + raise Exception(error_msg) + + if not task_attempt_id: + error_msg = f"No task_attempt_id for training {training_id}" + raise Exception(error_msg) + + # Получение данных о попытке + task_attempt_db = TaskAttemptsDBManager().get_task_attempt(task_attempt_id) + if task_attempt_db is None: + error_msg = f"Task attempt {task_attempt_id} not found" + raise Exception(error_msg) + + # Получение параметров для passback + params_for_passback = task_attempt_db.params_for_passback + consumer_secret = ConsumersDBManager().get_secret( + params_for_passback["oauth_consumer_key"] + ) + + # Вычисление нормализованной оценки + training_count = task_attempt_db.training_count + if training_count == 0: + normalized_score = 0 + else: + scores = list(task_attempt_db.training_scores.values()) + total_score = sum([s if s is not None else 0 for s in scores]) + normalized_score = total_score / training_count + + logger.info( + f"Sending score to LMS: task_attempt_id={task_attempt_id}, " + f"training_id={training_id}, score={score}, normalized={normalized_score}" + ) + + # Отправка оценки в LMS + response = ToolProvider.from_unpacked_request( + secret=consumer_secret, params=params_for_passback, headers=None, url=None + ).post_replace_result(score=normalized_score) + + # Проверка результата + if is_testing_active() or ( + response.code_major == "success" and response.severity == "status" + ): + TaskAttemptsDBManager().set_pass_back_status( + task_attempt_db, training_id, PassBackStatus.SUCCESS + ) + logger.info( + f"Score successfully sent to LMS: task_attempt_id={task_attempt_id}, " + f"training_id={training_id}, score={normalized_score}" + ) + return { + "status": "success", + "task_attempt_id": str(task_attempt_id), + "training_id": str(training_id), + "score": normalized_score, + } + else: + error_msg = ( + f"LMS returned error: {response.code_major} - {response.description}" + ) + raise Exception(error_msg) + + except Exception as exc: + if training_id is None: + logger.error(f"Error in recognize_audio_task") + raise + + logger.error( + f"Error in send_score_to_lms_task for training_id={training_id}: {exc}" + ) + if self.request.retries < self.max_retries and not isinstance( + exc, SoftTimeLimitExceeded + ): + logger.info( + f"Retrying send_score_to_lms_task for training_id={training_id}, attempt={self.request.retries + 1}" + ) + raise self.retry(exc=exc, countdown=60) + + if task_attempt_id is not None: + task_attempt_db = TaskAttemptsDBManager().get_task_attempt(task_attempt_id) + if task_attempt_db: + TaskAttemptsDBManager().set_pass_back_status( + task_attempt_db, training_id, PassBackStatus.FAILED + ) + + raise diff --git a/app/tasks/presentation_processing.py b/app/tasks/presentation_processing.py new file mode 100644 index 00000000..eb553ee8 --- /dev/null +++ b/app/tasks/presentation_processing.py @@ -0,0 +1,91 @@ +from app.celery_app import celery, DLQTask +from app.mongo_odm import DBManager, TrainingsDBManager +from app.status import PresentationStatus +from app.recognized_presentation import RecognizedPresentation +from app.presentation import Presentation +from celery.exceptions import SoftTimeLimitExceeded +from app.root_logger import get_root_logger + +logger = get_root_logger("presentation_processing_task") + + +@celery.task(bind=True, max_retries=3, base=DLQTask) +def process_recognized_presentation_task(self, result): + """ + Задача обработки обработки распознанной презентации. + """ + try: + training_id = None + recognized_presentation_id = None + + training_id = result["training_id"] + recognized_presentation_id = result["recognized_presentation_id"] + + logger.info( + f"Starting process_recognized_presentation_task for training_id={training_id}, recognized_presentation_id={recognized_presentation_id}" + ) + + # Обновление статуса + TrainingsDBManager().change_presentation_status( + training_id, PresentationStatus.PROCESSING + ) + + json_file = DBManager().get_file(recognized_presentation_id) + if json_file is None: + raise Exception( + f"Recognized presentation file {recognized_presentation_id} not found" + ) + + # Обработка + recognized_presentation = RecognizedPresentation.from_json_file(json_file) + json_file.close() + + slide_switch_timestamps = TrainingsDBManager().get_slide_switch_timestamps( + training_id + ) + + presentation = Presentation(recognized_presentation, slide_switch_timestamps) + + # Сохранение результата + presentation_id = DBManager().add_file(repr(presentation)) + TrainingsDBManager().add_presentation_id(training_id, presentation_id) + TrainingsDBManager().change_presentation_status( + training_id, PresentationStatus.PROCESSED + ) + + logger.info( + f"Finished process_recognized_presentation_task for training_id={training_id}" + ) + + return { + "status": "success", + "training_id": str(training_id), + "presentation_id": str(presentation_id), + "type": "presentation", + } + + except Exception as exc: + if training_id is None: + logger.error(f"Error in process_recognized_audio_task") + raise + + logger.error( + f"Error in process_recognized_presentation_task for training_id={training_id}: {exc}" + ) + if self.request.retries < self.max_retries and not isinstance( + exc, SoftTimeLimitExceeded + ): + logger.info( + f"Retrying process_recognized_presentation_task for training_id={training_id}, attempt={self.request.retries + 1}" + ) + raise self.retry(exc=exc, countdown=10) + + TrainingsDBManager().change_presentation_status( + training_id, PresentationStatus.PROCESSING_FAILED + ) + TrainingsDBManager().append_verdict( + training_id, f"Presentation processing failed after all retries: {exc}" + ) + TrainingsDBManager().set_score(training_id, 0) + + raise diff --git a/app/tasks/presentation_recognition.py b/app/tasks/presentation_recognition.py new file mode 100644 index 00000000..dc3d421c --- /dev/null +++ b/app/tasks/presentation_recognition.py @@ -0,0 +1,105 @@ +from app.celery_app import celery, DLQTask +from app.mongo_odm import DBManager, TrainingsDBManager, PresentationFilesDBManager +from app.presentation_recognizer import PRESENTATION_RECOGNIZERS +from app.status import PresentationStatus +from celery.exceptions import SoftTimeLimitExceeded +from app.root_logger import get_root_logger + +logger = get_root_logger("presentation_recognition_task") + + +@celery.task(bind=True, max_retries=3, base=DLQTask) +def recognize_presentation_task(self, training_id, presentation_file_id): + """ + Задача распознавания презентации. + """ + try: + logger.info( + f"Starting recognize_presentation_task for training_id={training_id}, presentation_file_id={presentation_file_id}" + ) + + # Обновление статуса + TrainingsDBManager().change_presentation_status( + training_id, PresentationStatus.RECOGNIZING + ) + + # Получение информации о файле презентации + presentation_file_info = PresentationFilesDBManager().get_presentation_file( + presentation_file_id + ) + if presentation_file_info is None: + raise Exception( + f"Presentation file info for {presentation_file_id} not found" + ) + + # Определение расширения и нужного recognizer + pres_extension = "pdf" + nonconverted_file_id = None + if presentation_file_info.presentation_info: + pres_extension = presentation_file_info.presentation_info.filetype + nonconverted_file_id = ( + presentation_file_info.presentation_info.nonconverted_file_id + ) + + file_id_to_fetch = ( + presentation_file_id if not nonconverted_file_id else nonconverted_file_id + ) + presentation_file = DBManager().get_file(file_id_to_fetch) + if presentation_file is None: + raise Exception(f"Presentation file {file_id_to_fetch} not found") + + # Выбор recognizer по расширению + recognizer = PRESENTATION_RECOGNIZERS.get(pres_extension) + if recognizer is None: + raise Exception(f"No recognizer for extension {pres_extension}") + + # Распознавание + recognized_presentation = recognizer.recognize(presentation_file) + + # Сохранение результата + recognized_presentation_id = DBManager().add_file(repr(recognized_presentation)) + TrainingsDBManager().add_recognized_presentation_id( + training_id, recognized_presentation_id + ) + TrainingsDBManager().change_presentation_status( + training_id, PresentationStatus.RECOGNIZED + ) + + TrainingsDBManager().change_presentation_status( + training_id, PresentationStatus.SENT_FOR_PROCESSING + ) + + logger.info( + f"Finished recognize_presentation_task for training_id={training_id}" + ) + return { + "status": "success", + "training_id": str(training_id), + "recognized_presentation_id": str(recognized_presentation_id), + } + + except Exception as exc: + if training_id is None: + logger.error(f"Error in recognize_presentation_task") + raise + + logger.error( + f"Error in recognize_presentation_task for training_id={training_id}: {exc}" + ) + if self.request.retries < self.max_retries and not isinstance( + exc, SoftTimeLimitExceeded + ): + logger.info( + f"Retrying recognize_presentation_task for training_id={training_id}, attempt={self.request.retries + 1}" + ) + raise self.retry(exc=exc, countdown=60) + + TrainingsDBManager().change_presentation_status( + training_id, PresentationStatus.RECOGNITION_FAILED + ) + TrainingsDBManager().append_verdict( + training_id, f"Presentation recognition failed after all retries: {exc}" + ) + TrainingsDBManager().set_score(training_id, 0) + + raise diff --git a/app/tasks/training_processing.py b/app/tasks/training_processing.py new file mode 100644 index 00000000..c5bfe7ef --- /dev/null +++ b/app/tasks/training_processing.py @@ -0,0 +1,164 @@ +from app.celery_app import celery, DLQTask +from app.audio import Audio +from app.criteria_pack import CriteriaPackFactory +from app.feedback_evaluator import FeedbackEvaluatorFactory +from app.mongo_odm import ( + CriterionPackDBManager, + DBManager, + TrainingsDBManager, + TaskAttemptsDBManager, +) +from app.presentation import Presentation +from celery.exceptions import SoftTimeLimitExceeded +from app.root_logger import get_root_logger +from app.status import TrainingStatus +from app.training import Training + +logger = get_root_logger("training_processing_task") + + +@celery.task(bind=True, max_retries=3, base=DLQTask) +def process_training_task(self, results): + """ + Финальная обработка тренировки: вычисление оценки по критериям. + Вызывается как callback после завершения обеих цепочек (audio и presentation). + + results: список результатов из group (два элемента: audio и presentation) + Пример: + [ + {'status': 'success', 'training_id': '...', 'audio_id': '...', 'type': 'audio'}, + {'status': 'success', 'training_id': '...', 'presentation_id': '...', 'type': 'presentation'} + ] + """ + try: + logger.info(f"Starting process_training_task with results: {results}") + + # Извлечение training_id + training_id = None + for result in results: + if result.get("training_id"): + training_id = result.get("training_id") + break + + if not training_id: + error_msg = "No training_id found in results" + raise Exception(error_msg) + + # Извлечение audio_id и presentation_id из результатов + audio_id = None + presentation_id = None + + for result in results: + if result.get("type") == "audio": + audio_id = result.get("audio_id") + elif result.get("type") == "presentation": + presentation_id = result.get("presentation_id") + + if not audio_id or not presentation_id: + error_msg = ( + f"Missing audio_id ({audio_id}) or presentation_id ({presentation_id})" + ) + raise Exception(error_msg) + + # Обновление статуса тренировки + TrainingsDBManager().change_training_status_by_training_id( + training_id, TrainingStatus.PROCESSING + ) + + # Загрузка аудио + audio_file = DBManager().get_file(audio_id) + if audio_file is None: + raise Exception(f"Audio file {audio_id} not found") + audio = Audio.from_json_file(audio_file) + audio_file.close() + logger.info(f"Loaded audio for training_id={training_id}") + + # Загрузка презентации + presentation_file = DBManager().get_file(presentation_id) + if presentation_file is None: + raise Exception(f"Presentation file {presentation_id} not found") + presentation = Presentation.from_json_file(presentation_file) + presentation_file.close() + logger.info(f"Loaded presentation for training_id={training_id}") + + # Получение тренировки из БД + training_db = TrainingsDBManager().get_training(training_id) + if training_db is None: + raise Exception(f"Training {training_id} not found") + + # Получение критериев и оценщика + criteria_pack = CriteriaPackFactory().get_criteria_pack( + training_db.criteria_pack_id + ) + criteria_pack_db = CriterionPackDBManager().get_criterion_pack_by_name( + criteria_pack.name + ) + + feedback_evaluator_id = training_db.feedback_evaluator_id + feedback_evaluator = FeedbackEvaluatorFactory().get_feedback_evaluator( + feedback_evaluator_id + )(criteria_pack_db.criterion_weights) + logger.info( + f"Loaded criteria pack and feedback evaluator for training_id={training_id}" + ) + + # Вычисление обратной связи + training = Training( + training_id, audio, presentation, criteria_pack, feedback_evaluator + ) + + feedback = training.evaluate_feedback() + logger.info( + f"Feedback evaluated for training_id={training_id}, score={feedback.score}" + ) + + # Сохраненеие результатов + TrainingsDBManager().set_score(training_id, feedback.score) + TrainingsDBManager().change_training_status_by_training_id( + training_id, TrainingStatus.PROCESSED + ) + + # Обновление scores в task_attempt + task_attempt_id = training_db.task_attempt_id + if task_attempt_id: + TaskAttemptsDBManager().update_scores( + task_attempt_id, training_id, feedback.score + ) + logger.info( + f"Updated task_attempt {task_attempt_id} with score={feedback.score}" + ) + + logger.info(f"Finished process_training_task for training_id={training_id}") + + return { + "status": "success", + "training_id": str(training_id), + "task_attempt_id": str(task_attempt_id), + "score": feedback.score, + } + + except Exception as exc: + if training_id is None: + logger.error(f"Error in recognize_presentation_task") + raise + + logger.error( + f"Error in process_training_task for training_id={training_id}: {exc}" + ) + if self.request.retries < self.max_retries and not isinstance( + exc, SoftTimeLimitExceeded + ): + logger.info( + f"Retrying process_training_task for training_id={training_id}, attempt={self.request.retries + 1}" + ) + raise self.retry(exc=exc, countdown=60) + + TrainingsDBManager().change_training_status_by_training_id( + training_id, TrainingStatus.PROCESSING_FAILED + ) + TrainingsDBManager().append_verdict( + training_id, f"Training processing failed after all retries: {exc}" + ) + TrainingsDBManager().set_score(training_id, 0) + + raise diff --git a/app/training_manager.py b/app/training_manager.py index 5ca4a139..cf7f731b 100644 --- a/app/training_manager.py +++ b/app/training_manager.py @@ -1,12 +1,22 @@ from datetime import datetime -from app.mongo_odm import TrainingsDBManager, PresentationsToRecognizeDBManager, AudioToRecognizeDBManager +from app.mongo_odm import TrainingsDBManager from app.status import TrainingStatus, AudioStatus, PresentationStatus +from celery import chord, group, chain +from app.tasks.audio_recognition import recognize_audio_task +from app.tasks.audio_processing import process_recognized_audio_task +from app.tasks.presentation_recognition import recognize_presentation_task +from app.tasks.presentation_processing import process_recognized_presentation_task +from app.tasks.training_processing import process_training_task +from app.tasks.passback_processing import send_score_to_lms_task +from app.root_logger import get_root_logger + +logger = get_root_logger("training_manager") class TrainingManager: def __new__(cls): - if not hasattr(cls, 'instance'): + if not hasattr(cls, "instance"): cls.instance = super(TrainingManager, cls).__new__(cls) return cls.instance @@ -14,9 +24,45 @@ def add_training(self, training_id): training = TrainingsDBManager().get_training(training_id) presentation_file_id = training.presentation_file_id presentation_record_file_id = training.presentation_record_file_id - PresentationsToRecognizeDBManager().add_presentation_to_recognize(presentation_file_id, training_id) - TrainingsDBManager().change_presentation_status(training_id, PresentationStatus.SENT_FOR_RECOGNITION) - AudioToRecognizeDBManager().add_audio_to_recognize(presentation_record_file_id, training_id) - TrainingsDBManager().change_audio_status(training_id, AudioStatus.SENT_FOR_RECOGNITION) + + TrainingsDBManager().change_presentation_status( + training_id, PresentationStatus.SENT_FOR_RECOGNITION + ) + TrainingsDBManager().change_audio_status( + training_id, AudioStatus.SENT_FOR_RECOGNITION + ) TrainingsDBManager().set_processing_start_timestamp(training_id, datetime.now()) - TrainingsDBManager().change_training_status_by_training_id(training_id, TrainingStatus.PREPARING) + TrainingsDBManager().change_training_status_by_training_id( + training_id, TrainingStatus.PREPARING + ) + + audio_chain = chain( + recognize_audio_task.s(str(training_id), str(presentation_record_file_id)), + process_recognized_audio_task.s(), + ) + + presentation_chain = chain( + recognize_presentation_task.s(str(training_id), str(presentation_file_id)), + process_recognized_presentation_task.s(), + ) + + workflow = ( + chord(group(audio_chain, presentation_chain), process_training_task.s()) + | send_score_to_lms_task.s() + ) + + try: + result = workflow.apply_async() + logger.info( + f"Pipeline started: training_id={training_id}, last pipeline task_id={result.id}" + ) + except Exception as exc: + logger.error( + f"Failed to start pipeline for training_id={training_id}: {exc}" + ) + TrainingsDBManager().change_training_status_by_training_id( + training_id, TrainingStatus.PROCESSING_FAILED + ) + TrainingsDBManager().append_verdict( + training_id, f"Failed to start pipeline: {exc}" + ) diff --git a/app_conf/testing.ini b/app_conf/testing.ini index de28da76..037f97b0 100644 --- a/app_conf/testing.ini +++ b/app_conf/testing.ini @@ -49,3 +49,8 @@ language=ru [bugreport] form_link=forms/ report_mail=mail + +[celery] +broker_url=amqp://wst_user:wst_pass@rabbitmq:5672// +result_backend=redis://redis:6379/0 + diff --git a/docker-compose.workers.yml b/docker-compose.workers.yml new file mode 100644 index 00000000..8ee42a13 --- /dev/null +++ b/docker-compose.workers.yml @@ -0,0 +1,96 @@ +version: '2' + +services: + flower: + image: wst-image:v0.2 + command: celery -A app.celery_app flower --port=5555 --basic_auth=${FLOWER_USER}:${FLOWER_PASSWORD} --broker_api=http://${RABBITMQ_DEFAULT_USER}:${RABBITMQ_DEFAULT_PASS}@rabbitmq:15672/api/ + ports: + - "5555:5555" + environment: + - APP_CONF=${APP_CONF} + - FLOWER_USER=${FLOWER_USER} + - FLOWER_PASSWORD=${FLOWER_PASSWORD} + - RABBITMQ_DEFAULT_USER=${RABBITMQ_DEFAULT_USER} + - RABBITMQ_DEFAULT_PASS=${RABBITMQ_DEFAULT_PASS} + depends_on: + - celery_worker_audio_rec + - celery_worker_audio_proc + - celery_worker_pres_rec + - celery_worker_pres_proc + - celery_worker_training + - celery_worker_passback + - celery_worker_default + restart: always + networks: + - wst_network + + celery_worker_audio_rec: + image: wst-image:v0.2 + command: celery -A app.celery_app worker --loglevel=info --concurrency=1 -Q audio_recognition + environment: + - APP_CONF=${APP_CONF} + networks: + - wst_network + + celery_worker_audio_proc: + image: wst-image:v0.2 + command: celery -A app.celery_app worker --loglevel=info --concurrency=1 -Q audio_processing + environment: + - APP_CONF=${APP_CONF} + networks: + - wst_network + + celery_worker_pres_rec: + image: wst-image:v0.2 + command: celery -A app.celery_app worker --loglevel=info --concurrency=1 -Q presentation_recognition + environment: + - APP_CONF=${APP_CONF} + volumes: + - nltk_data:/root/nltk_data + networks: + - wst_network + + celery_worker_pres_proc: + image: wst-image:v0.2 + command: celery -A app.celery_app worker --loglevel=info --concurrency=1 -Q presentation_processing + environment: + - APP_CONF=${APP_CONF} + networks: + - wst_network + + celery_worker_training: + image: wst-image:v0.2 + command: celery -A app.celery_app worker --loglevel=info --concurrency=1 -Q training + environment: + - APP_CONF=${APP_CONF} + volumes: + - nltk_data:/root/nltk_data + networks: + - wst_network + + celery_worker_passback: + image: wst-image:v0.2 + command: celery -A app.celery_app worker --loglevel=info --concurrency=1 -Q passback + environment: + - APP_CONF=${APP_CONF} + networks: + - wst_network + + celery_worker_default: + image: wst-image:v0.2 + command: celery -A app.celery_app worker --loglevel=info --concurrency=1 -Q default + environment: + - APP_CONF=${APP_CONF} + networks: + - wst_network + +volumes: + nltk_data: + external: true + name: web_speech_trainer_nltk_data + + +networks: + wst_network: + external: true + name: web_speech_trainer_network \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 15015d71..d62f021d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,70 +1,22 @@ services: web: image: wst-image:v0.2 + build: . init: true command: bash -c "python3 ./db_versioning/db_versioning.py --mongo mongodb://db:27017 && python3 -m web_speech_trainer $APP_CONF" + environment: + - APP_CONF=${APP_CONF} restart: always ports: - 5000:5000 depends_on: - db - - training_processor + - rabbitmq + - nltk_downloader volumes: - ../database-dump:/app/dump/database-dump/ - - audio_processor: - image: wst-image:v0.2 - command: python3 -m audio_processor $APP_CONF - restart: always - depends_on: - - db - - whisper - - presentation_processor - - recognized_audio_processor: - image: wst-image:v0.2 - command: python3 -m recognized_audio_processor $APP_CONF - restart: always - depends_on: - - db - - audio_processor - - presentation_processor - - presentation_processor: - image: wst-image:v0.2 - build: . - command: python3 -m presentation_processor $APP_CONF - restart: always - depends_on: - - db - volumes: - - nltk_data:/root/nltk_data - - recognized_presentation_processor: - image: wst-image:v0.2 - command: python3 -m recognized_presentation_processor $APP_CONF - restart: always - depends_on: - - db - - presentation_processor - - training_processor: - image: wst-image:v0.2 - command: python3 -m training_processor $APP_CONF - restart: always - depends_on: - - db - - recognized_audio_processor - - recognized_presentation_processor - volumes: - - nltk_data:/root/nltk_data - - task_attempt_to_pass_back_processor: - image: wst-image:v0.2 - command: python3 -m task_attempt_to_pass_back_processor $APP_CONF - restart: always - depends_on: - - db + networks: + - network db: image: "mongo:8-noble" @@ -73,6 +25,8 @@ services: - '--wiredTigerCacheSizeGB=2' volumes: - ../mongo_data:/data/db + networks: + - network whisper: image: "onerahmet/openai-whisper-asr-webservice:v1.3.0" @@ -84,7 +38,48 @@ services: mem_limit: ${WHISPER_MEM_LIMIT:-5g} volumes: - whisper_models:/root/.cache/whisper + networks: + - network + + nltk_downloader: + image: wst-image:v0.2 + command: python3 download_nltk.py + volumes: + - nltk_data:/root/nltk_data + restart: no + networks: + - network + + rabbitmq: + image: rabbitmq:3-management + restart: always + ports: + - "5672:5672" + - "15672:15672" + environment: + - RABBITMQ_DEFAULT_USER=${RABBITMQ_DEFAULT_USER} + - RABBITMQ_DEFAULT_PASS=${RABBITMQ_DEFAULT_PASS} + networks: + - network + + redis: + image: redis:7-alpine + command: redis-server --appendonly yes --appendfsync everysec + restart: always + ports: + - "6379:6379" + volumes: + - redis_data:/data + networks: + - network + volumes: whisper_models: - nltk_data: + redis_data: + nltk_data: + + +networks: + network: + driver: bridge \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 8a00c9cb..6659c107 100644 --- a/requirements.txt +++ b/requirements.txt @@ -27,3 +27,5 @@ scipy ==1.12.0 ua-parser ==0.18 scikit-learn ==1.4.2 gensim ==4.3.2 +celery[redis] ==5.6.2 +flower ==2.0.1