fix: 设置和修改用户组和角色时根据权限不同去修改数据

src/backend/bisheng/api/v1/user.py

@@ -145,7 +145,7 @@ async def login(*, user: UserLogin, Authorize: AuthJWT = Depends()):
     # 判断下密码是否长期未修改
     if password_conf.password_valid_period and password_conf.password_valid_period > 0:
         if (datetime.now() -
-                db_user.password_update_time).days >= password_conf.password_valid_period:
+            db_user.password_update_time).days >= password_conf.password_valid_period:
             return UserPasswordExpireError.return_resp()
 
     access_token, refresh_token, role = gen_user_jwt(db_user)
@@ -418,32 +418,41 @@ async def delete_role(*, role_id: int, Authorize: AuthJWT = Depends()):
 
 
 @router.post('/user/role_add', status_code=200)
-async def user_addrole(*, userRole: UserRoleCreate, Authorize: AuthJWT = Depends()):
-    Authorize.jwt_required()
-    if 'admin' != json.loads(Authorize.get_jwt_subject()).get('role'):
-        raise HTTPException(status_code=500, detail='无设置权限')
+async def user_addrole(*, user_role: UserRoleCreate, login_user: UserPayload = Depends(get_login_user)):
+    """
+    重新设置用户的角色。根据权限不同改动的数据范围不同
+    """
+    # 获取用户的之前的角色列表
+    old_roles = UserRoleDao.get_user_roles(user_role.user_id)
+    old_roles = [one.role_id for one in old_roles]
 
-    with session_getter() as session:
-        db_role = session.exec(select(UserRole).where(
-            UserRole.user_id == userRole.user_id, )).all()
-    role_ids = {role.role_id for role in db_role}
-    db_roles = []
-    for role_id in userRole.role_id:
-        if role_id not in role_ids:
-            db_role = UserRole(user_id=userRole.user_id, role_id=role_id)
-            db_roles.append(db_role)
+    if not login_user.is_admin():
+        # 判断拥有哪些用户组的管理权限
+        admin_group = UserGroupDao.get_user_admin_group(login_user.user_id)
+        admin_group = [one.group_id for one in admin_group]
+        if not admin_group:
+            raise HTTPException(status_code=500, detail='无权限')
+        # 获取管理组下的所有角色列表
+        admin_roles = RoleDao.get_role_by_groups(admin_group, '', 0, 0)
+        for i in range(len(old_roles) - 1, -1, -1):
+            if old_roles[i] not in admin_roles:
+                del old_roles[i]
+        if not old_roles:
+            raise HTTPException(status_code=500, detail='无权限')
+
+    need_add_role = []
+    for one in user_role.role_id:
+        if one not in old_roles:
+            # 需要新增的角色
+            need_add_role.append(one)
         else:
-            role_ids.remove(role_id)
-    if db_roles:
-        with session_getter() as session:
-            session.add_all(db_roles)
-            session.commit()
-    if role_ids:
-        with session_getter() as session:
-            session.exec(
-                delete(UserRole).where(UserRole.user_id == userRole.user_id,
-                                       UserRole.role_id.in_(role_ids)))
-            session.commit()
+            # 剩余的就是需要删除的角色列表
+            old_roles.remove(one)
+    if need_add_role:
+        UserRoleDao.add_user_roles(user_role.user_id, need_add_role)
+    if old_roles:
+        # 删除对应的角色列表
+        UserRoleDao.delete_user_roles(user_role.user_id, old_roles)
     return resp_200()
 
 
@@ -591,7 +600,7 @@ async def knowledge_list(*,
             'id': access[0].id
         } for access in db_role_access],
         'total':
-        total_count
+            total_count
     })
 
 
@@ -640,7 +649,7 @@ async def flow_list(*,
             'id': access[0]
         } for access in db_role_access],
         'total':
-        total_count
+            total_count
     })
 
 

src/backend/bisheng/database/models/user_group.py

@@ -71,6 +71,18 @@ def insert_user_group(cls, user_group: UserGroupCreate) -> UserGroup:
             session.refresh(user_group)
             return user_group
 
