You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
249 lines
11 KiB
249 lines
11 KiB
1 month ago
|
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
|