dash-fastapi-admin/dash-fastapi-backend/module_admin/service/fccbd_service.py

from module_admin.entity.vo.user_vo import *
from module_admin.entity.vo.fccbd_vo import *
from module_admin.dao.fccbd_dao import *
from utils.common_util import export_list2excel
from utils.common_util import *





class FccbdService:
    """
    词典配置表管理模块服务层
    """

    @classmethod
    def get_fccbd_select_option_services(cls, result_db: Session):
        """
        获取词典配置表列表不分页信息service
        :param result_db: orm对象
        :return: 词典配置表列表不分页信息对象
        """
        fccbd_list_result = FccbdDao.get_fccbd_select_option_dao(result_db)

        return fccbd_list_result

    @classmethod
    def get_fccbd_list_services(cls, result_db: Session, query_object: FccbdModel):
        """
        获取词典配置表列表信息service
        :param result_db: orm对象
        :param query_object: 查询参数对象
        :return: 词典配置表列表信息对象
        """
        fccbd_list_result = FccbdDao.get_fccbd_list(result_db, query_object)

        return fccbd_list_result


   
    @classmethod
    async def query_fccbd_list_from_cache_services(cls, redis, onum: int):
        """
        从缓存获取词典配置表列表信息service
        :param redis: redis对象
        :param  : 字典类型
        :return: 词典配置表列表信息对象
        """
        result = []
        dict_data_list_result = onum
        print('测试使用，没有实际意义的方法！')
        if dict_data_list_result:
            result = onum

        return result



    @classmethod
    def add_fccbd_services(cls, result_db: Session, page_object: FccbdModel):
        """
        新增词典配置表信息service
        :param result_db: orm对象
        :param page_object: 新增词典配置表对象
        :return: 新增词典配置表校验结果
        """
        fccbd = FccbdDao.get_fccbd_detail_by_info(result_db, FccbdModel(**dict(term=page_object.term)))
        if fccbd:
            result = dict(is_success=False, message='词典配置表名称已存在')
        else:
            try:
                FccbdDao.add_fccbd_dao(result_db, page_object)
                result_db.commit()
                result = dict(is_success=True, message='新增成功')
            except Exception as e:
                result_db.rollback()
                result = dict(is_success=False, message=str(e))

        return CrudFccbdResponse(**result)

    @classmethod
    def edit_fccbd_services(cls, result_db: Session, page_object: FccbdModel):
        """
        编辑词典配置表信息service
        :param result_db: orm对象
        :param page_object: 编辑词典配置表信息对象
        :return: 编辑词典配置表信息校验结果
        """
        edit_fccbd = page_object.dict(exclude_unset=True)

        fccbd_info = cls.detail_fccbd_services(result_db, edit_fccbd.get('onum'))
        if fccbd_info:
            if  fccbd_info.onum != page_object.onum:
                fccbd = FccbdDao.get_fccbd_by_info(result_db, FccbdModel(**dict(onum=page_object.onum)))
                if fccbd:
                    result = dict(is_success=False, message='字段名已存在')
                    return CrudFccbdResponse(**result)
            try:
                FccbdDao.edit_fccbd_dao(result_db, edit_fccbd)
                result_db.commit()
                result = dict(is_success=True, message='更新成功')
            except Exception as e:
                result_db.rollback()
                result = dict(is_success=False, message=str(e))
        else:
            result = dict(is_success=False, message='字段不存在')

        return CrudFccbdResponse(**result)

    @classmethod
    def delete_fccbd_services(cls, result_db: Session, page_object: DeleteFccbdModel):
        """
        删除词典配置表信息service
        :param result_db: orm对象
        :param page_object: 删除词典配置表对象
        :return: 删除词典配置表校验结果
        """
        if page_object.onums.split(','):
            onum_list = page_object.onums.split(',')
            try:
                for onum in onum_list:
                    onum_dict = dict(onum=onum)
                    FccbdDao.delete_fccbd_dao(result_db, FccbdModel(**onum_dict))
                result_db.commit()
                result = dict(is_success=True, message='删除成功')
            except Exception as e:
                result_db.rollback()
                result = dict(is_success=False, message=str(e))
        else:
            result = dict(is_success=False, message='传入词典配置表onum为空')
        return CrudFccbdResponse(**result)

    @classmethod
    def detail_fccbd_services(cls, result_db: Session, onum: int):
        """
        获取词典配置表详细信息service
        :param result_db: orm对象
        :param onum: 词典配置表id
        :return: 词典配置表id对应的信息
        """
        fccbd = FccbdDao.get_fccbd_detail_by_id(result_db, onum=onum)

        return fccbd

    @staticmethod
    def export_fccbd_list_services(fccbd_list: List):
        """
        导出词典配置表信息service
        :param fccbd_list: 词典配置表信息列表
        :return: 词典配置表信息对应excel的二进制数据
        """
        # 创建一个映射词典配置表，将英文键映射到中文键
        mapping_dict = {
            "onum": "序号",
			"term": "词语",
			"freq": "频次",
			"pos": "词性",
			"pos_name": "词性名称",
            "status": "状态",
            #"create_by": "创建者",
            #"create_time": "创建时间",
            "update_by": "负责人",
            "update_time": "更新时间",
        }

		
        data = [FccbdModel(**vars(row)).dict() for row in fccbd_list]

        for item in data:
            if item.get('status') == '0':
                item['status'] = '正常'
            else:
                item['status'] = '停用'
        new_data = [{mapping_dict.get(key): value for key, value in item.items() if mapping_dict.get(key)} for item in data]
        binary_data = export_list2excel(new_data)

        return binary_data

    @classmethod
    def batch_import_fccbd_services(cls, result_db: Session, fccbd_import: ImportFccbdModel, current_user: CurrentUserInfoServiceResponse):
        """
        批量导入词典配置表表service
        :param fccbd_import: 词典配置表表导入参数对象
        :param result_db: orm对象
        :param current_fccbd: 当前词典配置表表对象
        :return: 批量导入词典配置表表结果
        """
        header_dict = {
            #"序号": "onum",
			"词语": "term",
			"频次": "freq",
            "词性": "pos",
            "词性名称": "pos_name",
			"状态": "status"
        }
 
 
        filepath = get_filepath_from_url(fccbd_import.url)
        df = pd.read_excel(filepath)
        df.rename(columns=header_dict, inplace=True)
        add_error_result = []
        count = 0
        try:
            for index, row in df.iterrows():
                count = count + 1
                if row['status'] == '正常':
                    row['status'] = '0'
                if row['status'] == '停用':
                    row['status'] = '1'
                add_fccbd = FccbdModel(
                    **dict(
                        #onum=row['onum'],
						term=row['term'],
                        freq=row['freq'],
                        pos=row['pos'],
                        pos_name=row['pos_name'],
                        status=row['status'],
                        create_by=current_user.user.user_name,
                        update_by=current_user.user.user_name,
                    )
                )
                fccbd_infos = FccbdDao.get_fccbd_by_info_imp(result_db, FccbdModel(**dict(term=row['term'],pos=row['pos'])))            

                if fccbd_infos:
                    if fccbd_import.is_update:
                        edit_fccbd = FccbdModel(
                            **dict(
								#onum=row['onum'],
								term=row['term'],
								freq=row['freq'],
								pos=row['pos'],
								pos_name=row['pos_name'],
								status=row['status'],
								update_by=current_user.user.user_name
                            )
                        ).dict(exclude_unset=True)
                        
                        FccbdDao.edit_fccbd_dao(result_db, edit_fccbd)
                    else:
                        add_error_result.append(f"{count}.用户账号{row['user_name']}已存在")
                else:
                    FccbdDao.add_fccbd_dao(result_db, add_fccbd)
            result_db.commit()
            result = dict(is_success=True, message='\n'.join(add_error_result))
        except Exception as e:
            result_db.rollback()
            result = dict(is_success=False, message=str(e))

        return CrudFccbdResponse(**result)

    @staticmethod
    def get_fccbd_import_template_services():
        """
        获取词典配置表表导入模板service
        :return: 词典配置表表导入模板excel的二进制数据
        """
        #header_list = ["序号", "词语", "频次", "词性", "词性名称",  "状态"]
        header_list = ["词语", "频次", "词性", "词性名称",  "状态"]
        selector_header_list = ["状态"]
        option_list = [{"状态": ["正常", "停用"]}]
        binary_data = get_excel_template(header_list=header_list, selector_header_list=selector_header_list, option_list=option_list)

        return binary_data