from module_admin.entity.vo.user_vo import * from module_admin.entity.vo.dassetdict_vo import * from module_admin.dao.dassetdict_dao import * from utils.pwd_util import * from utils.common_util import * class DassetdictService: """ 字段管理模块服务层 """ @classmethod def get_dassetdict_list_services(cls, result_db: Session, query_object: DassetdictQueryModel, data_scope_sql: str): """ 获取字段列表信息service :param result_db: orm对象 :param query_object: 查询参数对象 :param data_scope_sql: 数据权限对应的查询sql语句 :return: 字段列表信息对象 """ dassetdict_list_result = DassetdictDao.get_dassetdict_list(result_db, query_object, data_scope_sql) return dassetdict_list_result @classmethod def add_dassetdict_services(cls, result_db: Session, page_object: AddDassetdictModel): """ 新增字段信息service :param result_db: orm对象 :param page_object: 新增字段对象 :return: 新增字段校验结果 """ add_dassetdict = DassetdictModel(**page_object.dict()) dassetdict = DassetdictDao.get_dassetdict_by_info(result_db, DassetdictModel(**dict(dict_fn=page_object.dict_fn))) if dassetdict: result = dict(is_success=False, message='字段名已存在') else: try: add_result = DassetdictDao.add_dassetdict_dao(result_db, add_dassetdict) dict_id = add_result.dict_id 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 CrudDassetdictResponse(**result) @classmethod def edit_dassetdict_services(cls, result_db: Session, page_object: AddDassetdictModel): """ 编辑字段信息service :param result_db: orm对象 :param page_object: 编辑字段对象 :return: 编辑字段校验结果 """ edit_dassetdict = page_object.dict(exclude_unset=True) if page_object.type == 'status' or page_object.type == 'avatar': del edit_dassetdict['type'] dassetdict_info = cls.detail_dassetdict_services(result_db, edit_dassetdict.get('dict_id')) if dassetdict_info: if page_object.type != 'status' and page_object.type != 'avatar' and dassetdict_info.dassetdict.dict_fn != page_object.dict_fn: dassetdict = DassetdictDao.get_dassetdict_by_info(result_db, DassetdictModel(**dict(dict_fn=page_object.dict_fn))) if dassetdict: result = dict(is_success=False, message='字段名已存在') return CrudDassetdictResponse(**result) try: DassetdictDao.edit_dassetdict_dao(result_db, edit_dassetdict) if page_object.type != 'status' and page_object.type != 'avatar': dict_id_dict = dict(dict_id=page_object.dict_id) 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 CrudDassetdictResponse(**result) @classmethod def delete_dassetdict_services(cls, result_db: Session, page_object: DeleteDassetdictModel): """ 删除字段信息service :param result_db: orm对象 :param page_object: 删除字段对象 :return: 删除字段校验结果 """ if page_object.dict_ids.split(','): dict_id_list = page_object.dict_ids.split(',') try: for dict_id in dict_id_list: dict_id_dict = dict(dict_id=dict_id) print(111111111111111111,dict_id_dict) DassetdictDao.delete_dassetdict_dao(result_db, DassetdictModel(**dict_id_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 CrudDassetdictResponse(**result) @classmethod def detail_dassetdict_services(cls, result_db: Session, dict_id: int): """ 获取字段详细信息service :param result_db: orm对象 :param dict_id: 字段id :return: 字段id对应的信息 """ dassetdict = DassetdictDao.get_dassetdict_detail_by_id(result_db, dict_id=dict_id) return DassetdictDetailModel( dassetdict=dassetdict.dassetdict_basic_info, dasset=dassetdict.dassetdict_dasset_info ) @staticmethod def export_dassetdict_list_services(dassetdict_list: List): """ 导出元数据字段信息service :param user_list: 元数据字段信息列表 :return: 元数据字段信息对应excel的二进制数据 """ # 创建一个映射字典，将英文键映射到中文键 mapping_dict = { "dasset_id": "数据资产ID", "dict_id": "序号", "tab_name": "数据资产名", "dict_fn": "字段名", "tab_cn_name": "资产中文名", "data_dict_no": "数据字典编号", "dict_fccn": "字段中文名", "dict_pk": "是否主键", "dict_dft": "数据类型", "dict_ft": "字段类型", "dict_bc": "业务含义", "dict_al": "别名", "memo": "备忘录", "dict_sg": "安全分级", "order_asc": "排序", # 添加新的键值对 "status": "状态", } data = [DassetdictModel(**row).dict() for row in dassetdict_list] # data = [DassetdictModel(**vars(row)).dict() for row in dassetdict_list] # print(data,77777) for item in data: if item.get('status') == '0': item['status'] = '正常' else: item['status'] = '停用' if item.get('dict_pk') == '1': item['dict_pk'] = '主键' else: item['dict_pk'] = '' 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_dassetdict_services(cls, result_db: Session, dassetdict_import: ImportDassetdictModel, current_user: CurrentUserInfoServiceResponse): """ 批量导入元数据字段表service :param dassetdict_import: 元数据字段表导入参数对象 :param result_db: orm对象 :param current_dassetdict: 当前元数据字段表对象 :return: 批量导入代码表结果 """ header_dict = { "数据资产ID": "dasset_id", "序号": "dict_id", "数据资产名": "tab_name", "字段名": "dict_fn", "资产中文名": "tab_cn_name", "数据字典编号": "data_dict_no", "字段中文名": "dict_fccn", "是否主键": "dict_pk", "数据类型": "dict_dft", "字段类型": "dict_ft", "业务含义": "dict_bc", "别名":"dict_al", "备忘录": "memo", "安全分级": "dict_sg", "排序": "order_asc", # 添加新的键值对 "状态": "status", } filepath = get_filepath_from_url(dassetdict_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_dassetdict = DassetdictModel( **dict( dasset_id=row['dasset_id'], dict_id=row['dict_id'], tab_name = str(row['tab_name']).replace('nan', ''), dict_fn = str(row['dict_fn']).replace('nan', ''), tab_cn_name = str(row['tab_cn_name']).replace('nan', ''), data_dict_no = str(row['data_dict_no']).replace('nan', ''), dict_fccn = str(row['dict_fccn']).replace('nan', ''), dict_pk = str(row['dict_pk']).replace('nan', ''), dict_dft = str(row['dict_dft']).replace('nan', ''), dict_ft = str(row['dict_ft']).replace('nan', ''), dict_bc = str(row['dict_bc']).replace('nan', ''), dict_al = str(row['dict_al']).replace('nan', ''), memo = str(row['memo']).replace('nan', ''), dict_sg = str(row['dict_sg']).replace('nan', ''), status=row['status'], order_asc=row['order_asc'], # 添加新的键值对 create_by=current_user.user.user_name, update_by=current_user.user.user_name, ) ) dassetdict_info = DassetdictDao.get_dassetdict_by_info(result_db, DassetdictModel(**dict(dict_id=row['dict_id']))) if dassetdict_info: if dassetdict_import.is_update: edit_dassetdict = DassetdictModel( **dict( dasset_id=row['dasset_id'], dict_id=row['dict_id'], tab_name = str(row['tab_name']).replace('nan', ''), dict_fn = str(row['dict_fn']).replace('nan', ''), tab_cn_name = str(row['tab_cn_name']).replace('nan', ''), data_dict_no = str(row['data_dict_no']).replace('nan', ''), dict_fccn = str(row['dict_fccn']).replace('nan', ''), dict_pk = str(row['dict_pk']).replace('nan', ''), dict_dft = str(row['dict_dft']).replace('nan', ''), dict_ft = str(row['dict_ft']).replace('nan', ''), dict_bc = str(row['dict_bc']).replace('nan', ''), dict_al = str(row['dict_al']).replace('nan', ''), memo = str(row['memo']).replace('nan', ''), dict_sg = str(row['dict_sg']).replace('nan', ''), order_asc=row['order_asc'], # 添加新的键值对 status=row['status'], update_by=current_user.user.user_name ) ).dict(exclude_unset=True) DassetdictDao.edit_dassetdict_dao(result_db, edit_dassetdict) else: add_error_result.append(f"{count}.用户账号{row['user_name']}已存在") else: DassetdictDao.add_dassetdict_dao(result_db, add_dassetdict) 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 CrudDassetdictResponse(**result) @staticmethod def get_dassetdict_import_template_services(): """ 获取代码表导入模板service :return: 代码表导入模板excel的二进制数据 """ header_list = ["数据资产ID", "序号", "数据资产名", "字段名", "资产中文名", "数据字典编号", "字段中文名", "是否主键", "数据类型", "字段类型", "业务含义", "别名", "备忘录", "安全分级", "排序", "状态"] 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