feat: 添加系统级管理专属页面权限

annotation/auth.py

@@ -9,6 +9,7 @@ from functools import wraps
 
 from fastapi import Request
 
+from config.constant import RedisKeyConfig
 from controller.login import LoginController
 from exceptions.exception import PermissionException
 
@@ -53,3 +54,16 @@ async def hasAuth(request: Request, permission: str) -> bool:
         return True
     else:
         return False
+
+
+async def hasAdmin(request: Request, department_id: str) -> bool:
+    """
+    判断是否有管理员权限
+    """
+    permissions = []
+    if ids := await request.app.state.redis.get(f'{RedisKeyConfig.SYSTEM_CONFIG.key}:permission_departments'):
+        permissions = eval(ids)
+    if department_id in permissions:
+        return True
+    else:
+        return False

api/permission.py

@@ -10,7 +10,7 @@ from typing import Optional
 from fastapi import APIRouter, Depends, Path, Query, Request
 from fastapi.responses import JSONResponse
 
-from annotation.auth import Auth
+from annotation.auth import Auth, hasAdmin
 from annotation.log import Log
 from config.constant import BusinessType, RedisKeyConfig
 from controller.login import LoginController
@@ -51,6 +51,7 @@ async def add_permission(request: Request, params: AddPermissionParams):
         leave_transition=params.leave_transition,
         fixed_tag=params.fixed_tag,
         hidden_tag=params.hidden_tag,
+        is_admin=params.is_admin
     )
     if permission:
         # 更新用户信息缓存
