feat: add elastic search commands

forum/management/commands/delete_unused_forum_indices.py

@@ -0,0 +1,25 @@
+"""Management command for deleting unused indices"""
+
+from django.core.management.base import BaseCommand
+
+from forum.search.es_helper import ElasticsearchHelper
+
+
+class Command(BaseCommand):
+    help = (
+        "Delete all Elasticsearch indices that are not the latest for each model type."
+    )
+
+    def handle(self, *args: list[str], **kwargs: dict[str, str]) -> None:
+        """
+        Handles the execution of the delete_unused_forum_indices command.
+
+        This command deletes all Elasticsearch indices that are not the latest for each model type.
+        """
+        es_helper = ElasticsearchHelper()
+        indices_deleted_count = es_helper.delete_unused_indices()
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"{indices_deleted_count} unused indices deleted successfully."
+            )
+        )

forum/management/commands/initialize_forum_indices.py

@@ -0,0 +1,41 @@
+"""Management command for initialize forum indices"""
+
+from argparse import ArgumentParser
+from typing import Any
+
+from django.core.management.base import BaseCommand
+
+from forum.search.es_helper import ElasticsearchHelper
+
+
+class Command(BaseCommand):
+    help = "Initialize Elasticsearch indices, optionally forcing the creation of new indices."
+
+    def add_arguments(self, parser: ArgumentParser) -> None:
+        """
+        Adds command line arguments to the initialize_forum_indices command.
+
+        Args:
+            parser: The argument parser to which the --force argument is added.
+        """
+        parser.add_argument(
+            "--force",
+            action="store_true",
+            help="Force the creation of new indices even if they exist.",
+        )
+
+    def handle(self, *args: list[str], **kwargs: dict[str, Any]) -> None:
+        """
+        Handles the execution of the initialize_indices command.
+
+        Initializes Elasticsearch indices. If the --force option is provided,
+        it forces the creation of new indices even if they exist.
+
+        Args:
+            args: Additional arguments.
+            kwargs: Command options.
+        """
+        es_helper = ElasticsearchHelper()
+        force_new_index = bool(kwargs.get("force", False))
+        es_helper.initialize_indices(force_new_index=force_new_index)
+        self.stdout.write(self.style.SUCCESS("Forum indices initialized successfully."))

forum/management/commands/rebuild_forum_indices.py

@@ -0,0 +1,55 @@
+"""Management command for rebuild forum indices"""
+
+from argparse import ArgumentParser
+
+from django.core.management.base import BaseCommand
+
+from forum.search.es_helper import ElasticsearchHelper
+from forum.utils import get_int_value_from_collection
+
+
+class Command(BaseCommand):
+    help = "Rebuild Elasticsearch indices by creating new indices and reindexing data."
+
+    def add_arguments(self, parser: ArgumentParser) -> None:
+        """
+        Adds command line arguments to the rebuild_forum_indices command.
+
+        Args:
+            parser: The argument parser to which the --batch_size and --extra_catchup_minutes arguments are added.
+        """
+        parser.add_argument(
+            "--batch_size",
+            type=int,
+            default=500,
+            help="Number of documents to process in each batch.",
+        )
+        parser.add_argument(
+            "--extra_catchup_minutes",
+            type=int,
+            default=5,
+            help="Extra minutes to adjust the start time for catch-up.",
+        )
+
+    def handle(self, *args: list[str], **kwargs: dict[str, int]) -> None:
+        """
+        Handles the execution of the rebuild_indices command.
+
+        Rebuilds Elasticsearch indices by creating new indices and reindexing data.
+        Batch size and extra catch-up minutes can be specified.
+
+        Args:
+            args: Additional arguments.
+            kwargs: Command options.
+        """
+        es_helper = ElasticsearchHelper()
+
+        batch_size = get_int_value_from_collection(kwargs, "batch_size", 500)
+        extra_catchup_minutes = get_int_value_from_collection(
+            kwargs, "extra_catchup_minutes", 5
+        )
+
+        es_helper.rebuild_indices(
+            batch_size=batch_size, extra_catchup_minutes=extra_catchup_minutes
+        )
+        self.stdout.write(self.style.SUCCESS("Forum indices rebuilt successfully."))

forum/management/commands/validate_forum_indices.py

@@ -0,0 +1,22 @@
+"""Management command for validating forum mappings"""
+
+from django.core.management.base import BaseCommand
+
+from forum.search.es_helper import ElasticsearchHelper
+
+
+class Command(BaseCommand):
+    help = "Validate Elasticsearch indices for correct mappings and properties."
+
+    def handle(self, *args: list[str], **kwargs: dict[str, str]) -> None:
+        """
+        Handles the execution of the validate_forum_indices command.
+
+        This command validates that the Elasticsearch indices have the correct mappings and properties.
+
+        Raises:
+            ValueError: If indices do not exist or if mappings/properties are missing or incorrect.
+        """
+        es_helper = ElasticsearchHelper()
+        es_helper.validate_indices()
+        self.stdout.write(self.style.SUCCESS("Forum indices validated successfully."))

