from sqlalchemy import and_, or_, desc, func from sqlalchemy.orm import Session from module_admin.entity.do.sscf_do import SysSscf from module_admin.entity.do.dasset_do import SysDasset from module_admin.entity.vo.sscf_vo import SscfModel, CurrentSscfInfo, SscfQueryModel from utils.time_format_util import object_format_datetime, list_format_datetime, format_datetime_dict_list from datetime import datetime, time from typing import Union, List class SscfDao: """ 短句配置模块数据库操作层 """ @classmethod def get_sscf_by_name(cls, db: Session, onum: int): """ 根据字段名获取字段信息 :param db: orm对象 :param onum: 序号id :return: 当前表名称信息对象 """ query_sscf_info = db.query(SysSscf) \ .filter(SysSscf.status == 0, SysSscf.del_flag == 0, SysSscf.onum == onum) \ .order_by(desc(SysSscf.create_time)).distinct().first() return query_sscf_info @classmethod def get_sscf_by_info(cls, db: Session, sscf: SscfModel): """ 根据字段参数获取字段信息 :param db: orm对象 :param sscf: 表名称参数 :return: 当前表名称参数的表名称信息对象 """ query_sscf_info = db.query(SysSscf) \ .filter(SysSscf.del_flag == 0, SysSscf.onum == sscf.onum) \ .order_by(desc(SysSscf.create_time)).distinct().first() return query_sscf_info @classmethod def get_sscf_by_id(cls, db: Session, onum: int): """ 根据onum获取字段信息 :param db: orm对象 :param onum: 序号id :return: 当前onum的信息对象 """ query_sscf_basic_info = db.query(SysSscf) \ .filter(SysSscf.status == 0, SysSscf.del_flag == 0, SysSscf.onum == onum) \ .distinct().first() query_sscf_dasset_info = db.query(SysDasset).select_from(SysSscf) \ .filter( SysSscf.status == 0, SysSscf.del_flag == 0, SysSscf.onum == onum) \ .join(SysDasset, and_(SysSscf.dasset_id == SysDasset.dasset_id, SysDasset.status == 0, SysDasset.del_flag == 0)) \ .distinct().first() results = dict( sscf_basic_info=object_format_datetime(query_sscf_basic_info), sscf_dasset_info=object_format_datetime(query_sscf_dasset_info) ) return CurrentSscfInfo(**results) @classmethod def get_sscf_detail_by_id(cls, db: Session, onum: int): """ 根据onum获取字段详细信息 :param db: orm对象 :param onum: 序号id :return: 当前onum的信息对象 """ query_sscf_basic_info = db.query(SysSscf) \ .filter(SysSscf.del_flag == 0, SysSscf.onum == onum) \ .distinct().first() query_sscf_dasset_info = db.query(SysDasset).select_from(SysSscf) \ .filter(SysSscf.del_flag == 0, SysSscf.onum == onum) \ .join(SysDasset, and_(SysSscf.dasset_id == SysDasset.dasset_id, SysDasset.status == 0, SysDasset.del_flag == 0)) \ .distinct().first() results = dict( sscf_basic_info=object_format_datetime(query_sscf_basic_info), sscf_dasset_info=object_format_datetime(query_sscf_dasset_info) ) return CurrentSscfInfo(**results) @classmethod def get_sscf_list(cls, db: Session, query_object: SscfQueryModel, data_scope_sql: str): """ 根据查询参数获取字段列表信息 :param db: orm对象 :param query_object: 查询参数对象 :param data_scope_sql: 数据权限对应的查询sql语句 :return: 字段列表信息对象 """ sscf_list = db.query(SysSscf, SysDasset) \ .filter(SysSscf.del_flag == 0, or_(SysSscf.dasset_id == query_object.dasset_id, SysSscf.dasset_id.in_( db.query(SysDasset.dasset_id).filter(func.find_in_set(query_object.dasset_id, SysDasset.dasset_ancestors)) )) if query_object.dasset_id else True, SysSscf.onum.like(f'%{query_object.onum}%') if query_object.onum else True, #SysSscf.keywords.like(f'%{query_object.keywords}%') if query_object.keywords else True, SysSscf.keyword.like(f'%{query_object.keyword}%') if query_object.keyword else True, SysSscf.algorithm.like(f'%{query_object.algorithm}%') if query_object.algorithm else True, SysSscf.order.like(f'%{query_object.order}%') if query_object.order else True, SysSscf.whole_sentence.like(f'%{query_object.whole_sentence}%') if query_object.whole_sentence else True, SysSscf.type.like(f'%{query_object.type}%') if query_object.type else True, SysSscf.supp_expl.like(f'%{query_object.supp_expl}%') if query_object.supp_expl else True, SysSscf.status == query_object.status if query_object.status else True, SysSscf.create_time.between( datetime.combine(datetime.strptime(query_object.create_time_start, '%Y-%m-%d'), time(00, 00, 00)), datetime.combine(datetime.strptime(query_object.create_time_end, '%Y-%m-%d'), time(23, 59, 59))) if query_object.create_time_start and query_object.create_time_end else True, eval(data_scope_sql) ) \ .outerjoin(SysDasset, and_(SysSscf.dasset_id == SysDasset.dasset_id, SysDasset.status == 0, SysDasset.del_flag == 0)) \ .distinct().all() result_list: List[Union[dict, None]] = [] if sscf_list: for item in sscf_list: obj = dict( onum=item[0].onum, dasset_id=item[0].dasset_id, dasset_name=item[1].dasset_name if item[1] else '', #keywords=item[0].keywords, keyword=item[0].keyword, algorithm=item[0].algorithm, order=item[0].order, whole_sentence=item[0].whole_sentence, type=item[0].type, supp_expl=item[0].supp_expl, # bak1=item[0].bak1, # bak2=item[0].bak2, # bak3=item[0].bak3, # bak4=item[0].bak4, # bak5=item[0].bak5, status=item[0].status, del_flag=item[0].del_flag, create_by=item[0].create_by, create_time=item[0].create_time, update_by=item[0].update_by, update_time=item[0].update_time ) result_list.append(obj) return format_datetime_dict_list(result_list) @classmethod def add_sscf_dao(cls, db: Session, sscf: SscfModel): """ 新增字段数据库操作 :param db: orm对象 :param sscf: 字段对象 :return: 新增校验结果 """ db_sscf = SysSscf(**sscf.dict()) db.add(db_sscf) db.commit() db.flush() # 假设 SysSscf 是一个模型类，并且它有 name1, name2, name3 这三个字段 db.query(SysSscf) \ .update({SysSscf.keywords: func.concat('{"keyword": ["',SysSscf.keyword,'"], "algorithm": "',SysSscf.algorithm,'", "order": "',SysSscf.order,'", "whole_sentence": "',SysSscf.whole_sentence,'"}')}) return db_sscf @classmethod def edit_sscf_dao(cls, db: Session, sscf: dict): """ 编辑字段数据库操作 :param db: orm对象 :param sscf: 需要更新的字段字典 :return: 编辑校验结果 """ db.query(SysSscf) \ .filter(SysSscf.keyword == sscf.get('keyword')) \ .update(sscf) # 假设 SysSscf 是一个模型类，并且它有 name1, name2, name3 这三个字段 db.query(SysSscf) \ .filter(SysSscf.keyword == sscf.get('keyword')) \ .update({SysSscf.keywords: func.concat('{"keyword": ["',SysSscf.keyword,'"], "algorithm": "',SysSscf.algorithm,'", "order": "',SysSscf.order,'", "whole_sentence": "',SysSscf.whole_sentence,'"}')}) #.update({SysSscf.keywords: func.concat(SysSscf.keyword, SysSscf.algorithm,SysSscf.order, SysSscf.whole_sentence)}) db.query(SysSscf) \ .filter(SysSscf.onum == sscf.get('onum')) \ .update(sscf) @classmethod def delete_sscf_dao(cls, db: Session, sscf: SscfModel): """ 删除字段数据库操作 :param db: orm对象 :param sscf: 字段对象 :return: """ db.query(SysSscf) \ .filter(SysSscf.onum == sscf.onum) \ .delete() #.update({SysSscf.del_flag: '2', SysSscf.update_by: sscf.update_by, SysSscf.update_time: sscf.update_time}) @classmethod def get_sscf_dasset_info(cls, db: Session, dasset_id: int): dasset_basic_info = db.query(SysDasset) \ .filter(SysDasset.dasset_id == dasset_id, SysDasset.status == 0, SysDasset.del_flag == 0 ) \ .first() return dasset_basic_info @classmethod def get_sscf_by_info_imp(cls, db: Session, sscf: SscfModel): """ 根据字段参数获取字段信息 :param db: orm对象 :param sscf: 表名称参数 :return: 当前表名称参数的表名称信息对象 """ query_sscf_info_imp = db.query(SysSscf) \ .filter(SysSscf.del_flag == 0, SysSscf.keyword == sscf.keyword) \ .order_by(desc(SysSscf.create_time)).distinct().first() return query_sscf_info_imp