@@ -114,6 +115,7 @@ async def update_permission(request: Request, params: AddPermissionParams, id: s
         permission.leave_transition = params.leave_transition
         permission.fixed_tag = params.fixed_tag
         permission.hidden_tag = params.hidden_tag
+        permission.is_admin = params.is_admin
         await permission.save()
         # 更新用户信息缓存
         userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*')
@@ -161,6 +163,7 @@ async def get_permission(request: Request, id: str = Path(description="权限ID"
             fixed_tag="fixed_tag",
             show_link="show_link",
             show_parent="show_parent",
+            is_admin="is_admin"
         )
         return Response.success(msg="查询权限详情成功！", data=permission)
     else:
@@ -195,7 +198,9 @@ async def get_permission_list(
         enterTransition: Optional[str] = Query(default=None, description="进场动画"),
         leaveTransition: Optional[str] = Query(default=None, description="离场动画"),
         fixedTag: Optional[bool] = Query(default=None, description="固定标签页"),
-        hiddenTag: Optional[bool] = Query(default=None, description="隐藏标签页")
+        hiddenTag: Optional[bool] = Query(default=None, description="隐藏标签页"),
+        isAdmin: Optional[bool] = Query(default=None, description="是否为管理专属页面"),
+        current_user: dict = Depends(LoginController.get_current_user),
 ):
     filterArgs = {
         f'{k}__contains': v for k, v in {
@@ -219,9 +224,13 @@ async def get_permission_list(
             "enter_transition": enterTransition,
             "leave_transition": leaveTransition,
             "fixed_tag": fixedTag,
-            "hidden_tag": hiddenTag
+            "hidden_tag": hiddenTag,
+            "is_admin": isAdmin
         }.items() if v
     }
+    department_id = current_user.get("department_id", "")
+    if not await hasAdmin(request, department_id):
+        filterArgs["is_admin"] = False
     total = await Permission.filter(**filterArgs, del_flag=1).count()
     result = await Permission.filter(**filterArgs, del_flag=1).offset((page - 1) * pageSize).limit(pageSize).order_by(
         'rank').values(
@@ -250,7 +259,8 @@ async def get_permission_list(
         hidden_tag="hidden_tag",
         fixed_tag="fixed_tag",
         show_link="show_link",
-        show_parent="show_parent"
+        show_parent="show_parent",
+        is_admin="is_admin"
     )
     return Response.success(data={
         "total": total,

api/role.py

@@ -10,7 +10,7 @@ from typing import Optional
 from fastapi import APIRouter, Depends, Path, Query, Request
 from fastapi.responses import JSONResponse
 
-from annotation.auth import Auth, hasAuth
+from annotation.auth import Auth, hasAuth, hasAdmin
 from annotation.log import Log
 from config.constant import BusinessType, RedisKeyConfig
 from controller.login import LoginController
@@ -235,6 +235,11 @@ async def add_role_permission(request: Request, params: AddRolePermissionParams,
                               id: str = Path(..., description="角色ID"),
                               current_user: dict = Depends(LoginController.get_current_user)):
     sub_departments = current_user.get("sub_departments")
+    if await hasAdmin(request, current_user.get("department_id")):
+        department_permissions = await Permission.filter(del_flag=1).values("id")
+    else:
+        department_permissions = await Permission.filter(is_admin=False, del_flag=1).values("id")
+    department_permissions = filterKeyValues(department_permissions, "id")
     if role := await Role.get_or_none(id=id, del_flag=1, department__id__in=sub_departments):
         # 已有角色权限
         rolePermissions = await RolePermission.filter(role_id=id, del_flag=1).values("permission_id")
@@ -243,6 +248,8 @@ async def add_role_permission(request: Request, params: AddRolePermissionParams,
         add_list = set(params.permission_ids).difference(set(rolePermissions))
         # 循环添加角色权限
         for item in add_list:
+            if item not in department_permissions:
+                continue
             permission = await Permission.get_or_none(id=item, del_flag=1)
             if permission:
                 await RolePermission.create(
@@ -294,6 +301,11 @@ async def update_role_permission(request: Request, params: AddRolePermissionPara
                                  id: str = Path(..., description="角色ID"),
                                  current_user: dict = Depends(LoginController.get_current_user)):
     sub_departments = current_user.get("sub_departments")
+    if await hasAdmin(request, current_user.get("department_id")):
+        department_permissions = await Permission.filter(del_flag=1).values("id")
+    else:
+        department_permissions = await Permission.filter(is_admin=False, del_flag=1).values("id")
+    department_permissions = filterKeyValues(department_permissions, "id")
     if role := await Role.get_or_none(id=id, del_flag=1, department__id__in=sub_departments):
         # 已有角色权限
         rolePermissions = await RolePermission.filter(role_id=role.id, del_flag=1).values("permission_id")
@@ -307,6 +319,8 @@ async def update_role_permission(request: Request, params: AddRolePermissionPara
             await RolePermission.filter(role_id=id, permission_id=item, del_flag=1).update(del_flag=0)
         # 循环添加角色权限
         for item in add_list:
+            if item not in department_permissions:
+                continue
             await RolePermission.create(role_id=id, permission_id=item)
         # 更新用户信息缓存
         userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*')

api/user.py

@@ -9,7 +9,7 @@ import os
 from datetime import datetime
 from typing import Optional
 
-from fastapi import APIRouter, Depends, Path, Query, UploadFile, File, Request
+from fastapi import APIRouter, Depends, Path, Query, UploadFile, File, Request, Form
 from fastapi.responses import JSONResponse
 
 from annotation.auth import Auth
@@ -27,7 +27,7 @@ from schemas.department import GetDepartmentListResponse
 from schemas.file import UploadFileResponse
 from schemas.user import AddUserParams, GetUserListResponse, GetUserInfoResponse, UpdateUserParams, \
     AddUserRoleParams, GetUserRoleInfoResponse, UpdateUserRoleParams, GetUserPermissionListResponse, \
-    ResetPasswordParams
+    ResetPasswordParams, UpdateBaseUserInfoParams
 from utils.common import filterKeyValues
 from utils.password import Password
 from utils.response import Response
@@ -434,3 +434,82 @@ async def reset_user_password(request: Request, params: ResetPasswordParams, id:
         await user.save()
         return Response.success(msg="重置密码成功！")
     return Response.failure(msg="用户不存在!")
+
+
+@userAPI.put("/updateBaseUserInfo", response_model=BaseResponse, response_class=JSONResponse,
+             summary="更新基础个人信息")
+@userAPI.post("/updateBaseUserInfo", response_model=BaseResponse, response_class=JSONResponse,
+              summary="更新基础个人信息")
+@Log(title="更新基础个人信息", business_type=BusinessType.UPDATE)
+async def update_base_userinfo(params: UpdateBaseUserInfoParams, request: Request,
+                               current_user: dict = Depends(LoginController.get_current_user)):
+    user = await User.get_or_none(id=current_user.get("id"), del_flag=1)
+    if user:
+        user.nickname = params.name
+        user.gender = params.gender
+        await user.save()
+        if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:{user.id}'):
+            await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:{user.id}')
+        return Response.success(msg="更新成功！")
+    return Response.error(msg="更新失败！")
+
+
+@userAPI.put("/updatePassword", response_class=JSONResponse, response_model=BaseResponse, summary="用户更新密码")
+@userAPI.post("/updatePassword", response_class=JSONResponse, response_model=BaseResponse, summary="用户更新密码")
+@Log(title="用户更新密码", business_type=BusinessType.UPDATE)
+async def update_user_password(request: Request, oldPassword: str = Form(description="用户旧密码"),
+                               newPassword: str = Form(description="用户新密码"),
+                               current_user: dict = Depends(LoginController.get_current_user)):
+    if user := await User.get_or_none(id=current_user.get("id"), del_flag=1):
+        password = await Password.get_password_hash(oldPassword)
+        if user.password != password:
+            return Response.error(msg="旧密码错误！")
+        newPassword = await Password.get_password_hash(newPassword)
+        user.password = newPassword
+        await user.save()
+        if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:{user.id}'):
+            await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:{user.id}')
+        return Response.success(msg="更新成功！")
+    return Response.error(msg="更新失败！")
+
+
+@userAPI.put("/updatePhone", response_class=JSONResponse, response_model=BaseResponse, summary="用户更新手机号")
+@userAPI.post("/updatePhone", response_class=JSONResponse, response_model=BaseResponse, summary="用户更新手机号")
+@Log(title="用户更新手机号", business_type=BusinessType.UPDATE)
+async def update_user_phone(request: Request, password: str = Form(description="用户密码"),
+                            phone: str = Form(description="用户手机号"),
+                            current_user: dict = Depends(LoginController.get_current_user)):
+    if user := await User.get_or_none(id=current_user.get("id"), del_flag=1):
+        password = await Password.get_password_hash(password)
+        if user.password != password:
+            return Response.error("更改失败，请正确输入旧密码")
+        phoneStatus = await User.filter(phone=phone, del_flag=1).count()
+        if phoneStatus:
+            return Response.error(f"更改失败，手机号:{phone}已绑定其他账号！")
+        user.phone = phone
+        await user.save()
+        if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:{user.id}'):
+            await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:{user.id}')
+        return Response.success(msg="更新成功！")
+    return Response.error(msg="更新失败！")
+
+
+@userAPI.put("/updateEmail", response_class=JSONResponse, response_model=BaseResponse, summary="用户更新邮箱")
+@userAPI.post("/updateEmail", response_class=JSONResponse, response_model=BaseResponse, summary="用户更新邮箱")
+@Log(title="用户更新邮箱", business_type=BusinessType.UPDATE)
+async def update_user_email(request: Request, password: str = Form(description="用户密码"),
+                            email: str = Form(description="用户邮箱"),
+                            current_user: dict = Depends(LoginController.get_current_user)):
+    if user := await User.get_or_none(id=current_user.get("id"), del_flag=1):
+        password = await Password.get_password_hash(password)
+        if user.password != password:
+            return Response.error("更改失败，请正确输入旧密码")
+        emailStatus = await User.filter(email=email, del_flag=1).count()
+        if emailStatus:
+            return Response.error(f"更改失败，邮箱:{email}已绑定其他账号！")
+        user.email = email
+        await user.save()
+        if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:{user.id}'):
+            await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:{user.id}')
+        return Response.success(msg="更新成功！")
+    return Response.error(msg="更新失败！")

config/database.py

@@ -5,9 +5,13 @@
 # @File : database.py
 # @Software : PyCharm
 # @Comment : 本程序
+import asyncio
 import logging
+import subprocess
 import sys
+from datetime import datetime
 from logging.handlers import RotatingFileHandler
+from pathlib import Path
 
 from tortoise import Tortoise
 
@@ -136,3 +140,41 @@ async def configure_tortoise_logging(enable_logging: bool = True, log_level: int
     else:
         # 如果禁用日志，设置日志级别为 WARNING 以抑制大部分输出
         tortoise_logger.setLevel(logging.WARNING)
+
+
+async def backup_database():
+    """
+    备份数据库
+    """
+    logger.info("开始备份数据库")
+    # 配置数据库连接信息
+    backup_dir = Path().cwd() / "sql"  # 备份文件存储的目录
+
+    # 如果 migrations 目录不存在，则创建
+    backup_dir.mkdir(parents=True, exist_ok=True)
+
+    # 生成备份文件名，格式为 dbYYYYMMDDHHMMSS.sql
+    timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
+    backup_filename = f"db{timestamp}.sql"
+    backup_filepath = backup_dir / backup_filename  # 使用 Path 对象组合路径
+
+    # 构造 mysqldump 命令
+    command = [
+        "mysqldump",
+        "-u", DataBaseConfig.db_username,
+        f"-p{DataBaseConfig.db_password}",  # 直接传递密码
+        DataBaseConfig.db_database,
+        "--result-file=" + str(backup_filepath)  # 指定备份文件路径
+    ]
+
+    # 使用 asyncio.to_thread 来在线程中执行阻塞操作
+    await asyncio.to_thread(run_mysqldump, command)
+
+
+def run_mysqldump(command):
+    """在阻塞线程中执行 mysqldump 命令"""
+    try:
+        subprocess.run(command, check=True)
+        logger.info(f"数据库备份已完成，文件保存为 {command[-1]}")
+    except subprocess.CalledProcessError as e:
+        logger.error(f"备份失败，错误: {e}")

models/permission.py

@@ -280,6 +280,18 @@ class Permission(BaseModel):
     - 映射到数据库字段 show_parent。
     """
 
+    is_admin = fields.BooleanField(
+        default=False,
+        description="是否为管理专属页面",
+        source_field="is_admin"  # 映射到数据库字段 is_admin
+    )
+    """
+    是否为管理专属页面。
+    - 是否为管理专属页面，仅管理员可见。
+    - 默认为 False。
+    - 映射到数据库字段 is_admin。
+    """
+
     class Meta:
         table = "permission"  # 数据库表名
         table_description = "权限表"  # 表描述

schemas/permission.py

@@ -43,6 +43,7 @@ class PermissionInfo(BaseModel):
     fixed_tag: bool = Field(default=False, description="固定标签页")
     show_link: bool = Field(default=True, description="显示菜单")
     show_parent: bool = Field(default=True, description="显示父级菜单")
+    is_admin: bool = Field(default=False, description="是否为管理专属页面")
 
     class Config:
         json_schema_extra = {
@@ -72,7 +73,8 @@ class PermissionInfo(BaseModel):
                 "hidden_tag": False,
                 "fixed_tag": False,
                 "show_link": True,
-                "show_parent": True
+                "show_parent": True,
+                "is_admin": False
             }
         }
 
@@ -109,6 +111,7 @@ class AddPermissionParams(BaseModel):
     show_parent: bool = Field(default=True, description="显示父级菜单")
     parent_id: str = Field(default="", max_length=36, description="父级菜单ID")
     menu_type: int = Field(default=0, description="菜单类型")
+    is_admin: bool = Field(default=False, description="是否为管理专属页面")
 
     class Config:
         json_schema_extra = {
@@ -133,7 +136,8 @@ class AddPermissionParams(BaseModel):
                 "show_link": True,
                 "show_parent": True,
                 "parent_id": "",
-                "menu_type": 0
+                "menu_type": 0,
+                "is_admin": False
             }
         }
 

schemas/user.py

@@ -6,6 +6,7 @@
 # @Software : PyCharm
 # @Comment : 本程序
 from datetime import datetime
+from enum import IntEnum
 from typing import Optional, List
 from uuid import UUID
 
@@ -15,6 +16,11 @@ from pydantic_validation_decorator import Xss, NotBlank, Size, Network
 from schemas.common import BaseResponse, ListQueryResult
 
 
+class Gender(IntEnum):
+    MAN = 0
+    WOMAN = 1
+
+
 class UserBase(BaseModel):
     """
     用户表基础模型。
@@ -387,3 +393,19 @@ class GetUserStatisticsResponse(BaseResponse):
     获取用户统计信息响应模型。
     """
     data: GetUserStatisticsResult = Field(default=None, description="响应数据")
+
+
+class UpdateBaseUserInfoParams(BaseModel):
+    """修改基础信息参数"""
+    name: str
+    """姓名"""
+    gender: Gender
+    """性别"""
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "name": "张三",
+                "gender": 1,
+            }
+        }