group_manage/ruoyi_system/service/sys_dept.py

# -*- coding: utf-8 -*-
# @Author  : YY

from types import NoneType
from typing import Any, List, Literal

from ruoyi_common.sqlalchemy.transaction import Transactional
from ruoyi_common.utils import security_util as SecurityUtil
from ruoyi_common.constant import UserConstants
from ruoyi_common.domain.entity import SysDept, SysRole, TreeSelect
from ruoyi_common.exception import ServiceException
from ruoyi_system.mapper import SysDeptMapper, SysRoleMapper
from ruoyi_admin.ext import db


class SysDeptService:
    
    @classmethod
    def select_dept_list(cls, dept:SysDept) -> List[SysDept]:
        """
        查询部门列表

        Args:
            dept (SysDept): 包含查询条件的传输对象

        Returns:
            List[SysDept]: 部门列表
        """
        return SysDeptMapper.select_dept_list(dept or SysDept())
    
    @classmethod
    def select_dept_tree_list(cls, dept:SysDept):
        """
        查询部门树列表

        Args:
            dept (SysDept): 包含查询条件的传输对象

        Returns:
            List[TreeSelect]: 部门树列表
        """
        eos: List[SysDept] = cls.select_dept_list(dept)
        return cls.build_dept_tree_select(eos)
        
    @classmethod
    def build_dept_tree(cls, depts:List[SysDept]) -> List[SysDept]:
        """
        构建部门树

        Args:
            depts (List[SysDept]): 部门列表

        Returns:
            List[SysDept]: 部门树列表
        """
        return_list = []
        temp_list = []
        for dept in depts:
            temp_list.append(dept.dept_id)
        for dept in depts:
            if dept.parent_id not in temp_list:
                cls.recursion_fn(depts, dept)
                return_list.append(dept)
        if not return_list:
            return_list = depts
        return return_list

    @classmethod
    def build_dept_tree_select(cls, depts:List[SysDept]) -> List[TreeSelect]:
        """
        构建部门树选择框

        Args:
            depts (List[SysDept]): 部门列表

        Returns:
            List[TreeSelect]: 部门树选择框列表  
        """
        dept_trees = cls.build_dept_tree(depts)
        return [TreeSelect.from_dept(dept) for dept in dept_trees]

    @classmethod
    def select_dept_list_by_role_id(cls, role_id:int) -> List[int]:
        """
        根据角色ID，查询部门列表
        
        Args:
            role_id (int): 角色ID

        Returns:
            List[int]: 部门列表
        """
        role:SysRole = SysRoleMapper.select_role_by_id(role_id)
        return SysDeptMapper.select_dept_list_by_role_id(
            role_id, role.dept_check_strictly
        )

    @classmethod
    def select_dept_by_id(cls, dept_id:int) -> SysDept|NoneType:
        """
        根据部门ID，查询部门信息
        
        Args:
            dept_id (int): 部门ID

        Returns:
            SysDept|NoneType: 部门信息
        """
        return SysDeptMapper.select_dept_by_id(dept_id)        

    @classmethod
    def select_normal_children_dept_by_id(cls, dept_id:int) -> int:
        """
        根据部门ID，查询正常部门的数量
        
        Args:
            dept_id (int): 部门ID

        Returns:
            int: 正常部门的数量
        """
        return SysDeptMapper.select_normal_children_dept_by_id(dept_id)

    @classmethod
    def has_child_by_dept_id(cls, dept_id:int) -> int:
        """
        根据部门ID，判断是否有子部门
        
        Args:
            dept_id (int): 部门ID

        Returns:
            int: 数量
        """
        return SysDeptMapper.has_child_by_dept_id(dept_id)

    @classmethod
    def check_dept_exist_user(cls, dept_id:int) -> int:
        """
        判断部门是否存在用户
        
        Args:
            dept_id (int): 部门ID

        Returns:
            int: 数量
        """
        return SysDeptMapper.check_dept_exist_user(dept_id)

    @classmethod
    def check_dept_name_unique(cls, dept:SysDept) -> Literal["0","1"]:
        """
        判断部门名称是否唯一
        
        Args:
            dept (SysDept): 部门信息

        Returns:
            Literal["0","1"]: 唯一或不唯一, 0: 唯一, 1: 不唯一
        """
        dept_id = -1 if dept.dept_id is None else dept.dept_id
        eo:SysDept = SysDeptMapper.check_dept_name_unique(dept.dept_name, dept.parent_id)
        if eo is None:
            return UserConstants.UNIQUE
        if eo.dept_id != dept_id:
            return UserConstants.NOT_UNIQUE
        return UserConstants.UNIQUE

    @classmethod
    def check_dept_data_scope(cls, dept_id:int):
        """
        判断部门数据权限
        
        Args:
            dept_id (int): 部门ID

        Raises:
            ServiceException: 无权限访问部门数据
        """
        if not SecurityUtil.login_user_is_admin():
            dept = SysDept(dept_id=dept_id)
            depts = cls.select_dept_list(dept)
            if not depts:
                raise ServiceException("没有权限访问部门数据！")
            
    @classmethod
    def insert_dept(cls, dept:SysDept) -> int:
        """
        新增部门信息

        Args:
            dept (SysDept): 部门信息

        Returns:
            int: 部门ID
        """
        parent = SysDeptMapper.select_dept_by_id(dept.parent_id)
        if not UserConstants.DEPT_NORMAL == parent.status:
            raise ServiceException("部门停用，不允许新增")
        dept.ancestors = f"{parent.ancestors},{dept.parent_id}"
        return SysDeptMapper.insert_dept(dept)

    @classmethod
    @Transactional(db.session)
    def update_dept(cls, dept:SysDept) -> int:
        """
        修改部门信息

        Args:
            dept (SysDept): 部门信息

        Returns:
            int: 数量
        """
        parent_dept = SysDeptMapper.select_dept_by_id(dept.parent_id)
        old_dept = SysDeptMapper.select_dept_by_id(dept.dept_id)
        if parent_dept is not None and old_dept is not None:
            new_ancestors = f"{parent_dept.ancestors},{parent_dept.dept_id}"
            old_ancestors = old_dept.ancestors
            dept.ancestors = new_ancestors
            cls.update_dept_children(dept.dept_id, new_ancestors, old_ancestors)
        num = SysDeptMapper.update_dept(dept)
        if UserConstants.DEPT_NORMAL == dept.status and dept.ancestors and dept.ancestors != "0":
            # 如果该部门是启用状态，则启用该部门的所有上级部门
            cls.update_parent_dept_status_normal(dept)
        return num

    @classmethod
    def update_parent_dept_status_normal(cls, dept:SysDept):
        """
        修改该部门的父级部门

        Args:
            dept (SysDept): 部门信息
        """
        dept_ids = dept.ancestors.split(",")
        SysDeptMapper.update_dept_status_normal(dept_ids)

    @classmethod
    def update_dept_children(cls, dept_id, new_ancestors, old_ancestors):
        """
        修改子元素关系

        Args:
            dept_id (int): 部门ID
            new_ancestors (str): 新的祖先路径
            old_ancestors (str): 旧的祖先路径
        """
        children:List[SysDept] = SysDeptMapper.select_children_dept_by_id(dept_id)
        for child in children:
            child.ancestors = (child.ancestors.replace(old_ancestors, new_ancestors, 1))
        if children:
            SysDeptMapper.update_dept_children(children)

    @classmethod
    def delete_dept_by_id(cls, dept_id) -> int:
        """
        删除部门管理信息

        Args:
            dept_id (int): 部门ID

        Returns:
            int: 数量
        """
        return SysDeptMapper.delete_dept_by_id(dept_id)

    @classmethod
    def recursion_fn(cls, list:List[SysDept], t:SysDept):
        """
        递归列表
        
        Args:
            list (List[SysDept]): 部门列表
            t (SysDept): 父节点
        """
        child_list = cls.get_child_list(list, t)
        t.children = child_list
        for t_child in child_list:
            if cls.has_child(list, t_child):
                cls.recursion_fn(list, t_child)

    @classmethod
    def get_child_list(cls, list:List[SysDept], t:SysDept) -> List[Any]:
        """
        获取子节点列表
        
        Args:
            list (List[SysDept]): 部门列表
            t (SysDept): 父节点

        Returns:
            List[Any]: 子节点列表
        """
        tlist = []
        for n in list:
            if n.parent_id is not None and n.parent_id==t.dept_id:
                tlist.append(n)
        return tlist

    @classmethod
    def has_child(cls, list:List[SysDept], t:SysDept) -> bool:
        """
        判断是否有子节点
        
        Args:
            list (List[SysDept]): 部门列表
            t (SysDept): 父节点

        Returns:
            bool: 是否有子节点
        """
        for n in list:
            if n.parent_id is not None and n.parent_id==t.dept_id:
                return True
        return False