forum/models/base_model.py

@@ -20,6 +20,7 @@ class MongoBaseModel(ABC):
 
     MONGODB_DATABASE: Optional[Database] = None
     COLLECTION_NAME: str = "default"
+    index_name: str = "default"
 
     @property
     def _collection(self) -> Collection:

forum/models/comments.py

@@ -21,6 +21,47 @@ def override_query(self, query: dict[str, Any]) -> dict[str, Any]:
         query = {**query, "_type": self.content_type}
         return super().override_query(query)
 
+    @classmethod
+    def mapping(cls) -> dict[str, Any]:
+        """
+        Mapping function for the Thread class
+        """
+        return {
+            "dynamic": "false",
+            "properties": {
+                "body": {
+                    "type": "text",
+                    "store": True,
+                    "term_vector": "with_positions_offsets",
+                },
+                "course_id": {"type": "keyword"},
+                "comment_thread_id": {"type": "keyword"},
+                "commentable_id": {"type": "keyword"},
+                "group_id": {"type": "keyword"},
+                "context": {"type": "keyword"},
+                "created_at": {"type": "date"},
+                "updated_at": {"type": "date"},
+                "title": {"type": "keyword"},
+            },
+        }
+
+    @classmethod
+    def doc_to_hash(cls, doc: dict[str, Any]) -> dict[str, Any]:
+        """
+        Converts comment document to the dict
+        """
+        return {
+            "body": doc.get("body"),
+            "course_id": doc.get("course_id"),
+            "comment_thread_id": str(doc.get("comment_thread_id")),
+            "commentable_id": doc.get("commentable_id"),
+            "group_id": doc.get("group_id"),
+            "context": doc.get("context", "course"),
+            "created_at": doc.get("created_at"),
+            "updated_at": doc.get("updated_at"),
+            "title": doc.get("title"),
+        }
+
     def insert(
         self,
         body: str,

forum/models/contents.py

@@ -17,6 +17,27 @@ class BaseContents(MongoBaseModel):
     content_type: str = ""
     COLLECTION_NAME: str = "contents"
 
+    @classmethod
+    def mapping(cls) -> Dict[str, Any]:
+        """
+        Implement this method in the child class
+        """
+        raise NotImplementedError
+
+    @classmethod
+    def doc_to_hash(cls, doc: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Implement this method in the child class
+        """
+        raise NotImplementedError
+
+    def override_query(self, query: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Override the query with the _type field.
+        """
+        query = {**query, "_type": self.content_type}
+        return super().override_query(query)
+
     def list(self, **kwargs: Any) -> Any:
         """
         Retrieves a list of all content documents in the database based on provided filters.

forum/models/threads.py

@@ -16,6 +16,62 @@ class CommentThread(BaseContents):
     index_name = "comment_threads"
     content_type = "CommentThread"
 
+    @classmethod
+    def mapping(cls) -> Dict[str, Any]:
+        """
+        Mapping function for the Thread class
+        """
+        return {
+            "dynamic": "false",
+            "properties": {
+                "title": {
+                    "type": "text",
+                    "boost": 5.0,
+                    "store": True,
+                    "term_vector": "with_positions_offsets",
+                },
+                "body": {
+                    "type": "text",
+                    "store": True,
+                    "term_vector": "with_positions_offsets",
+                },
+                "created_at": {"type": "date"},
+                "updated_at": {"type": "date"},
+                "last_activity_at": {"type": "date"},
+                "comment_count": {"type": "integer"},
+                "votes_point": {"type": "integer"},
+                "context": {"type": "keyword"},
+                "course_id": {"type": "keyword"},
+                "commentable_id": {"type": "keyword"},
+                "author_id": {"type": "keyword"},
+                "group_id": {"type": "integer"},
+                "id": {"type": "keyword"},
+                "thread_id": {"type": "keyword"},
+            },
+        }
+
+    @classmethod
+    def doc_to_hash(cls, doc: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Converts thread document to the dict
+        """
+        return {
+            "id": str(doc.get("_id")),
+            "title": doc.get("title"),
+            "body": doc.get("body"),
+            "created_at": doc.get("created_at"),
+            "updated_at": doc.get("updated_at"),
+            "last_activity_at": doc.get("last_activity_at"),
+            "comment_count": doc.get("comment_count"),
+            "votes_point": doc.get("votes", {}).get("point"),
+            "context": doc.get("context"),
+            "course_id": doc.get("course_id"),
+            "commentable_id": doc.get("commentable_id"),
+            "author_id": doc.get("author_id"),
+            "group_id": doc.get("group_id"),
+            "thread_id": str(doc.get("_id")),
+        }
+
     def insert(
         self,
         title: str,