Merge pull request #1856 from gtech-mulearn:dev

[FEAT] Bulk role API updates

alter-scripts/alter-1.46.py

@@ -0,0 +1,69 @@
+import os
+import sys
+import uuid
+from decouple import config
+import django
+
+from connection import execute
+
+os.chdir('..')
+sys.path.append(os.getcwd())
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mulearnbackend.settings')
+django.setup()
+
+from utils.utils import DiscordWebhooks
+from utils.types import WebHookActions, WebHookCategory
+
+def is_role_exist(title):
+    query = f"SELECT id FROM role WHERE title = '{title}'"
+    return True if execute(query) else False
+
+def get_intrest_groups():
+    query = f"SELECT name,code, created_by FROM interest_group"
+    return execute(query)
+
+def create_ig_lead_roles():
+    for name,code,created_by in get_intrest_groups():
+        role_name = f"{code} CampusLead"
+        if not is_role_exist(role_name):
+            query = f"""INSERT INTO role (id, title, description ,created_by, updated_by,updated_at,created_at)
+                        VALUES (
+                            '{uuid.uuid4()}',
+                            '{role_name}',
+                            '{f'Campus Lead of {name} Interest Group'}',
+                            '{created_by}',
+                            '{created_by}',
+                            UTC_TIMESTAMP,
+                            UTC_TIMESTAMP
+                        )
+                """
+            execute(query)
+            DiscordWebhooks.general_updates(
+                WebHookCategory.ROLE.value,
+                WebHookActions.CREATE.value,
+                role_name
+            )
+        role_name = f'{code} IGLead'
+        if not is_role_exist(role_name):
+            query = f"""INSERT INTO role (id, title, description ,created_by, updated_by,updated_at,created_at)
+                        VALUES (
+                            '{uuid.uuid4()}',
+                            '{role_name}',
+                            '{f'Interest Group Lead of {name} Interest Group'}',
+                            '{created_by}',
+                            '{created_by}',
+                            UTC_TIMESTAMP,
+                            UTC_TIMESTAMP
+                        )
+                """
+            execute(query)
+            DiscordWebhooks.general_updates(
+                WebHookCategory.ROLE.value,
+                WebHookActions.CREATE.value,
+                role_name
+            )
+
+if __name__ == '__main__':
+    create_ig_lead_roles()
+    execute("UPDATE system_setting SET value = '1.46', updated_at = now() WHERE `key` = 'db.version';")
+

api/dashboard/error_log/error_view.py

@@ -1,16 +1,15 @@
 import logging
 import os
 
-from decouple import config
+from django.conf import settings
 from django.http import FileResponse
 from rest_framework.views import APIView
 
-from api.dashboard.error_log.log_helper import logHandler
 from utils.permission import CustomizePermission, role_required
 from utils.response import CustomResponse
 from utils.types import RoleType
 
-LOG_PATH = config("LOGGER_DIR_PATH")
+from .log_helper import ManageURLPatterns, logHandler
 
 
 class DownloadErrorLogAPI(APIView):
@@ -20,7 +19,7 @@ class DownloadErrorLogAPI(APIView):
         [RoleType.ADMIN.value, RoleType.FELLOW.value, RoleType.TECH_TEAM.value]
     )
     def get(self, request, log_name):
-        error_log = f"{LOG_PATH}/{log_name}.log"
+        error_log = f"{settings.LOG_PATH}/{log_name}.log"
         if os.path.exists(error_log):
             response = FileResponse(
                 open(error_log, "rb"), content_type="application/octet-stream"
@@ -39,7 +38,7 @@ class ViewErrorLogAPI(APIView):
         [RoleType.ADMIN.value, RoleType.FELLOW.value, RoleType.TECH_TEAM.value]
     )
     def get(self, request, log_name):
-        error_log = f"{LOG_PATH}/{log_name}.log"
+        error_log = f"{settings.LOG_PATH}/{log_name}.log"
         if os.path.exists(error_log):
             try:
                 with open(error_log, "r") as log_file:
@@ -62,7 +61,7 @@ class ClearErrorLogAPI(APIView):
         [RoleType.ADMIN.value, RoleType.FELLOW.value, RoleType.TECH_TEAM.value]
     )
     def post(self, request, log_name):
-        error_log = f"{LOG_PATH}/{log_name}.log"
+        error_log = f"{settings.LOG_PATH}/{log_name}.log"
         if os.path.exists(error_log):
             try:
                 with open(error_log, "w") as log_file:
@@ -82,27 +81,166 @@ def post(self, request, log_name):
 
 
 class LoggerAPI(APIView):
