from module_admin.entity.vo.nddict_vo import * from module_admin.entity.vo.user_vo import * from module_admin.dao.nddict_dao import * from utils.common_util import * from utils.pwd_util import * class NddictService: """ 数据字典管理模块服务层 """ @classmethod def get_nddict_select_option_services(cls, result_db: Session): """ 获取数据字典列表不分页信息service :param result_db: orm对象 :return: 数据字典列表不分页信息对象 """ nddict_list_result = NddictDao.get_nddict_select_option_dao(result_db) return nddict_list_result @classmethod def get_nddict_list_services(cls, result_db: Session, query_object: NddictModel): """ 获取数据字典列表信息service :param result_db: orm对象 :param query_object: 查询参数对象 :return: 数据字典列表信息对象 """ nddict_list_result = NddictDao.get_nddict_list(result_db, query_object) return nddict_list_result @classmethod def add_nddict_services(cls, result_db: Session, page_object: NddictModel): """ 新增代码管理信息service :param result_db: orm对象 :param page_object: 新增数据字典对象 :return: 新增数据字典校验结果 """ nddict = NddictDao.get_nddict_detail_by_info(result_db, NddictModel(**dict(data_dict_cn_name=page_object.data_dict_cn_name))) if nddict: result = dict(is_success=False, message='数据字典名称已存在') else: try: NddictDao.add_nddict_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 CrudNddictResponse(**result) @classmethod def edit_nddict_services(cls, result_db: Session, page_object: NddictModel): """ 数据字典信息service :param result_db: orm对象 :param page_object: 数据字典信息对象 :return: 编辑数据字典校验结果 """ edit_nddict = page_object.dict(exclude_unset=True) nddict_info = cls.detail_nddict_services(result_db, edit_nddict.get('onum')) if nddict_info: if nddict_info.data_dict_cn_name != page_object.data_dict_cn_name: nddict = NddictDao.get_nddict_by_info(result_db, NddictModel(**dict(data_dict_cn_name=page_object.data_dict_cn_name))) if nddict: result = dict(is_success=False, message='数据字典名称已存在') return CrudNddictResponse(**result) try: NddictDao.edit_nddict_dao(result_db, edit_nddict) 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 CrudNddictResponse(**result) @classmethod def delete_nddict_services(cls, result_db: Session, page_object: DeleteNddictModel): """ 删除代码管理信息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) NddictDao.delete_nddict_dao(result_db, NddictModel(**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='传入数据字典id为空') return CrudNddictResponse(**result) @classmethod def detail_nddict_services(cls, result_db: Session, onum: int): """ 获取数据字典详细信息service :param result_db: orm对象 :param onum: 数据字典id :return: 数据字典id对应的信息 """ nddict = NddictDao.get_nddict_detail_by_id(result_db, onum=onum) return nddict @staticmethod def export_nddict_list_services(nddict_list: List): """ 导出数据字典信息service :param nddict_list: 数据字典信息列表 :return: 数据字典信息对应excel的二进制数据 """ # 创建一个映射数据字典，将英文键映射到中文键 mapping_dict = { "onum": "序号", "data_dict_no": "数据字典编号", "data_std_no": "数据标准编号", "data_dict_cn_name": "数据字典中文名称", "data_dict_eng_name": "数据字典英文名称", "dict_busi_mean": "字典业务含义", "fld_bype": "字段类型", "falias": "字典别名", "fld_dbype": "字段数据类型", "status": "状态", } data = [NddictModel(**vars(row)).dict() for row in nddict_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_nddict_services(cls, result_db: Session, nddict_import: ImportNddictModel, current_user: CurrentUserInfoServiceResponse): """ 批量导入数据字典表service :param nddict_import: 数据字典表导入参数对象 :param result_db: orm对象 :param current_nddict: 当前数据字典表对象 :return: 批量导入数据字典表结果 """ header_dict = { "序号": "onum", "数据字典编号": "data_dict_no", "数据标准编号": "data_std_no", "数据字典中文名称": "data_dict_cn_name", "数据字典英文名称": "data_dict_eng_name", "字典业务含义": "dict_busi_mean", "字段类型": "fld_bype", "字典别名": "falias", "字段数据类型": "fld_dbype", "状态": "status" } filepath = get_filepath_from_url(nddict_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_nddict = NddictModel( **dict( onum = str(row['onum']).replace('nan', ''), data_dict_no = str(row['data_dict_no']).replace('nan', ''), data_std_no = str(row['data_std_no']).replace('nan', ''), data_dict_cn_name = str(row['data_dict_cn_name']).replace('nan', ''), data_dict_eng_name = str(row['data_dict_eng_name']).replace('nan', ''), dict_busi_mean = str(row['dict_busi_mean']).replace('nan', ''), fld_bype = str(row['fld_bype']).replace('nan', ''), falias = str(row['falias']).replace('nan', ''), fld_dbype = str(row['fld_dbype']).replace('nan', ''), status = str(row['status']).replace('nan', ''), create_by = current_user.user.user_name, update_by = current_user.user.user_name ) ) nddict_info = NddictDao.get_nddict_by_info(result_db, NddictModel(**dict(onum=row['onum']))) if nddict_info: if nddict_import.is_update: edit_nddict = NddictModel( **dict( onum = str(row['onum']).replace('nan', ''), data_dict_no = str(row['data_dict_no']).replace('nan', ''), data_std_no = str(row['data_std_no']).replace('nan', ''), data_dict_cn_name = str(row['data_dict_cn_name']).replace('nan', ''), data_dict_eng_name = str(row['data_dict_eng_name']).replace('nan', ''), dict_busi_mean = str(row['dict_busi_mean']).replace('nan', ''), fld_bype = str(row['fld_bype']).replace('nan', ''), falias = str(row['falias']).replace('nan', ''), fld_dbype = str(row['fld_dbype']).replace('nan', ''), status = str(row['status']).replace('nan', ''), update_by=current_user.user.user_name, ) ).dict(exclude_unset=True) NddictDao.edit_nddict_dao(result_db, edit_nddict) else: add_error_result.append(f"{count}.用户账号{row['user_name']}已存在") else: NddictDao.add_nddict_dao(result_db, add_nddict) 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 CrudNddictResponse(**result) @staticmethod def get_nddict_import_template_services(): """ 获取数据字典表导入模板service :return: 数据字典表导入模板excel的二进制数据 """ 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