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.
274 lines
11 KiB
274 lines
11 KiB
from module_admin.entity.vo.user_vo import *
|
|
from module_admin.entity.vo.fddict_vo import *
|
|
from module_admin.dao.fddict_dao import *
|
|
from utils.common_util import export_list2excel
|
|
from utils.common_util import *
|
|
|
|
|
|
|
|
|
|
class FddictService:
|
|
"""
|
|
代码管理模块服务层
|
|
"""
|
|
|
|
@classmethod
|
|
def get_fddict_select_option_services(cls, result_db: Session):
|
|
"""
|
|
获取代码列表不分页信息service
|
|
:param result_db: orm对象
|
|
:return: 代码列表不分页信息对象
|
|
"""
|
|
fddict_list_result = FddictDao.get_fddict_select_option_dao(result_db)
|
|
|
|
return fddict_list_result
|
|
|
|
@classmethod
|
|
def get_fddict_list_services(cls, result_db: Session, query_object: FddictModel):
|
|
"""
|
|
获取代码列表信息service
|
|
:param result_db: orm对象
|
|
:param query_object: 查询参数对象
|
|
:return: 代码列表信息对象
|
|
"""
|
|
fddict_list_result = FddictDao.get_fddict_list(result_db, query_object)
|
|
|
|
return fddict_list_result
|
|
|
|
|
|
|
|
@classmethod
|
|
async def query_fddict_list_from_cache_services(cls, redis, fddict_col_no: str):
|
|
"""
|
|
从缓存获取字典数据列表信息service
|
|
:param redis: redis对象
|
|
:param dict_type: 字典类型
|
|
:return: 字典数据列表信息对象
|
|
"""
|
|
result = []
|
|
dict_data_list_result = fddict_col_no
|
|
print('测试使用，没有实际意义的方法！')
|
|
if dict_data_list_result:
|
|
result = fddict_col_no
|
|
|
|
return result
|
|
|
|
|
|
|
|
@classmethod
|
|
def add_fddict_services(cls, result_db: Session, page_object: FddictModel):
|
|
"""
|
|
新增代码信息service
|
|
:param result_db: orm对象
|
|
:param page_object: 新增代码对象
|
|
:return: 新增代码校验结果
|
|
"""
|
|
fddict = FddictDao.get_fddict_detail_by_info(result_db, FddictModel(**dict(fddict_name=page_object.fddict_name)))
|
|
if fddict:
|
|
result = dict(is_success=False, message='代码名称已存在')
|
|
else:
|
|
try:
|
|
FddictDao.add_fddict_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 CrudFddictResponse(**result)
|
|
|
|
@classmethod
|
|
def edit_fddict_services(cls, result_db: Session, page_object: FddictModel):
|
|
"""
|
|
编辑代码信息service
|
|
:param result_db: orm对象
|
|
:param page_object: 编辑代码信息对象
|
|
:return: 编辑代码信息校验结果
|
|
"""
|
|
edit_fddict = page_object.dict(exclude_unset=True)
|
|
|
|
fddict_info = cls.detail_fddict_services(result_db, edit_fddict.get('fddict_id'))
|
|
if fddict_info:
|
|
if fddict_info.fddict_id != page_object.fddict_id:
|
|
fddict = FddictDao.get_fddict_by_info(result_db, FddictModel(**dict(fddict_id=page_object.fddict_id)))
|
|
if fddict:
|
|
result = dict(is_success=False, message='字段名已存在')
|
|
return CrudFddictResponse(**result)
|
|
try:
|
|
FddictDao.edit_fddict_dao(result_db, edit_fddict)
|
|
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 CrudFddictResponse(**result)
|
|
|
|
@classmethod
|
|
def delete_fddict_services(cls, result_db: Session, page_object: DeleteFddictModel):
|
|
"""
|
|
删除代码信息service
|
|
:param result_db: orm对象
|
|
:param page_object: 删除代码对象
|
|
:return: 删除代码校验结果
|
|
"""
|
|
if page_object.fddict_ids.split(','):
|
|
fddict_id_list = page_object.fddict_ids.split(',')
|
|
try:
|
|
for fddict_id in fddict_id_list:
|
|
fddict_id_dict = dict(fddict_id=fddict_id)
|
|
FddictDao.delete_fddict_dao(result_db, FddictModel(**fddict_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 CrudFddictResponse(**result)
|
|
|
|
@classmethod
|
|
def detail_fddict_services(cls, result_db: Session, fddict_id: int):
|
|
"""
|
|
获取代码详细信息service
|
|
:param result_db: orm对象
|
|
:param fddict_id: 代码id
|
|
:return: 代码id对应的信息
|
|
"""
|
|
fddict = FddictDao.get_fddict_detail_by_id(result_db, fddict_id=fddict_id)
|
|
|
|
return fddict
|
|
|
|
@staticmethod
|
|
def export_fddict_list_services(fddict_list: List):
|
|
"""
|
|
导出代码信息service
|
|
:param fddict_list: 代码信息列表
|
|
:return: 代码信息对应excel的二进制数据
|
|
"""
|
|
# 创建一个映射代码，将英文键映射到中文键
|
|
mapping_dict = {
|
|
"fddict_id": "代码序号",
|
|
"fddict_col_no": "代码编号",
|
|
"fddict_col_cn": "代码中文名称",
|
|
"fddict_code": "代码码值",
|
|
"fddict_name": "码值说明",
|
|
"fddict_col_en": "代码别名",
|
|
"fddict_sort": "码值排序",
|
|
"remark": "业务描述",
|
|
"status": "状态",
|
|
"create_by": "创建者",
|
|
"create_time": "创建时间",
|
|
"update_by": "更新者",
|
|
"update_time": "更新时间",
|
|
}
|
|
|
|
data = [FddictModel(**vars(row)).dict() for row in fddict_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_fddict_services(cls, result_db: Session, fddict_import: ImportFddictModel, current_user: CurrentUserInfoServiceResponse):
|
|
"""
|
|
批量导入代码表service
|
|
:param fddict_import: 代码表导入参数对象
|
|
:param result_db: orm对象
|
|
:param current_fddict: 当前代码表对象
|
|
:return: 批量导入代码表结果
|
|
"""
|
|
header_dict = {
|
|
"代码序号": "fddict_id",
|
|
"代码编号": "fddict_col_no",
|
|
"代码中文名称": "fddict_col_cn",
|
|
"代码码值": "fddict_code",
|
|
"码值说明": "fddict_name",
|
|
"代码别名": "fddict_col_en",
|
|
"码值排序": "fddict_sort",
|
|
"业务描述": "remark",
|
|
"状态": "status"
|
|
}
|
|
|
|
filepath = get_filepath_from_url(fddict_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_fddict = FddictModel(
|
|
**dict(
|
|
fddict_id=row['fddict_id'],
|
|
fddict_col_no = str(row['fddict_col_no']).replace('nan', ''),
|
|
fddict_col_cn = str(row['fddict_col_cn']).replace('nan', ''),
|
|
fddict_code = str(row['fddict_code']).replace('nan', ''),
|
|
fddict_name = str(row['fddict_name']).replace('nan', ''),
|
|
fddict_col_en = str(row['fddict_col_en']).replace('nan', ''),
|
|
fddict_sort = str(row['fddict_sort']).replace('nan', ''),
|
|
remark = str(row['remark']).replace('nan', ''),
|
|
status=row['status'],
|
|
create_by=current_user.user.user_name,
|
|
update_by=current_user.user.user_name,
|
|
)
|
|
)
|
|
#print(12345,fddict_import,3456,row['fddict_id'])
|
|
# fddict_info = FddictDao.get_fddict_by_info(result_db, FddictModel(**dict(fddict_name=row['fddict_name'])))
|
|
fddict_info = FddictDao.get_fddict_by_info(result_db, FddictModel(**dict(fddict_id=row['fddict_id'])))
|
|
|
|
|
|
if fddict_info:
|
|
if fddict_import.is_update:
|
|
edit_fddict = FddictModel(
|
|
**dict(
|
|
fddict_id=row['fddict_id'],
|
|
fddict_col_no = str(row['fddict_col_no']).replace('nan', ''),
|
|
fddict_col_cn = str(row['fddict_col_cn']).replace('nan', ''),
|
|
fddict_code = str(row['fddict_code']).replace('nan', ''),
|
|
fddict_name = str(row['fddict_name']).replace('nan', ''),
|
|
fddict_col_en = str(row['fddict_col_en']).replace('nan', ''),
|
|
fddict_sort = str(row['fddict_sort']).replace('nan', ''),
|
|
remark = str(row['remark']).replace('nan', ''),
|
|
status=row['status'],
|
|
update_by=current_user.user.user_name
|
|
)
|
|
).dict(exclude_unset=True)
|
|
|
|
FddictDao.edit_fddict_dao(result_db, edit_fddict)
|
|
else:
|
|
add_error_result.append(f"{count}.用户账号{row['user_name']}已存在")
|
|
else:
|
|
FddictDao.add_fddict_dao(result_db, add_fddict)
|
|
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 CrudFddictResponse(**result)
|
|
|
|
@staticmethod
|
|
def get_fddict_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
|