Merge pull request #1629 from gtech-mulearn/dev

devserv

api/dashboard/roles/dash_roles_serializer.py

@@ -80,3 +80,26 @@ def create(self, validated_data):
         validated_data["created_at"] = DateTimeUtils.get_current_utc_time()
 
         return super().create(validated_data)
+
+class UserRoleBulkAssignSerializer(serializers.ModelSerializer):
+    user_id = serializers.CharField(required=True)
+    role_id = serializers.CharField(required=True)
+    created_by_id = serializers.CharField(required=True, allow_null=False)
+    class Meta:
+        model = UserRoleLink
+        fields = [
+            "id",
+            "user_id",
+            "role_id",
+            "verified",
+            "created_by_id",
+            "created_at",
+        ]
+
+    def to_representation(self, instance):
+        representation = super().to_representation(instance)
+
+        representation['user_id'] = instance.user.fullname if instance.user else None
+        representation['role_id'] = instance.role.title if instance.role else None
+        return representation
+

api/dashboard/roles/dash_roles_views.py

@@ -1,13 +1,18 @@
+import uuid
 from django.db import IntegrityError
 from rest_framework.views import APIView
 
 from db.user import Role, User, UserRoleLink
-from utils.permission import CustomizePermission, role_required
+from utils.permission import CustomizePermission, role_required, JWTUtils
 from utils.response import CustomResponse
 from utils.types import RoleType, WebHookActions, WebHookCategory
-from utils.utils import CommonUtils, DiscordWebhooks
+from utils.utils import CommonUtils, DiscordWebhooks, ImportCSV
 from . import dash_roles_serializer
 
+from openpyxl import load_workbook
+from tempfile import NamedTemporaryFile
+from io import BytesIO
+from django.http import FileResponse
 
 class RoleAPI(APIView):
     authentication_classes = [CustomizePermission]
@@ -202,3 +207,156 @@ def delete(self, request):
         return CustomResponse(
             general_message="User Role deleted successfully"
         ).get_success_response()
