from sqlalchemy.ext.asyncio import AsyncSession from module_admin.dao.metadata_config_dao import MetadataConfigDao from module_admin.entity.vo.common_vo import CrudResponseModel from module_admin.entity.vo.metadata_config_vo import MetadataClasModel,SecuBizConfigAddModel,SecuBizConfigRelaModel,\ SecuBizPermiConfigBatchModel,MetadataClasPageQueryModel,MetadataSecModel,MetadataSecPageQueryModel,SecuBizConfigModel ,\ SecuBizConfigQueryModel,SecuBizPermiConfigModel, SecuBizPermiConfigQueryModel,\ TaskBizConfigAddModel, TaskBizConfigQueryModel,TaskBizConfigModel,TaskBizConfigRelaModel from utils.common_util import CamelCaseUtil from module_admin.entity.do.metadata_config_do import SecuBizConfigRela,TaskBizConfigRela # ORM 类 from exceptions.exception import ServiceException,ServiceWarning import uuid from collections import defaultdict from module_admin.entity.vo.data_ast_content_vo import DataCatalogPageQueryModel, DeleteDataCatalogModel,DataCatalogResponseWithChildren,DataCatalogMovedRequest,DataCatalogMergeRequest,DataCatalogChild,DataCatalogMoverelRequest from module_admin.entity.vo.dataSource_vo import ProcessDefinition,ParmScheduleVo,ProcessInstancePage,ParmSchedule from typing import List from datetime import datetime from module_admin.entity.vo.metaprocessconfig_vo import MetaprocessconfigQueryModel,MetaprocessconfigModel from module_admin.service.metaprocessconfig_service import MetaprocessconfigService import requests from fastapi import Request,Depends import json import re from config.enums import RedisInitKeyConfig from config.env import AppConfig from module_admin.entity.vo.user_vo import CurrentUserModel from module_admin.entity.vo.metatask_vo import DeleteMetataskModel from datetime import datetime class MetadataConfigService: """ 元数据分类管理 Service """ @classmethod async def get_metadata_clas_list_services( cls, query_db: AsyncSession, query_object: MetadataClasPageQueryModel, is_page: bool = False ): """ 查询元数据分类列表 """ result = await MetadataConfigDao.get_metadata_clas_list(query_db, query_object, is_page) return result @classmethod async def add_metadata_clas_services( cls, query_db: AsyncSession, page_object: MetadataClasModel ): """ 新增元数据分类，带 clas_onum 主键唯一性校验 """ try: # 校验 clas_onum 是否已存在 existing = await MetadataConfigDao.get_clas_by_onum(query_db, page_object.clas_onum) if existing: raise ServiceException(message=f"主键重复，clas_onum={page_object.clas_onum} 已存在") # 默认临时 if not page_object.belt_batch_content: page_object.belt_batch_content=2 await MetadataConfigDao.add_metadata_clas_dao(query_db, page_object) await query_db.commit() return CrudResponseModel(is_success=True, message="新增成功") except Exception as e: await query_db.rollback() raise e @classmethod async def edit_metadata_clas_services( cls, query_db: AsyncSession, page_object: MetadataClasModel ): """ 编辑元数据分类，包含 clas_onum 唯一性校验（如果主键被修改） """ edit_data = page_object.model_dump(exclude_unset=True) # 当前记录旧主键（假设表单传入） original_onum = page_object.clas_onum # 你需要在 model 中加上这个字段 # 查询当前记录是否存在 existing = await cls.get_metadata_clas_detail_services(query_db, original_onum) if not existing: raise ServiceException(message="元数据标签不存在") # 如果修改了 clas_onum，则校验唯一性 if page_object.clas_onum != original_onum: duplicated = await MetadataConfigDao.get_clas_by_onum(query_db, page_object.clas_onum) if duplicated: raise ServiceException(message=f"clas_onum={page_object.clas_onum} 已存在，不能重复") try: await MetadataConfigDao.edit_metadata_clas_dao(query_db, original_onum, edit_data) await query_db.commit() return CrudResponseModel(is_success=True, message="更新成功") except Exception as e: await query_db.rollback() raise e @classmethod async def delete_metadata_clas_services( cls, query_db: AsyncSession, clas_ids: str ): """ 删除元数据分类（支持多个ID用逗号分隔） """ if not clas_ids: raise ServiceException(message="传入的分类ID为空") id_list = [int(id_str) for id_str in clas_ids.split(",") if id_str.isdigit()] if not id_list: raise ServiceException(message="无效的分类ID列表") try: await MetadataConfigDao.delete_metadata_clas_dao(query_db, id_list) await query_db.commit() return CrudResponseModel(is_success=True, message="删除成功") except Exception as e: await query_db.rollback() raise e @classmethod async def get_metadata_clas_detail_services(cls, query_db: AsyncSession, clas_id: int): """ 查询元数据分类详情 """ result = await MetadataConfigDao.get_clas_detail_by_id(query_db, clas_id) if result: return MetadataClasModel(**CamelCaseUtil.transform_result(result)) else: return MetadataClasModel(**dict()) @classmethod async def get_metadata_sec_list_services( cls, query_db: AsyncSession, query_object: MetadataSecPageQueryModel, is_page: bool = False ): """ 查询数据安全配置列表 """ result = await MetadataConfigDao.get_metadata_sec_list(query_db, query_object, is_page) return result @classmethod async def get_metadata_sec_option_list_services(cls, query_db: AsyncSession): """ 查询数据安全配置列表 """ result = await MetadataConfigDao.get_metadata_sec_option_list(query_db) return CamelCaseUtil.transform_result(result) @classmethod async def add_metadata_sec_services(cls, query_db: AsyncSession, page_object: MetadataSecModel): """ 新增数据安全配置 """ try: # 校验 secLevelSummary 是否已存在 existing = await MetadataConfigDao.get_by_sec_level_summary( query_db, page_object.sec_level_summary,page_object.onum) if existing: raise ServiceWarning(message=page_object.sec_level_summary+"该等级简称已存在，不能重复") page_object.onum = str(uuid.uuid4()) await MetadataConfigDao.add_metadata_sec_dao(query_db, page_object) await query_db.commit() return CrudResponseModel(is_success=True, message="新增成功") except Exception as e: await query_db.rollback() raise e @classmethod async def edit_metadata_sec_services(cls, query_db: AsyncSession, page_object: MetadataSecModel): """ 编辑数据安全配置 """ edit_data = page_object.model_dump(exclude_unset=True) info = await cls.get_metadata_sec_detail_services(query_db, page_object.onum) if info.onum: try: # 校验 secLevelSummary 是否重复（排除当前记录） existing = await MetadataConfigDao.get_by_sec_level_summary( query_db, page_object.sec_level_summary,page_object.onum) if existing: raise ServiceWarning(message=page_object.sec_level_summary+"该等级简称已存在，不能重复") await MetadataConfigDao.edit_metadata_sec_dao(query_db, page_object.onum, edit_data) await query_db.commit() return CrudResponseModel(is_success=True, message="更新成功") except Exception as e: await query_db.rollback() raise e else: raise ServiceException(message="数据安全配置不存在") @classmethod async def delete_metadata_sec_services(cls, query_db: AsyncSession, onum_list: str): """ 删除数据安全配置（支持多个编号用逗号分隔） """ if not onum_list: raise ServiceException(message="传入的编号为空") id_list = [onum.strip() for onum in onum_list.split(",") if onum.strip()] if not id_list: raise ServiceException(message="无效的编号列表") try: await MetadataConfigDao.delete_metadata_sec_dao(query_db, id_list) await query_db.commit() return CrudResponseModel(is_success=True, message="删除成功") except Exception as e: await query_db.rollback() raise e @classmethod async def get_metadata_sec_detail_services(cls, query_db: AsyncSession, onum: str): """ 查询数据安全配置详情 """ result = await MetadataConfigDao.get_sec_detail_by_id(query_db, onum) if result: return MetadataSecModel(**CamelCaseUtil.transform_result(result)) else: return MetadataSecModel(**dict()) # ----------- t_secu_biz_config 服务方法 ----------- @classmethod async def get_biz_config_list_services( cls, db: AsyncSession, query_object: SecuBizConfigQueryModel, is_page: bool = False ): result = await MetadataConfigDao.get_biz_config_list(db, query_object, is_page) return result @classmethod async def add_biz_config_services(cls, db: AsyncSession, page_object: SecuBizConfigAddModel): try: configMode = SecuBizConfigModel() configMode.onum=page_object.onum configMode.biz_name=page_object.biz_name configMode.risk_lvl=page_object.risk_lvl configMode.isStop=page_object.isStop configMode.create_by=page_object.create_by configMode.update_by=page_object.update_by configMode.update_time=page_object.update_time configMode.create_time=page_object.create_time obj=await MetadataConfigDao.add_biz_config(db, configMode) records: List[SecuBizConfigRela] = [] for tab_onum in page_object.tab_onum_list: record = SecuBizConfigRela() record.biz_onum = obj.onum record.tab_onum = tab_onum record.create_by = obj.create_by record.create_time = obj.create_time records.append(record) await MetadataConfigDao.add_batch_biz_rela_dao(db,records) await db.commit() return CrudResponseModel(is_success=True, message="新增成功") except Exception as e: await db.rollback() raise e @classmethod async def edit_biz_config_services(cls, db: AsyncSession, page_object: SecuBizConfigAddModel): configMode = SecuBizConfigModel() configMode.onum=page_object.onum configMode.biz_name=page_object.biz_name configMode.risk_lvl=page_object.risk_lvl configMode.isStop=page_object.isStop configMode.create_by=page_object.create_by configMode.update_by=page_object.update_by configMode.update_time=page_object.update_time configMode.create_time=page_object.create_time edit_data = configMode.model_dump(exclude_unset=True) info = await cls.get_biz_config_detail_services(db, page_object.onum) if info.onum: try: await MetadataConfigDao.edit_biz_config(db, page_object.onum, edit_data) await MetadataConfigDao.delete_biz_rela_dao(db, page_object.onum) records: List[SecuBizConfigRela] = [] for tab_onum in page_object.tab_onum_list: record = SecuBizConfigRela() record.biz_onum = page_object.onum record.tab_onum = tab_onum record.create_by = page_object.update_by record.create_time = page_object.update_time records.append(record) await MetadataConfigDao.add_batch_biz_rela_dao(db,records) await db.commit() return CrudResponseModel(is_success=True, message="更新成功") except Exception as e: await db.rollback() raise e else: raise ServiceException(message="业务配置不存在") @classmethod async def delete_biz_config_services(cls, db: AsyncSession, onum_list: str): if not onum_list: raise ServiceException(message="传入的编号为空") id_list = [onum.strip() for onum in onum_list.split(",") if onum.strip()] if not id_list: raise ServiceException(message="无效的编号列表") try: await MetadataConfigDao.delete_biz_config(db, id_list) await MetadataConfigDao.delete_biz_rela_batch(db, id_list) await db.commit() return CrudResponseModel(is_success=True, message="删除成功") except Exception as e: await db.rollback() raise e @classmethod async def get_biz_config_detail_services(cls, db: AsyncSession, onum: str): result = await MetadataConfigDao.get_biz_config_detail_by_id(db, onum) if result: return SecuBizConfigModel(**CamelCaseUtil.transform_result(result)) else: return SecuBizConfigModel(**dict()) # ----------- t_secu_biz_permi_config 服务方法 ----------- @classmethod async def get_biz_permi_config_list_services( cls, db: AsyncSession, query_object: SecuBizPermiConfigQueryModel, is_page: bool = False ): result = await MetadataConfigDao.get_biz_permi_config_list(db, query_object, is_page) return result @classmethod async def add_biz_permi_config_services( cls, db: AsyncSession, batch_model: SecuBizPermiConfigBatchModel, create_by: str, create_time: datetime ) -> CrudResponseModel: try: records: List[SecuBizPermiConfigModel] = [] for biz_onum in batch_model.biz_onum_list: record = SecuBizPermiConfigModel() record.biz_onum = biz_onum record.obj_type = batch_model.obj_type record.obj_value = batch_model.obj_value record.obj_name = batch_model.obj_name record.isStop = batch_model.isStop record.create_by = create_by record.create_time = create_time records.append(record) await MetadataConfigDao.add_biz_permi_config_batch(db, records) await db.commit() return CrudResponseModel(is_success=True, message="新增成功") except Exception as e: await db.rollback() raise e @classmethod async def edit_biz_permi_config_services(cls, db: AsyncSession, page_object: SecuBizPermiConfigModel): edit_data = page_object.model_dump(exclude_unset=True) info = await cls.get_biz_permi_config_detail_services(db, page_object.onum) if info.onum: try: await MetadataConfigDao.edit_biz_permi_config(db, page_object.onum, edit_data) await db.commit() return CrudResponseModel(is_success=True, message="更新成功") except Exception as e: await db.rollback() raise e else: raise ServiceException(message="权限配置不存在") @classmethod async def delete_biz_permi_config_services(cls, db: AsyncSession, onum_list: str): if not onum_list: raise ServiceException(message="传入的编号为空") id_list = [onum.strip() for onum in onum_list.split(",") if onum.strip()] if not id_list: raise ServiceException(message="无效的编号列表") try: await MetadataConfigDao.delete_biz_permi_config(db, id_list) await db.commit() return CrudResponseModel(is_success=True, message="删除成功") except Exception as e: await db.rollback() raise e @classmethod async def get_biz_permi_config_detail_services(cls, db: AsyncSession, onum: str): result = await MetadataConfigDao.get_biz_permi_config_detail_by_id(db, onum) if result: return SecuBizPermiConfigModel(**CamelCaseUtil.transform_result(result)) else: return SecuBizPermiConfigModel(**dict()) @classmethod async def get_biz_config_rela_list_services(cls, result_db: AsyncSession, biz_onum: int): ai_session_list = await MetadataConfigDao.get_biz_rela_by_biz_id(result_db, biz_onum) # 查询最新的20条 return CamelCaseUtil.transform_result(ai_session_list) # ----------元数据标签调度任务 ---------- @classmethod async def get_task_biz_config_list_services( cls, db: AsyncSession, query_object: TaskBizConfigQueryModel, is_page: bool = False ): return await MetadataConfigDao.get_task_biz_config_list(db, query_object, is_page) @classmethod async def add_task_biz_config_services(cls, db: AsyncSession, page_object: TaskBizConfigAddModel, current_user: CurrentUserModel, request: Request): try: config = TaskBizConfigModel() config.onum = page_object.onum config.biz_name = page_object.biz_name config.risk_lvl = page_object.risk_lvl # config.status = page_object.status # 映射 data_sec_lvl # config.ds_ids = page_object.ds_ids # 映射 applyType config.isStop = page_object.isStop config.create_by = page_object.create_by config.create_time = page_object.create_time config.update_by = page_object.update_by config.update_time = page_object.update_time obj = await MetadataConfigDao.add_task_biz_config(db, config) page_object.onum=obj.onum # 添加关联信息 records: List[TaskBizConfigRela] = [] for tab_onum in page_object.tab_onum_list: record = TaskBizConfigRela() record.biz_onum = obj.onum record.tab_onum = tab_onum record.create_by = obj.create_by record.create_time = obj.create_time records.append(record) await MetadataConfigDao.add_batch_task_rela_dao(db, records) # 获取流程配置 processconfig = MetaprocessconfigQueryModel() processconfig.db_type = "BizConfig" processConfigList = await MetaprocessconfigService.get_metaprocessconfig_list_services(db, processconfig, False) message = await cls.biz_process_defind_change_add(request, processConfigList, page_object, current_user) if "成功" not in message: await db.rollback() raise ServiceException(message=f'新增批次标签任务 {page_object.biz_name} 失败，dolphinscheduler 创建失败'+message) config.onum=page_object.onum config.ds_ids=page_object.ds_ids edit_data = config.model_dump(exclude_unset=True) await MetadataConfigDao.edit_task_biz_config(db, page_object.onum, edit_data) await db.commit() return CrudResponseModel(is_success=True, message="新增成功") except Exception as e: await db.rollback() raise e @classmethod async def biz_process_defind_change_add(cls,request: Request, processConfigList: list[MetaprocessconfigModel],page_object:TaskBizConfigAddModel,current_user: CurrentUserModel): projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') url = f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/task-definition/gen-task-codes?genNum=5' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password} # 新增接口 url2=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/process-definition' headers2 = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} response = requests.get(url, headers=headers, verify=False) if response.reason == 'OK': intdsids=[] message='' dstypes=[] response_text = response.text data = json.loads(response_text) code_list = data["data"] str_list = list(map(str, code_list)) for config in processConfigList: modified_json_str = config.taskDefinitionJson.replace("18093081592672", str_list[0]).replace("sh /home/xx/code/remote_python.sh", "sh /home/xx/code/remote_python.sh "+str(page_object.onum)) modified_json_str2=config.taskRelationJson.replace("18093081592672", str_list[0]) modified_json_str3=config.locations.replace("18093081592672", str_list[0]) metaprocessconfig_dict = MetaprocessconfigModel(taskDefinitionJson=modified_json_str,# 替换taskDefinitionJson description="", # 替换工作流备注 locations=modified_json_str3,# 替换locations name =page_object.biz_name,# 替换工作流名称 timeout=config.timeout, globalParams =config.globalParams , tenantCode =config.tenantCode , taskRelationJson =modified_json_str2,# 替换taskRelationJson executionType =config.executionType , releaseState=config.releaseState ).model_dump(exclude_unset=True, by_alias=True) form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response_post0 = requests.post(url2, headers=headers2,data=form_data,verify=False) text= response_post0.text responsJson = json.loads(text) if responsJson['msg'] == 'success': dstypes.append('1') intdsids.append(responsJson['data']['code']) if message: message += ", " message += page_object.biz_name + "-批次标签新增成功" else: if message: message += ", " message += page_object.biz_name + "-批次标签新增失败" if len(intdsids)>0: page_object.ds_ids=','.join([str(i) for i in intdsids]) return message @classmethod async def edit_task_biz_config_services(cls, db: AsyncSession, page_object: TaskBizConfigAddModel,current_user: CurrentUserModel,request: Request): config = TaskBizConfigModel() config.onum = page_object.onum config.biz_name = page_object.biz_name config.risk_lvl = page_object.risk_lvl # config.status = page_object.status # 映射 data_sec_lvl # config.ds_ids = page_object.ds_ids # 映射 applyType config.isStop = page_object.isStop config.create_by = page_object.create_by config.create_time = page_object.create_time config.update_by = page_object.update_by config.update_time = page_object.update_time edit_data = config.model_dump(exclude_unset=True) info = await cls.get_task_biz_config_detail_services(db, page_object.onum) if info.onum: try: await MetadataConfigDao.edit_task_biz_config(db, page_object.onum, edit_data) await MetadataConfigDao.delete_task_rela_dao(db, page_object.onum) records: List[TaskBizConfigRela] = [] for tab_onum in page_object.tab_onum_list: record = TaskBizConfigRela() record.biz_onum = page_object.onum record.tab_onum = tab_onum record.create_by = page_object.update_by record.create_time = page_object.update_time records.append(record) await MetadataConfigDao.add_batch_task_rela_dao(db, records) message= await cls.sec_process_defind_change_update(request,page_object,info,current_user) if "成功" not in message: await db.rollback() raise ServiceException(message=f'更新批次标签任务 {page_object.biz_name} 失败，dolphinscheduler 创建失败'+message) await db.commit() return CrudResponseModel(is_success=True, message="更新成功") except Exception as e: await db.rollback() raise e else: raise ServiceException(message="任务业务配置不存在") @classmethod async def sec_process_defind_change_update(cls,request: Request,page_object:TaskBizConfigAddModel,metatask_old:TaskBizConfigAddModel,current_user: CurrentUserModel): projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') dsids=page_object.ds_ids.split(",") result_list = [{'dstype': t, 'dsid': i} for t, i in zip(dsids, dsids)] message='' # 查询接口 url=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/process-definition' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, } headers2 = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} for config in result_list: response = requests.get(f"{url}/{config['dsid']}", headers=headers, verify=False) text= response.text responsJson = json.loads(text) if responsJson['msg'] == 'success': task_def_list = responsJson['data']['taskDefinitionList'] # 遍历修改 rawScript task_def_list = responsJson['data']['taskDefinitionList'] # 再序列化为 JSON 字符串 modified_json_str = json.dumps(task_def_list, ensure_ascii=False, indent=0) getTaskRelationList=responsJson['data']['processTaskRelationList'] putTaskRelationList=[] for item in getTaskRelationList: new_item = { "name": item['name'], "preTaskCode":item['preTaskCode'] , "preTaskVersion":item['preTaskVersion'] , "postTaskCode":item['postTaskCode'] , "conditionType":item['conditionType'] , "conditionParams":item['conditionParams'] } putTaskRelationList.append(new_item) modified_json_str2= json.dumps(putTaskRelationList, ensure_ascii=False, indent=0) modified_json_str2=re.sub(r'\s+', '', modified_json_str2) metaprocessconfig_dict = MetaprocessconfigModel(taskDefinitionJson=modified_json_str,# 替换taskDefinitionJson description="", # 替换工作流备注 locations=responsJson['data']['processDefinition']['locations'] ,# 替换locations name =page_object.biz_name,# 替换工作流名称 tenantCode =responsJson['data']['processDefinition']['tenantCode'] , taskRelationJson =modified_json_str2,# 替换taskRelationJson ).model_dump(exclude_unset=True, by_alias=True) form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response_put0 = requests.put(f"{url}/{config['dsid']}", headers=headers2,data=form_data,verify=False) putText= response_put0.text responsPutJson=json.loads(putText) if responsPutJson['msg'] == 'success': if message: message += ", " message += page_object.biz_name + "-批次标签成功" else: if message: message += ", " message += page_object.biz_name + "-批次标签修改失败:"+responsPutJson['msg'] return message @classmethod async def delete_task_biz_config_services( cls, request: Request, query_db: AsyncSession, page_object: DeleteMetataskModel, current_user: CurrentUserModel ): if page_object.metatask_ids and page_object.ds_ids: metatask_id_list = page_object.metatask_ids.split(',') try: projectCode = await request.app.state.redis.get( f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode' ) # 构造请求参数 url = f'{AppConfig.ds_server_url}/dolphinscheduler/projects/{projectCode}/process-definition/batch-delete' form_data = {'codes': page_object.ds_ids} headers = { 'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded' } # 发送 POST 请求 response = requests.post(url, headers=headers, data=form_data, verify=False) responsJson = json.loads(response.text) if responsJson['success'] is True: # 删除本地元任务配置 await MetadataConfigDao.delete_task_biz_config(query_db, metatask_id_list) await MetadataConfigDao.delete_task_rela_batch(query_db, metatask_id_list) await query_db.commit() return CrudResponseModel(is_success=True, message='删除成功') else: raise ServiceException(message='ds删除失败') except Exception as e: await query_db.rollback() raise e else: raise ServiceException(message='传入参数配置id为空') @classmethod async def up_or_down_metatask_services( cls, request: Request, query_db: AsyncSession, current_user: CurrentUserModel, id: str, type: str ): metatask_info = await cls.get_task_biz_config_detail_services(query_db, id) metatask_info.update_by = current_user.user.user_name metatask_info.update_time = datetime.now() type_str: str projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') dsids=metatask_info.ds_ids.split(",") result_list = [{'dstype': t, 'dsid': i} for t, i in zip(dsids, dsids)] message='' # 查询接口 url=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/process-definition' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} for config in result_list: metaprocessconfig_dict = { 'name': metatask_info.biz_name, 'releaseState': type } form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response = requests.post(f"{url}/{config['dsid']}/release", headers=headers, data=form_data, verify=False) text= response.text responsJson = json.loads(text) if responsJson['success'] is True: message='成功！' else: raise ServiceException(message='失败'+responsJson['msg']) if type == 'OFFLINE': # 下线 type_str = '下线' metatask_info.status = 'OFFLINE' else: # 上线 type_str = '上线' metatask_info.status = 'ONLINE' edit_metatask = metatask_info.model_dump(exclude_unset=True) try: await MetadataConfigDao.edit_task_biz_config(query_db, metatask_info.onum, edit_metatask) await query_db.commit() return CrudResponseModel(is_success=True, message=message) except Exception as e: await query_db.rollback() raise e else: raise ServiceException(message='更新失败') @classmethod async def ds_metatask_services( cls, request: Request, query_db: AsyncSession, process: ParmScheduleVo,current_user: CurrentUserModel ): projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') # 先查询是否建立定时任务 getdsurl=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/schedules?pageSize=10&pageNo=1&processDefinitionCode='+str(process.processDefinitionCode) headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} getdsresponse = requests.get(getdsurl, headers=headers, verify=False) getdstext= getdsresponse.text responsJson = json.loads(getdstext) if responsJson['msg'] == 'success': if responsJson['data']['total']>0: # getds_json_list = json.dumps(responsJson['data']['totalList'], ensure_ascii=False, indent=4) getds_json_list = responsJson['data']['totalList'] for item in getds_json_list: if item['releaseState']=='ONLINE': # 先下线在删除 offdsurl=f"{AppConfig.ds_server_url}/dolphinscheduler/projects/{projectCode}/schedules/{item['id']}/offline" offresponse = requests.post(offdsurl, headers=headers, verify=False) # 删除对应的调度 deldsurl=f"{AppConfig.ds_server_url}/dolphinscheduler/projects/{projectCode}/schedules/{item['id']}?scheduleId={item['id']}" delresponse = requests.delete(deldsurl, headers=headers, verify=False) parm =ParmSchedule() parm.failureStrategy='CONTINUE' parm.warningType='NONE' parm.warningGroupId=process.warningGroupId parm.workerGroup=process.workerGroup parm.processDefinitionCode =process.processDefinitionCode parm.environmentCode=process.environmentCode parm.processInstancePriority='MEDIUM' parm.schedule = ( '{"startTime":"' + process.beginTime.strftime('%Y-%m-%d %H:%M:%S') + '", "endTime":"' + process.endTime.strftime('%Y-%m-%d %H:%M:%S') + '", "crontab":"' + process.crontab + '", "timezoneId":"Asia/Shanghai"}') url=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/schedules' # form_data = {key: str(value) for key, value in process.__dict__.items()} form_data = {key: value for key, value in parm.__dict__.items()} response = requests.post(url, headers=headers, data=form_data, verify=False) text= response.text responsJson = json.loads(text) if responsJson['msg'] == 'success': scheduleId= responsJson['data']['id'] ondsurl=f"{AppConfig.ds_server_url}/dolphinscheduler/projects/{projectCode}/schedules/{scheduleId}/online" ondsurl = requests.post(ondsurl, headers=headers, verify=False) metatask_info = await cls.get_task_biz_config_detail_services(query_db, process.metaTaskId) metatask_info.schId=scheduleId metatask_info2 = metatask_info.model_dump(exclude_unset=True) await MetadataConfigDao.edit_task_biz_config(query_db,metatask_info.onum, metatask_info2) await query_db.commit() return "调度运行成功！" else: raise ServiceException(message='运行失败！') @classmethod async def ds_metatask_delete( cls, request: Request, query_db: AsyncSession, process: ParmScheduleVo,current_user: CurrentUserModel ): projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') # 先查询是否建立定时任务 getdsurl=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/schedules?pageSize=10&pageNo=1&processDefinitionCode='+str(process.processDefinitionCode) headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} getdsresponse = requests.get(getdsurl, headers=headers, verify=False) getdstext= getdsresponse.text responsJson = json.loads(getdstext) if responsJson['msg'] == 'success': if responsJson['data']['total']>0: # getds_json_list = json.dumps(responsJson['data']['totalList'], ensure_ascii=False, indent=4) getds_json_list = responsJson['data']['totalList'] for item in getds_json_list: if item['releaseState']=='ONLINE': # 先下线在删除 offdsurl=f"{AppConfig.ds_server_url}/dolphinscheduler/projects/{projectCode}/schedules/{item['id']}/offline" offresponse = requests.post(offdsurl, headers=headers, verify=False) # 删除对应的调度 deldsurl=f"{AppConfig.ds_server_url}/dolphinscheduler/projects/{projectCode}/schedules/{item['id']}?scheduleId={item['id']}" delresponse = requests.delete(deldsurl, headers=headers, verify=False) deldstext= delresponse.text delresponsJson = json.loads(deldstext) if delresponsJson['msg'] == 'success': metatask_info = await cls.get_task_biz_config_detail_services(query_db, process.metaTaskId) metatask_info2 = metatask_info.model_dump(exclude_unset=True) metatask_info2['schId']="" await MetadataConfigDao.edit_task_biz_config(query_db,metatask_info.onum, metatask_info2) await query_db.commit() return "调度删除成功！" return "调度删除成功！" @classmethod async def get_task_biz_config_detail_services(cls, db: AsyncSession, onum: str): result = await MetadataConfigDao.get_task_biz_config_detail_by_id(db, onum) if result: return TaskBizConfigModel(**CamelCaseUtil.transform_result(result)) else: return TaskBizConfigModel(**dict()) @classmethod async def get_task_biz_config_rela_list_services(cls, result_db: AsyncSession, biz_onum: int): ai_session_list = await MetadataConfigDao.get_task_rela_by_biz_id(result_db, biz_onum) # 查询最新的20条 return CamelCaseUtil.transform_result(ai_session_list) # -------------------------------标签分类分级---------------------------- @classmethod async def get_catalog_list_services( cls, query_db: AsyncSession, query_object: DataCatalogPageQueryModel, user_id: int, is_page: bool = False ): """ 获取数据目录列表信息service :param query_db: orm对象 :param query_object: 查询参数对象 :param is_page: 是否开启分页 :return: 数据目录列表信息对象 """ catalog_list_result = await MetadataConfigDao.get_catalog_list(query_db, query_object, user_id, is_page) # 按contentOnum分组 grouped = defaultdict(list) for item in catalog_list_result.rows: grouped[item['contentOnum']].append(item) nodes = {} # 存储所有处理后的节点 # 处理每个组，生成节点 for belt_data_std_content, items in grouped.items(): first_item = items[0] is_leaf = first_item['leafNodeFlag'] == 1 rela_onum = first_item['relaOnum'] is not None # 公共字段提取 common_fields = { 'contentOnum': first_item['contentOnum'], 'contentName': first_item['contentName'], 'contentStat': first_item['contentStat'], 'contentIntr': first_item['contentIntr'], 'contentPic': first_item['contentPic'], 'suprContentOnum': first_item['suprContentOnum'], 'leafNodeFlag': first_item['leafNodeFlag'], 'updPrsn': first_item['updPrsn'], 'updTime': first_item['updTime'], 'children': [] } nodes[belt_data_std_content] = common_fields # 构建父子关系 root = None for belt_data_std_content, node in nodes.items(): supr = node['suprContentOnum'] if supr is None: root = node else: parent = nodes.get(supr) if parent: parent['children'].append(node) # 对每个父节点的children进行排序，将'临时的节点'放到最后 for belt_data_std_content, node in nodes.items(): if 'children' in node: # 排序时，'临时的节点'会被放到最后 node['children'] = sorted( node['children'], key=lambda x: x['contentName'] == '临时', reverse=False # True会将'临时的节点'排在最前面，False是排在最后 ) print('获取数据清单内容：',root) catalog_list_result.rows = [root] return catalog_list_result @classmethod async def delete_catalog_services(cls, query_db: AsyncSession, request: DeleteDataCatalogModel,user_id): """ 删除数据目录信息service :param query_db: orm对象 :param request: 删除数据目录请求对象 :return: 删除目录操作结果 """ if request.content_onums: content_onum_list = request.content_onums.split(',') try: for belt_data_std_content in content_onum_list: catalog = await cls.get_catalog_detail_services(query_db, int(belt_data_std_content)) if not catalog: raise ServiceException(message=f'目录ID {belt_data_std_content} 不存在') await MetadataConfigDao.delete_catalog_dao(query_db, DeleteDataCatalogModel(content_onums=belt_data_std_content)) # await MetadataConfigDao.delete_ast_book_mark_rela_by_content_onum(query_db, int(belt_data_std_content), user_id) await query_db.commit() return CrudResponseModel(is_success=True, message='删除成功') except Exception as e: await query_db.rollback() raise e else: raise ServiceException(message='传入目录id为空') @classmethod async def removerel_data_ast_catalog_services(cls, query_db: AsyncSession, request: DataCatalogChild): """ 移除数据资产目录service """ removerel_catalog_data = { 'rela_onum': request.rela_onum, 'content_onum': request.content_onum, 'ast_onum': request.ast_onum, 'rela_type': request.rela_type, 'rela_eff_begn_date': request.rela_eff_begn_date, 'rela_eff_end_date': request.rela_eff_end_date, 'upd_prsn': request.upd_prsn, 'rela_status': request.rela_status } try: await MetadataConfigDao.removerel_data_ast_catalog_dao(query_db, removerel_catalog_data) await query_db.commit() return CrudResponseModel(is_success=True, message='资产移除成功') except Exception as e: await query_db.rollback() raise ServiceException(message=f"移除资产时发生错误: {str(e)}") @classmethod async def get_catalog_detail_services(cls, query_db: AsyncSession, belt_data_std_content: int): """ 获取数据目录详细信息service :param query_db: orm对象 :param belt_data_std_content: 数据目录ID :return: 数据目录详细信息对象 """ catalog_detail_result = await MetadataConfigDao.get_catalog_by_id(query_db, belt_data_std_content) return catalog_detail_result @classmethod async def moved_catalog_instr_services(cls, query_db: AsyncSession, request: DataCatalogMovedRequest): """ 移动数据目录service """ moved_catalog_data = { 'content_onum': request.content_onum, 'supr_content_onum': request.supr_content_onum, 'supr_content_onum_after': request.supr_content_onum_after } try: await MetadataConfigDao.moved_catalog_instr_dao(query_db, moved_catalog_data) await query_db.commit() return CrudResponseModel(is_success=True, message='目录移动成功') except Exception as e: await query_db.rollback() raise ServiceException(message=f"移动目录时发生错误: {str(e)}") @classmethod async def merge_catalog_instr_services(cls, query_db: AsyncSession, request: DataCatalogMergeRequest): """ 移动数据目录service """ merge_catalog_data = { 'content_onum': request.content_onum, 'supr_content_onum': request.supr_content_onum, 'content_onum_after': request.content_onum_after, 'supr_content_onum_after': request.supr_content_onum_after } try: await MetadataConfigDao.merge_catalog_instr_dao(query_db, merge_catalog_data) await query_db.commit() return CrudResponseModel(is_success=True, message='目录合并成功') except Exception as e: await query_db.rollback() raise ServiceException(message=f"目录合并时发生错误: {str(e)}") @classmethod async def edit_catalog_child_services(cls, query_db: AsyncSession, request: DataCatalogResponseWithChildren): """ 编辑数据目录信息service :param query_db: orm对象 :param request: 编辑数据目录请求对象 :return: 编辑目录操作结果 """ catalog_data = { 'content_onum': request.content_onum, 'content_name': request.content_name, 'content_stat': request.content_stat, 'content_intr': request.content_intr, 'content_pic': request.content_pic, 'supr_content_onum': request.supr_content_onum, 'leaf_node_flag': request.leaf_node_flag, 'upd_prsn': request.upd_prsn, 'children': [child.model_dump() for child in request.children] # 将 children 转换为字典列表 } try: for child in catalog_data["children"]: # 设置 rela_eff_begn_date if child.get("rela_eff_begn_date"): child["rela_eff_begn_date"] = child["rela_eff_begn_date"].strftime("%Y-%m-%d %H:%M:%S") else: child["rela_eff_begn_date"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # 设置 rela_eff_end_date if child.get("rela_eff_end_date"): child["rela_eff_end_date"] = child["rela_eff_end_date"].strftime("%Y-%m-%d %H:%M:%S") else: child["rela_eff_end_date"] = datetime(year=2999, month=12, day=31, hour=0, minute=0, second=0).strftime("%Y-%m-%d %H:%M:%S") child["upd_prsn"] = request.upd_prsn child["rela_status"] = "1" await MetadataConfigDao.edit_catalog_child_dao(query_db, catalog_data) await query_db.commit() return CrudResponseModel(is_success=True, message='更新成功') except Exception as e: await query_db.rollback() raise ServiceException(message=f"更新目录时发生错误: {str(e)}") @classmethod async def add_catalog_services(cls, query_db: AsyncSession, request: DataCatalogResponseWithChildren): """ 新增数据目录信息service :param query_db: orm对象 :param request: 新增数据目录请求对象 :return: 新增目录操作结果 """ catalog_data1 = { 'content_name': request.content_name, 'content_stat': request.content_stat, 'content_intr': request.content_intr, 'content_pic': request.content_pic, 'supr_content_onum': request.supr_content_onum, 'leaf_node_flag': request.leaf_node_flag, 'upd_prsn': request.upd_prsn } catalog_data2 = { 'content_name': request.content_name, 'content_stat': request.content_stat, 'content_intr': request.content_intr, 'content_pic': request.content_pic, 'supr_content_onum': request.supr_content_onum, 'leaf_node_flag': request.leaf_node_flag, 'upd_prsn': request.upd_prsn, 'children': [child.model_dump() for child in request.children] # 将 children 转换为字典列表 } try: for child in catalog_data2["children"]: child["rela_eff_begn_date"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S"), # 设置默认值，当前时间 child["rela_eff_end_date"] = datetime(year=2999, month=12, day=31, hour=0, minute=0, second=0).strftime("%Y-%m-%d %H:%M:%S"), # 设置默认值，2999-12-31 child["upd_prsn"] = request.upd_prsn, child["rela_status"] = "1" new_catalog = await MetadataConfigDao.add_catalog_dao(query_db, catalog_data1, catalog_data2) await query_db.commit() return CrudResponseModel(is_success=True, message='新增成功', data=new_catalog) except Exception as e: await query_db.rollback() raise ServiceException(message=f"创建目录时发生错误: {str(e)}") @classmethod async def edit_catalog_leaf_services(cls, query_db: AsyncSession,content_onum : int, leaf_node_flag : int): """ 编辑数据目录信息service :param query_db: orm对象 :param request: 编辑数据目录请求对象 :return: 编辑目录操作结果 """ catalog_data1 = { 'content_onum': content_onum, 'leaf_node_flag': leaf_node_flag } try: await MetadataConfigDao.edit_catalog_leaf_dao(query_db, catalog_data1) await query_db.commit() return CrudResponseModel(is_success=True, message='更新成功') except Exception as e: await query_db.rollback() raise ServiceException(message=f"更新目录时发生错误: {str(e)}")