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.
266 lines
10 KiB
266 lines
10 KiB
1 month ago
|
from module_admin.entity.vo.user_vo import *
|
||
|
|
from module_admin.entity.vo.cdplb_vo import *
|
||
|
|
|
||
|
|
from utils.common_util import export_list2excel
|
||
|
|
from utils.common_util import *
|
||
|
|
|
||
|
|
from module_admin.dao.cdplb_dao import *
|
||
|
|
|
||
|
|
|
||
|
|
class CdplbService:
|
||
|
|
"""
|
||
|
|
批量词典配置表管理模块服务层
|
||
|
|
"""
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def get_cdplb_select_option_services(cls, result_db: Session):
|
||
|
|
"""
|
||
|
|
获取批量词典配置表列表不分页信息service
|
||
|
|
:param result_db: orm对象
|
||
|
|
:return: 批量词典配置表列表不分页信息对象
|
||
|
|
"""
|
||
|
|
cdplb_list_result = CdplbDao.get_cdplb_select_option_dao(result_db)
|
||
|
|
|
||
|
|
return cdplb_list_result
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def get_cdplb_list_services(cls, result_db: Session, query_object: CdplbModel):
|
||
|
|
"""
|
||
|
|
获取批量词典配置表列表信息service
|
||
|
|
:param result_db: orm对象
|
||
|
|
:param query_object: 查询参数对象
|
||
|
|
:return: 批量词典配置表列表信息对象
|
||
|
|
"""
|
||
|
|
cdplb_list_result = CdplbDao.get_cdplb_list(result_db, query_object)
|
||
|
|
|
||
|
|
return cdplb_list_result
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def query_cdplb_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_cdplb_services(cls, result_db: Session, page_object: CdplbModel):
|
||
|
|
"""
|
||
|
|
新增批量词典配置表信息service
|
||
|
|
:param result_db: orm对象
|
||
|
|
:param page_object: 新增批量词典配置表对象
|
||
|
|
:return: 新增批量词典配置表校验结果
|
||
|
|
"""
|
||
|
|
cdplb = CdplbDao.get_cdplb_detail_by_info(result_db, CdplbModel(**dict(bath_obj_fld_name=page_object.bath_obj_fld_name)))
|
||
|
|
if cdplb:
|
||
|
|
result = dict(is_success=False, message='批量词典配置表名称已存在')
|
||
|
|
else:
|
||
|
|
try:
|
||
|
|
CdplbDao.add_cdplb_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 CrudCdplbResponse(**result)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def edit_cdplb_services(cls, result_db: Session, page_object: CdplbModel):
|
||
|
|
"""
|
||
|
|
编辑批量词典配置表信息service
|
||
|
|
:param result_db: orm对象
|
||
|
|
:param page_object: 编辑批量词典配置表信息对象
|
||
|
|
:return: 编辑批量词典配置表信息校验结果
|
||
|
|
"""
|
||
|
|
edit_cdplb = page_object.dict(exclude_unset=True)
|
||
|
|
|
||
|
|
cdplb_info = cls.detail_cdplb_services(result_db, edit_cdplb.get('onum'))
|
||
|
|
if cdplb_info:
|
||
|
|
if cdplb_info.onum != page_object.onum:
|
||
|
|
cdplb = CdplbDao.get_cdplb_by_info(result_db, CdplbModel(**dict(onum=page_object.onum)))
|
||
|
|
if cdplb:
|
||
|
|
result = dict(is_success=False, message='字段名已存在')
|
||
|
|
return CrudCdplbResponse(**result)
|
||
|
|
try:
|
||
|
|
CdplbDao.edit_cdplb_dao(result_db, edit_cdplb)
|
||
|
|
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 CrudCdplbResponse(**result)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def delete_cdplb_services(cls, result_db: Session, page_object: DeleteCdplbModel):
|
||
|
|
"""
|
||
|
|
删除批量词典配置表信息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)
|
||
|
|
CdplbDao.delete_cdplb_dao(result_db, CdplbModel(**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 CrudCdplbResponse(**result)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def detail_cdplb_services(cls, result_db: Session, onum: int):
|
||
|
|
"""
|
||
|
|
获取批量词典配置表详细信息service
|
||
|
|
:param result_db: orm对象
|
||
|
|
:param onum: 批量词典配置表id
|
||
|
|
:return: 批量词典配置表id对应的信息
|
||
|
|
"""
|
||
|
|
cdplb = CdplbDao.get_cdplb_detail_by_id(result_db, onum=onum)
|
||
|
|
|
||
|
|
return cdplb
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def export_cdplb_list_services(cdplb_list: List):
|
||
|
|
"""
|
||
|
|
导出批量词典配置表信息service
|
||
|
|
:param cdplb_list: 批量词典配置表信息列表
|
||
|
|
:return: 批量词典配置表信息对应excel的二进制数据
|
||
|
|
"""
|
||
|
|
# 创建一个映射批量词典配置表，将英文键映射到中文键
|
||
|
|
mapping_dict = {
|
||
|
|
"onum": "序号",
|
||
|
|
"bath_obj_tab_name": "批量对象表名",
|
||
|
|
"bath_obj_fld_name": "批里对象字段名",
|
||
|
|
"freq": "频次",
|
||
|
|
"pos": "词性",
|
||
|
|
"pos_name": "词性名称",
|
||
|
|
"status": "状态",
|
||
|
|
#"create_by": "创建者",
|
||
|
|
#"create_time": "创建时间",
|
||
|
|
"update_by": "负责人",
|
||
|
|
"update_time": "更新时间",
|
||
|
|
}
|
||
|
|
|
||
|
|
data = [CdplbModel(**vars(row)).dict() for row in cdplb_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_cdplb_services(cls, result_db: Session, cdplb_import: ImportCdplbModel, current_user: CurrentUserInfoServiceResponse):
|
||
|
|
"""
|
||
|
|
批量导入批量词典配置表表service
|
||
|
|
:param cdplb_import: 批量词典配置表表导入参数对象
|
||
|
|
:param result_db: orm对象
|
||
|
|
:param current_cdplb: 当前批量词典配置表表对象
|
||
|
|
:return: 批量导入批量词典配置表表结果
|
||
|
|
"""
|
||
|
|
header_dict = {
|
||
|
|
#"序号": "onum",
|
||
|
|
"批量对象表名": "bath_obj_tab_name",
|
||
|
|
"批里对象字段名": "bath_obj_fld_name",
|
||
|
|
"频次": "freq",
|
||
|
|
"词性": "pos",
|
||
|
|
"词性名称": "pos_name",
|
||
|
|
"状态": "status"
|
||
|
|
}
|
||
|
|
|
||
|
|
filepath = get_filepath_from_url(cdplb_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_cdplb = CdplbModel(
|
||
|
|
**dict(
|
||
|
|
#onum=row['onum'],
|
||
|
|
bath_obj_tab_name=row['bath_obj_tab_name'],
|
||
|
|
bath_obj_fld_name=row['bath_obj_fld_name'],
|
||
|
|
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,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
#cdplb_info = CdplbDao.get_cdplb_by_info(result_db, CdplbModel(**dict(onum=row['onum'])))
|
||
|
|
# query_cdplb_info_imp bath_obj_tab_name, bath_obj_fld_name
|
||
|
|
cdplb_infos = CdplbDao.get_cdplb_by_info_imp(result_db, CdplbModel(**dict(bath_obj_tab_name=row['bath_obj_tab_name'],bath_obj_fld_name=row['bath_obj_fld_name'])))
|
||
|
|
|
||
|
|
if cdplb_infos:
|
||
|
|
if cdplb_import.is_update:
|
||
|
|
edit_cdplb = CdplbModel(
|
||
|
|
**dict(
|
||
|
|
#onum=row['onum'],
|
||
|
|
bath_obj_tab_name=row['bath_obj_tab_name'],
|
||
|
|
bath_obj_fld_name=row['bath_obj_fld_name'],
|
||
|
|
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)
|
||
|
|
|
||
|
|
CdplbDao.edit_cdplb_dao(result_db, edit_cdplb)
|
||
|
|
else:
|
||
|
|
add_error_result.append(f"{count}.用户账号{row['user_name']}已存在")
|
||
|
|
else:
|
||
|
|
CdplbDao.add_cdplb_dao(result_db, add_cdplb)
|
||
|
|
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 CrudCdplbResponse(**result)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def get_cdplb_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
|