import json from enum import Enum from django.core.paginator import Paginator from django.db import transaction from django.forms import model_to_dict from oss2.exceptions import status from DCbackend.utils.common import success, fail, pageData from backend.Form.AdminRoleForm import AdminRoleForm from backend.Service.AuthMenuService import AuthMenuService from backend.models import AdminRole, AuthUserMenu, TemplateCateGoryUser, Admin, DocumentCategoryUser, DocumentCategory, \ SourceDataCategoryUser, SourceCategory, UserOperation, Operation from django.db.models import F, Case, When, IntegerField, Exists, OuterRef class AdminRoleService: @staticmethod def Create(request): name = request.POST.get("name") adminRola = AdminRole.objects.filter(name=name).exclude(status=4).exists() if adminRola: return fail("已存在同名,请修改后重试。") form = AdminRoleForm(request.POST) authList = request.POST.get("authList") template_cate_auth = request.POST.get("template_cate_auth") doc_cate_auth = request.POST.get("doc_cate_auth") source_cate_auth = request.POST.get("source_cate_auth") if form.is_valid(): admin_role = form.save() # 获取保存后生成的 ID generated_id = admin_role.id if generated_id is None: return fail("保存失败:无法获取生成的ID") # 保存菜单信息 if authList: auth_list = json.loads(authList) for menu in auth_list: AuthUserMenu.objects.create( rola_id=generated_id, menu_id=menu.get("value") ) if template_cate_auth: template_cate_auth_list = json.loads(template_cate_auth) for auth in template_cate_auth_list: UserOperation.objects.create( role_id=generated_id, operation_id=auth.get("operation_id") ) if doc_cate_auth: doc_cate_auth_list = json.loads(doc_cate_auth) for auth in doc_cate_auth_list: UserOperation.objects.create( role_id=generated_id, operation_id=auth.get("operation_id") ) if source_cate_auth: source_cate_auth_list = json.loads(source_cate_auth) for auth in source_cate_auth_list: UserOperation.objects.create( role_id=generated_id, operation_id=auth.get("operation_id") ) return success({"id": generated_id}) return fail(form.errors) @staticmethod @transaction.atomic def Update(request): try: data = AdminRole.objects.filter(id=request.POST.get("id")).first() authList = request.POST.get("authList") template_cate_auth = request.POST.get("template_cate_auth") doc_cate_auth = request.POST.get("doc_cate_auth") source_cate_auth = request.POST.get("source_cate_auth") if data is not None: data.name = request.POST.get("name") data.intro = request.POST.get("intro") data.status = request.POST.get("status") data.is_admin = request.POST.get("is_admin") data.save() generated_id = request.POST.get("id") # 删除旧的菜单信息 AuthUserMenu.objects.filter(rola_id=generated_id).delete() UserOperation.objects.filter(role_id=generated_id).delete() # 保存菜单信息 if authList: auth_list = json.loads(authList) for menu in auth_list: AuthUserMenu.objects.create( rola_id=generated_id, menu_id=menu.get("value") ) if template_cate_auth: template_cate_auth_list = json.loads(template_cate_auth) for auth in template_cate_auth_list: UserOperation.objects.create( role_id=generated_id, operation_id=auth.get("operation_id") ) if doc_cate_auth: doc_cate_auth_list = json.loads(doc_cate_auth) for auth in doc_cate_auth_list: UserOperation.objects.create( role_id=generated_id, operation_id=auth.get("operation_id") ) if source_cate_auth: source_cate_auth_list = json.loads(source_cate_auth) for auth in source_cate_auth_list: UserOperation.objects.create( role_id=generated_id, operation_id=auth.get("operation_id") ) return success("成功") else: return fail("更新参数信息失败") except Exception as e: return fail(str(e)) @staticmethod def Info(request): info = AdminRole.objects.filter(id=request.POST.get("id")).first() return success(AdminRoleService.getDetail(info)) @staticmethod def Delete(request): AdminRole.objects.filter(id=request.POST.get("id")).update(status=4) return success("成功") @staticmethod def getDetail(item: AdminRole): return { "id": item.id, 'name': item.name, 'intro': item.intro, 'is_admin': item.is_admin, 'createTime': item.create_time, 'status': item.status } @staticmethod def Search(request): page = request.POST.get("page") pageSize = request.POST.get("pageSize") if page is None: page = 1 if pageSize is None: pageSize = 1 where = {} name = request.POST.get("name", '') if name: where["name__icontains"] = name isAdmin = request.POST.get("is_admin") if isAdmin: where['is_admin'] = isAdmin status = request.POST.get("status") if status: where['status'] = status paginator = Paginator(AdminRole.objects.filter(**where).exclude(status=4).order_by("-id"), pageSize) # 每页显示10条数据 page_obj = paginator.get_page(page) dataList = [] for item in page_obj: info = AdminRoleService.getDetail(item) dataList.append(info) return success(pageData(page, pageSize, paginator.num_pages, paginator.count, dataList)) @staticmethod def selectInfo(request): id = request.POST.get("id") if id is None: return fail("id为空") #查询角色信息 info = AdminRole.objects.filter(id=id).first() info_dist = model_to_dict(info) if info else None #查询菜单 menuList = AuthMenuService.menuInfo(id) or [] role_id = id # 使用通用函数获取不同 content_type_id 的操作列表 # 1模板分类 template_category_users_list = get_operations_with_type(1, role_id) # 2文档分类 doc_cate_auth = get_operations_with_type(2, role_id) # 3原始数据分类 source_cate_auth = get_operations_with_type(3, role_id) data = { "info": info_dist, "menuList": menuList, "templateCategoryUsers": template_category_users_list, "doc_cate_auth": doc_cate_auth, "source_cate_auth": source_cate_auth } return success(data) def get_operations_with_type(content_type_id, role_id): operations = Operation.objects.filter(content_type_id=content_type_id) user_operation_exists = UserOperation.objects.filter( role_id=role_id, operation_id=OuterRef('pk') ) operations = operations.annotate( type=Case( When(Exists(user_operation_exists), then=1), default=0, output_field=IntegerField() ) ).values('id', 'content_type_id', 'name', 'type', 'codename') # 执行查询并获取结果 result = list(operations) return result