+
+class RoleBaseTemplateAPI(APIView):
+    authentication_classes = [CustomizePermission]
+
+    def get(self, request):
+        wb = load_workbook('./api/dashboard/roles/assets/role_base_template.xlsx')
+        ws = wb['Data Definitions']
+
+        roles = Role.objects.all().values_list('title', flat=True)
+        data = {
+            'role': roles
+        }
+        # Write data column-wise
+        for col_num, (col_name, col_values) in enumerate(data.items(), start=1):
+            for row, value in enumerate(col_values, start=2):
+                ws.cell(row=row, column=col_num, value=value)
+
+        # Save the file
+        with NamedTemporaryFile() as tmp:
+            tmp.close() # with statement opened tmp, close it so wb.save can open it
+            wb.save(tmp.name)
+            with open(tmp.name, 'rb') as f:
+                f.seek(0)
+                new_file_object = f.read()
+        return FileResponse(BytesIO(new_file_object), as_attachment=True, filename='role_base_template.xlsx')
+
+class UserRoleBulkAssignAPI(APIView):
+    authentication_classes = [CustomizePermission]
+
+    @role_required([RoleType.ADMIN.value])
+
+    def post(self, request):
+        try:
+            file_obj = request.FILES["user_roles_list"]
+        except KeyError:
+            return CustomResponse(
+                general_message="File not found."
+            ).get_failure_response()
+
+        excel_data = ImportCSV()
+        excel_data = excel_data.read_excel_file(file_obj)
+
+        if not excel_data:
+            return CustomResponse(
+                general_message="Empty csv file."
+            ).get_failure_response()
+
+        temp_headers = [
+            "muid",
+            "role"
+        ]
+        first_entry = excel_data[0]
+        for key in temp_headers:
+            if key not in first_entry:
+                return CustomResponse(
+                    general_message=f"{key} does not exist in the file."
+                ).get_failure_response()
+
+        excel_data = [row for row in excel_data if any(row.values())]
+        valid_rows = []
+        error_rows = []
+
+        users_to_fetch = set()
+        roles_to_fetch = set()
+        user_role_link_to_check = set()
+
+        for row in excel_data[1:]:
+            keys_to_keep = ["muid", "role"]
+            row_keys = list(row.keys())
+
+            # Remove columns other than "muid" and "role"
+            for key in row_keys:
+                if key not in keys_to_keep:
+                    del row[key]
+
+        for row in excel_data[1:]:
+            user = row.get("muid")
+            role = row.get("role")
+            users_to_fetch.add(user)
+            roles_to_fetch.add(role)
+            if (user, role) in user_role_link_to_check:
+                row["error"] = "Duplicate entry"
+                error_rows.append(row)
+                excel_data.remove(row)
+            else:
+                user_role_link_to_check.add((user, role))
+
+        users = User.objects.filter(
+            muid__in=users_to_fetch
+            ).values(
+                "id",
+                "muid",
+            )
+        roles = Role.objects.filter(
+            title__in=roles_to_fetch
+            ).values(
+                "id",
+                "title",
+            )
+        existing_user_role_links = list(UserRoleLink.objects.filter(
+            user__muid__in=users_to_fetch, 
+            role__title__in=roles_to_fetch
+            ).values_list('user__muid', 'role__title'))
+        users_dict = {user["muid"]: user["id"] for user in users}
+        roles_dict = {role["title"]: role["id"] for role in roles}
+
+        for row in excel_data[1:]:
+            user = row.pop("muid")
+            role = row.pop("role")
+
+            user_id = users_dict.get(user) 
+            role_id = roles_dict.get(role)
+
+            if not user_id:
+                row["muid"] = user
+                row["role"] = role
+                row["error"] = f"Invalid user muid: {user}"
+                error_rows.append(row)
+            elif not role_id:
+                row["muid"] = user
+                row["role"] = role
+                row["error"] = f"Invalid role: {role}"
+                error_rows.append(row)
+            elif (user, role) in existing_user_role_links:
+                row["muid"] = user
+                row["role"] = role
+                row["error"] = f"User {user} already has role {role}"
+                error_rows.append(row)
+            else:
+                request_user_id = JWTUtils.fetch_user_id(request)
+                row["id"] = str(uuid.uuid4())
+                row["user_id"] = user_id
+                row["role_id"] = role_id
+                row["verified"] = True
+                row["created_by_id"] = request_user_id
+                valid_rows.append(row)
+
+        user_roles_serializer = dash_roles_serializer.UserRoleBulkAssignSerializer(data=valid_rows, many=True)
+        success_data = []
+        if user_roles_serializer.is_valid():
+            user_roles_serializer.save()
+            for user_role_data in user_roles_serializer.data:
+                    success_data.append({
+                        'user': user_role_data.get('user_id', ''),
+                        'role': user_role_data.get('role_id', ''),
+                })
+        else:
+            error_rows.append(user_roles_serializer.errors)
+
+        return CustomResponse(
+            response={"Success": success_data, "Failed": error_rows}
+        ).get_success_response()
+

api/dashboard/roles/urls.py

@@ -5,6 +5,8 @@
 # app_name will help us do a reverse look-up latter.
 urlpatterns = [
     path('user-role/<str:role_id>/', dash_roles_views.UserRoleSearchAPI.as_view(), name='search-user-role'),
+    path('base-template/', dash_roles_views.RoleBaseTemplateAPI.as_view(), name="roles-base-template"),
+    path('bulk-assign/', dash_roles_views.UserRoleBulkAssignAPI.as_view(), name="user-roles-import"),
     path('user-role/', dash_roles_views.UserRole.as_view(), name='create-delete-user-role'),
     path('', dash_roles_views.RoleAPI.as_view(), name="roles-list"),
     path('', dash_roles_views.RoleAPI.as_view(), name="roles-create"),