123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260 |
- import json
- import uuid
- from datetime import timedelta
- from django.core.exceptions import ObjectDoesNotExist
- from django.core.paginator import Paginator
- from django.db import transaction
- from django.shortcuts import get_object_or_404
- from DCbackend.utils.common import success, fail, pageData
- from DCbackend.utils.token import encrypt, decrypt
- from backend.Form.UserForm import UserForm
- from backend.Service.AdminRoleService import AdminRoleService
- from backend.Service.AuthMenuService import AuthMenuService
- from backend.models import Admin, AdminRole
- from django.core.cache import cache
- class AdminService:
- # 获取管理员信息
- @staticmethod
- def getAdminInfo(request):
- adminId = request.POST.get('id')
- adminInfo = Admin.objects.filter(id=adminId).first()
- if adminInfo is not None:
- return success(AdminService.getAdminDetail(adminInfo))
- else:
- return fail("管理员信息不存在")
- # return success(Admin.objects.filter(pk=request.POST.get('id', 0)).first())
- # 输出
- @staticmethod
- def getAdminDetail(admin: Admin):
- role = AdminRole.objects.filter(id=admin.role_id).first()
- roleInfo = {}
- if role:
- roleInfo = AdminRoleService.getDetail(role)
- #菜單
- results = AuthMenuService.menuInfo(admin.role_id)
- data = {
- 'id': admin.id,
- 'username': admin.username,
- 'email': admin.email,
- 'mobile': admin.mobile,
- 'realName': admin.real_name,
- 'role_id': admin.role_id,
- 'roleInfo': roleInfo,
- 'token': admin.token,
- 'createTime': admin.create_time,
- 'status': admin.status,
- 'avatar': admin.header,
- 'job': admin.job,
- 'code': admin.code,
- 'contact': admin.contact,
- 'introduction': '',
- 'roles': "[admin]",
- 'authList': results
- }
- return data
- # 创建管理员
- @staticmethod
- def create(request):
- form = UserForm(request.POST)
- if form.is_valid():
- form.instance.password = encrypt(form.instance.password)
- form.save()
- return success("保存成功")
- else:
- return fail(form.errors)
- @staticmethod
- def update(request):
- data = Admin.objects.filter(id=request.POST.get("id")).first()
- if data is not None:
- if request.POST.get("password"):
- data.password = encrypt(request.POST.get("password"))
- data.email = request.POST.get("email")
- data.mobile = request.POST.get("mobile")
- data.status = request.POST.get("status")
- data.real_name = request.POST.get("real_name")
- data.role_id = request.POST.get("role_id")
- data.code = request.POST.get("code")
- data.header = request.POST.get("header")
- data.job = request.POST.get("job")
- data.contact = request.POST.get("contact")
- data.save()
- return success("成功")
- else:
- return fail("更新用户信息失败")
- @staticmethod
- def updateProfile(request):
- data = AdminService.verify_token(request.POST.get("token"))
- if data is not None:
- data.email = request.POST.get("email")
- data.real_name = request.POST.get("realName")
- data.code = request.POST.get("code")
- data.header = request.POST.get("header")
- data.job = request.POST.get("job")
- data.contact = request.POST.get("contact")
- data.save()
- return success("成功")
- else:
- return fail("更新用户信息失败")
- @staticmethod
- def changePassword(request):
- oldPassword = request.POST.get("oldPassword")
- newPassword = request.POST.get("newPassword")
- confirmPassword = request.POST.get("confirmPassword")
- if newPassword!=confirmPassword:
- return fail("两次填写的密码不相同")
- data = AdminService.verify_token(request.POST.get("token"))
- if data is not None:
- if oldPassword != decrypt(data.password):
- return fail("原密码不正确")
- data.password=encrypt(newPassword)
- data.save()
- return success("成功")
- else:
- return fail("更新用户信息失败")
- # 通过token获取管理员信息
- @staticmethod
- def getAdminByToken(request):
- adminInfo = AdminService.verify_token(request.POST.get("token"))
- if adminInfo is not None:
- return success(AdminService.getAdminDetail(adminInfo))
- else:
- return fail("用户信息不存在")
- # 账户登录
- @staticmethod
- def login1(request):
- username = request.POST.get("username", "")
- password = request.POST.get("password", "")
- adminInfo = Admin.objects.filter(username=username).first()
- if adminInfo is not None:
- if password == decrypt(adminInfo.password):
- adminInfo.token = uuid.uuid4().hex
- adminInfo.save()
- return success({"token": adminInfo.token})
- else:
- return fail("登录密码不正确")
- else:
- return fail("登录失败,管理员用户名不存在")
- @staticmethod
- def login(request):
- username = request.POST.get("username", "")
- password = request.POST.get("password", "")
- adminInfo = Admin.objects.filter(username=username).first()
- if adminInfo is not None:
- if password == decrypt(adminInfo.password):
- # 检查是否存在有效的 token
- existing_token = AdminService.get_existing_token(adminInfo.id)
- if existing_token:
- # 如果存在有效的 token,直接返回
- return success({"token": existing_token})
- else:
- # 如果不存在有效的 token,生成新的
- new_token = uuid.uuid4().hex
- cache.set(f"admin_token:{new_token}", adminInfo.id, timeout=60 * 60 * 24)
- cache.set(f"admin_id_to_token:{adminInfo.id}", new_token, timeout=60 * 60 * 24)
- return success({"token": new_token})
- else:
- return fail("登录密码不正确")
- else:
- return fail("登录失败,管理员用户名不存在")
- @staticmethod
- def verify_token(token):
- admin_id = cache.get(f"admin_token:{token}")
- if admin_id:
- return Admin.objects.filter(id=admin_id).first()
- return None
- @staticmethod
- def get_existing_token(admin_id):
- # 尝试从缓存中获取该管理员的 token
- token = cache.get(f"admin_id_to_token:{admin_id}")
- if token:
- # 如果找到 token,检查它是否仍然有效
- if cache.get(f"admin_token:{token}"):
- return token
- return None
- # 账户登出
- @staticmethod
- def logout(request):
- token = request.POST.get("token")
- adminInfo = AdminService.verify_token(token)
- if adminInfo is not None:
- # Invalidate the token in the cache
- cache.delete(f"admin_token:{token}")
- cache.delete(f"admin_id_to_token:{adminInfo.id}")
- return success({"result": True})
- else:
- return fail("登出失败,token无效")
- # 搜索管理员列表
- @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 = {}
- username = request.POST.get("username", '')
- if username:
- where["username__contains"] = username
- mobile = request.POST.get("mobile", '')
- if mobile:
- where["mobile__contains"] = mobile
- status = request.POST.get("status")
- if status:
- where['status'] = status
- paginator = Paginator(Admin.objects.filter(**where).exclude(status=4).order_by("-id"), pageSize) # 每页显示10条数据
- page_obj = paginator.get_page(page)
- dataList = []
- for item in page_obj:
- adminInfo = AdminService.getAdminDetail(item)
- dataList.append(adminInfo)
- return success(pageData(page, pageSize, paginator.num_pages, paginator.count, dataList))
- @staticmethod
- @transaction.atomic
- def delete(request):
- user_id = request.POST.get("id")
- if not user_id:
- return fail("User ID is required")
- try:
- # 尝试获取用户对象
- user = get_object_or_404(Admin, id=user_id)
- # 更新用户状态为逻辑删除(status=4)
- updated_count = Admin.objects.filter(id=user_id).update(status=4)
- if updated_count == 0:
- return fail("指定的用户不存在")
- return success("用户已成功删除")
- except ObjectDoesNotExist:
- return fail("指定的用户不存在")
- except Exception as e:
- return fail(f"删除用户时发生错误: {str(e)}")
|