Migration to Celery RabbitMQ

.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}

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

app/download_nltk.py

@@ -0,0 +1,4 @@
+import nltk
+
+nltk.download('punkt')
+nltk.download('stopwords')

app/mongo_odm.py

@@ -315,15 +315,15 @@ 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(
             filter={'_id': ObjectId(training_id)},
             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):

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

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