cjy-agent/packages/dbgpt-system/src/dbgpt_system/dept/dept_db.py

from dbgpt.storage.metadata import BaseDao
from dbgpt_system.dept.dept import  DeptEntity,DeptResponse,Dept
from dbgpt_system.sys_user.user import UserEntity
from dbgpt_system.sys_user.user_db import  UserDao
from dbgpt_serve.utils.auth import UserRequest
from datetime import datetime
deptEntity = DeptEntity
userDao = UserDao
class DeptDao(BaseDao):
    def select_dept_list(self, dept: Dept):
        session = self.get_raw_session()
        try:
            # 查询本级单位
            dept_qry = session.query(DeptEntity).filter(DeptEntity.deleted == 0, DeptEntity.id == dept.id,)
            if dept.name:
                dept_qry = dept_qry.filter(
                    DeptEntity.name.like(f"%{dept.name}%")
                )
            if dept.status is not None:
                dept_qry = dept_qry.filter(
                    DeptEntity.status == dept.status
                )
            parent_results = dept_qry.all()

            # 初始化结果列表
            all_depts = []

            # 定义递归函数来查找子部门
            def find_subdepartments(parent_id):
                subdept_qry = session.query(DeptEntity).filter(
                    DeptEntity.parent_id == parent_id,
                    DeptEntity.deleted == 0
                )
                subdepts = subdept_qry.all()
                for subdept in subdepts:
                    all_depts.append(subdept)
                    find_subdepartments(subdept.id)

            # 遍历父部门并查找其子部门
            for parent_dept in parent_results:
                all_depts.append(parent_dept)
                find_subdepartments(parent_dept.id)

        finally:
            session.close()

        if all_depts:
            depts = []
            dept_resp = DeptResponse()

            for dept in all_depts:
                depts.append(
                    Dept.from_dict({
                        "id": dept.id,
                        "name": dept.name,
                        "sort": dept.sort,
                        "leader_user_id": dept.leader_user_id,
                        "phone": dept.phone,
                        "email": dept.email,
                        "status": dept.status,
                        "creator": dept.creator,
                        "create_time": dept.create_time,
                        "updater": dept.updater,
                        "update_time": dept.update_time,
                        "parent_id": dept.parent_id
                    })
                )
            dept_resp.dept_list = depts
            return dept_resp

    def create(self,user_info : UserRequest,dept : Dept):
        try:
            with self.session() as session:
                dept_entity = DeptEntity(
                    name = dept.name,
                    sort = dept.sort,
                    leader_user_id = dept.leader_user_id,
                    phone = dept.phone,
                    email = dept.email,
                    status = dept.status,
                    creator = user_info.user_id,
                    create_time = datetime.now(),
                    updater =  user_info.user_id,
                    update_time = datetime.now(),
                    parent_id = dept.parent_id,
                    deleted = 0

                )
                session.add(dept_entity)
                session.commit()
                session.close()
            return True
        except Exception as e:
            session.rollback()
            raise RuntimeError(f"创建部门失败: {str(e)}")

    def update(self,user_info : UserRequest,dept:Dept):
        session = self.get_raw_session()
        with self.session() as session:
            dept_entity = session.query(DeptEntity).filter_by(id=dept.id, deleted=0).first()
            if dept_entity:
                dept_entity.id =dept.id
                dept_entity.name = dept.name
                dept.sort   = dept.sort
                dept.parent_id = dept.parent_id
                dept_entity.leader_user_id = dept.leader_user_id
                dept_entity.phone = dept.phone
                dept_entity.email = dept.email
                dept_entity.status = dept.status
                dept_entity.updater =  user_info.user_id
                dept_entity.update_time = datetime.now()
                session.commit()
            else:
                raise ValueError("部门不存在或已被删除")
            session.close()
            return True
    def select_dept_details(self, dept_id:int):
        session = self.get_raw_session()
        try:
            dept_qry = session.query(DeptEntity).filter(DeptEntity.deleted == 0,DeptEntity.id == dept_id)
            dept_details = dept_qry.first()
        finally:
            session.close()
        return Dept.from_dict(
            {
                "id": dept_details.id,
                "name": dept_details.name,
                "sort": dept_details.sort,
                "leader_user_id": dept_details.leader_user_id,
                "phone": dept_details.phone,
                "email": dept_details.email,
                "status": dept_details.status,
                "parent_id": dept_details.parent_id,
                "create_time": dept_details.create_time,
                "creator": dept_details.creator,
                "updater": dept_details.updater,
                "update_time": dept_details.update_time,
            }
        )

    def delete_dept(self, dept_id :int):
        with self.session() as session:
            count = session.query(UserEntity).filter_by(dept_id = dept_id, deleted=0).count()
            if count >0:
                raise ValueError("该单位下存在人员信息,不能删除")
            else:
                dept_entity = session.query(DeptEntity).filter_by(id=dept_id,deleted=0).first()
                if dept_entity is not None:
                    if dept_entity.id == 1:
                        raise ValueError("顶级单位不能删除")
                    dept_entity.deleted =1
                else:
                    raise ValueError("部门信息不存在或已被删除")
            session.commit()
            return True