+    """
+    API view for logging errors.
+
+    Args:
+        request: The HTTP request object.
+
+    Returns:
+        CustomResponse: The response object containing formatted error logs.
+
+    Raises:
+        IOError: If there is an error reading the error log file.
+
+    Examples:
+        >>> logger_api = LoggerAPI()
+        >>> response = logger_api.get(request)
+    """
+
     authentication_classes = [CustomizePermission]
-    
+
     @role_required(
         [RoleType.ADMIN.value, RoleType.FELLOW.value, RoleType.TECH_TEAM.value]
     )
     def get(self, request):
-        error_log = f"{LOG_PATH}/error.log"
+        """
+        Get the error logs.
+
+        Args:
+            request: The HTTP request object.
+
+        Returns:
+            CustomResponse: The response object containing formatted error logs.
+
+        Raises:
+            IOError: If there is an error reading the error log file.
+
+        Examples:
+            >>> logger_api = LoggerAPI()
+            >>> response = logger_api.get(request)
+        """
+        error_log = f"{settings.LOG_PATH}/error.log"
         try:
             with open(error_log, "r") as file:
                 log_data = file.read()
         except IOError as e:
             return CustomResponse(response=str(e)).get_failure_response()
-        
-        log_handler = logHandler()
-        formatted_errors = log_handler.parse_logs(log_data)
+
+        log_handler = logHandler(log_data)
+        formatted_errors = log_handler.parse_logs()
         return CustomResponse(response=formatted_errors).get_success_response()
-    
+
     @role_required(
         [RoleType.ADMIN.value, RoleType.FELLOW.value, RoleType.TECH_TEAM.value]
     )
     def patch(self, request, error_id):
+        """
+        Patch the error log.
+
+        Args:
+            request: The HTTP request object.
+            error_id: The ID of the error to be marked as patched.
+
+        Returns:
+            CustomResponse: The response object indicating the success of the patch.
+
+        Examples:
+            >>> logger_api = LoggerAPI()
+            >>> response = logger_api.patch(request, error_id)
+        """
         logger = logging.getLogger("django")
         logger.error(f"PATCHED : {error_id}")
         return CustomResponse(response="Updated patch list").get_success_response()
+
+
+class ErrorGraphAPI(APIView):
+    """
+    A class representing the ErrorGraphAPI view.
+
+    This view handles the GET request to retrieve formatted error data including a heatmap of URL hits,
+    incident information, and affected users. It requires authentication and specific roles to access.
+
+    Args:
+        self: The instance of the class itself.
+    """
+
+    authentication_classes = [CustomizePermission]
+
+    @role_required(
+        [RoleType.ADMIN.value, RoleType.FELLOW.value, RoleType.TECH_TEAM.value]
+    )
+    def get(self, request):
+        """
+        Handle the GET request to retrieve formatted error data.
+
+        Returns:
+            CustomResponse: The success response containing the formatted error data.
+
+        Raises:
+            IOError: If an error occurs while reading the error log file.
+
+        """
+        try:
+            error_log = f"{settings.LOG_PATH}/error.log"
+
+            with open(error_log, "r") as file:
+                log_data = file.read()
+
+            log_handler = logHandler(log_data)
+
+            formatted_errors = {
+                "heatmap": log_handler.get_urls_heatmap(),
+                "incident_info": log_handler.get_incident_info(),
+                "affected_users": log_handler.get_affected_users(),
+            }
+
+            return CustomResponse(response=formatted_errors).get_success_response()
+
+        except IOError as e:
+            return CustomResponse(response=str(e)).get_failure_response()
+
+
+class ErrorTabAPI(APIView):
+    """
+    A class representing the ErrorTabAPI view.
+
+    This view handles the GET request to retrieve grouped URL patterns based on user roles.
+    It requires authentication and specific roles to access.
+
+    Args:
+        self: The instance of the class itself.
+    """
+
+    authentication_classes = [CustomizePermission]
+
+    @role_required(
+        [RoleType.ADMIN.value, RoleType.FELLOW.value, RoleType.TECH_TEAM.value]
+    )
+    def get(self, request):
+        """
+        Handle the GET request to retrieve grouped URL patterns.
+
+        Returns:
+            CustomResponse: The success response containing the grouped URL patterns.
+
+        Raises:
+            IOError: If an error occurs while retrieving the URL patterns.
+
+        """
+        try:
+            error_log = f"{settings.LOG_PATH}/error.log"
+
+            with open(error_log, "r") as file:
+                log_data = file.read()
+
+            log_handler = logHandler(log_data)
+            parsed_errors = log_handler.parse_logs()
+
+            urlpatterns = ManageURLPatterns().urlpatterns
+            grouped_patterns = ManageURLPatterns.group_patterns(urlpatterns)
+
+            return CustomResponse(response=parsed_errors).get_success_response()
+
+        except IOError as e:
+            return CustomResponse(response=str(e)).get_failure_response()