+    @classmethod
+    def insert_user_group_admin(cls, user_id: int, group_id: int) -> UserGroup:
+        """
+        将用户设置为组管理员
+        """
+        with session_getter() as session:
+            user_group = UserGroup(user_id=user_id, group_id=group_id, is_group_admin=True)
+            session.add(user_group)
+            session.commit()
+            session.refresh(user_group)
+            return user_group
+
     @classmethod
     def delete_user_group(cls, user_id: int, group_id: int) -> None:
         with session_getter() as session:
@@ -81,17 +93,28 @@ def delete_user_group(cls, user_id: int, group_id: int) -> None:
             session.commit()
 
     @classmethod
-    def replace_user_groups(cls, user_id: int, group_ids: List[int]):
+    def delete_user_groups(cls, user_id: int, group_ids: List[int]):
         """
-        修改用户所属的用户组
+        将用户从某些组中移除
         """
         with session_getter() as session:
             # 先把旧的用户组全部清空
-            statement = delete(UserGroup).where(UserGroup.user_id == user_id).where(UserGroup.is_group_admin == 0)
+            statement = delete(UserGroup).where(
+                UserGroup.user_id == user_id).where(
+                UserGroup.is_group_admin == 0).where(
+                UserGroup.group_id.in_(group_ids)
+            )
             session.exec(statement)
-            # 再把新的用户组添加
-            for one in group_ids:
-                user_group = UserGroup(user_id=user_id, group_id=one, is_group_admin=False)
+            session.commit()
+
+    @classmethod
+    def add_user_groups(cls, user_id: int, group_ids: List[int]):
+        """
+        将用户添加到某些组
+        """
+        with session_getter() as session:
+            for group_id in group_ids:
+                user_group = UserGroup(user_id=user_id, group_id=group_id, is_group_admin=0)
                 session.add(user_group)
             session.commit()
 
@@ -168,5 +191,3 @@ def delete_group_admins(cls, group_id: int, admin_ids: List[int]) -> None:
                 UserGroup.is_group_admin == 1)
             session.exec(statement)
             session.commit()
-
-

src/backend/bisheng/database/models/user_role.py

@@ -4,9 +4,11 @@
 from bisheng.database.base import session_getter
 from bisheng.database.models.base import SQLModelSerializable
 from pydantic import BaseModel
-from sqlalchemy import Column, DateTime, text
+from sqlalchemy import Column, DateTime, text, delete
 from sqlmodel import Field, select
 
+from bisheng.database.models.role import AdminRole
+
 
 class UserRoleBase(SQLModelSerializable):
     user_id: int = Field(index=True)
@@ -58,7 +60,7 @@ def get_admins_user(cls) -> List[UserRole]:
         获取所有超级管理的账号
         """
         with session_getter() as session:
-            statement = select(UserRole).where(UserRole.role_id == 1)
+            statement = select(UserRole).where(UserRole.role_id == AdminRole)
             return session.exec(statement).all()
 
     @classmethod
@@ -67,8 +69,29 @@ def set_admin_user(cls, user_id: int) -> UserRole:
         设置用户为超级管理员
         """
         with session_getter() as session:
-            user_role = UserRole(user_id=user_id, role_id=1)
+            user_role = UserRole(user_id=user_id, role_id=AdminRole)
             session.add(user_role)
             session.commit()
             session.refresh(user_role)
             return user_role
+
+    @classmethod
+    def add_user_roles(cls, user_id: int, role_ids: List[int]) -> List[UserRole]:
+        """
+        给用户批量添加角色
+        """
+        with session_getter() as session:
+            user_roles = [UserRole(user_id=user_id, role_id=role_id) for role_id in role_ids]
+            session.add_all(user_roles)
+            session.commit()
+            return user_roles
+
+    @classmethod
+    def delete_user_roles(cls, user_id: int, role_ids: List[int]) -> None:
+        """
+        将用户从某些角色中移除
+        """
+        with session_getter() as session:
+            statement = delete(UserRole).where(UserRole.user_id == user_id).where(UserRole.role_id.in_(role_ids))
+            session.exec(statement)
+            session.commit()