import json import uuid from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession from module_admin.entity.vo.common_vo import CrudResponseModel from module_admin.entity.vo.approval_vo import ApplyModel, OperateModel, ApprovalQueryObject, EditObjectModel, SaveConfModel from module_admin.entity.vo.user_vo import CurrentUserModel from module_admin.entity.do.approval_do import FlowApproval from exceptions.exception import ServiceException, ServiceWarning from datetime import datetime from utils.common_util import CamelCaseUtil from module_admin.dao.approval_dao import ApprovalDao from module_admin.dao.user_dao import UserDao from module_admin.dao.meta_dao import MetaDao from module_admin.dao.datastd_dao import DataStdDao from module_admin.entity.vo.datastd_vo import DataStdCodeModel,DataStdDictModel,DataStdMainModel,DataStdMainApprModel class ApprovalService: """ 智能问答服务层 """ @classmethod async def apply_services(cls, result_db: AsyncSession, apply: ApplyModel, module: str): flow_approval = FlowApproval() flow_approval.id = uuid.uuid4() flow_approval.businessType = apply.businessType flow_approval.businessId = apply.businessId flow_approval.applicant = apply.applicant flow_approval.applyTime = datetime.now().strftime("%Y-%m-%d %H:%M:%S") flow_approval.status = 'waiting' listConf = await ApprovalDao.get_first_node_conf(module, result_db) flow_approval.nextStep = json.dumps([item.id for item in listConf]) flow_approval.nextStepRole = json.dumps([item.code for item in listConf if item.type == 'Role']) flow_approval.nextStepUser = json.dumps([item.code for item in listConf if item.type == 'User']) await ApprovalDao.add_flow_approval(result_db, flow_approval) await result_db.commit() return CrudResponseModel(is_success=True, message='申请成功') @classmethod async def operate_services(cls, result_db: AsyncSession, operate: OperateModel, current_user: CurrentUserModel): flow_approval = await ApprovalDao.get_flow_by_id(result_db, operate.flowId) if flow_approval is None: raise ServiceException(message=f'所操作的流程不存在') if flow_approval.status == 'succeed' or flow_approval.status == 'rejected': raise ServiceException(message='所操作的流程已结束') if flow_approval.status == 'canceled': raise ServiceException(message='所操作的流程已撤回') edit = EditObjectModel() edit.id = flow_approval.id flowConfList = await ApprovalDao.get_conf_list(result_db, flow_approval.businessType) nextSteps = json.loads(flow_approval.nextStep) nextStepFlowList = [item for item in flowConfList if item.id in nextSteps] nextFlowList = [item for item in nextStepFlowList if item.code in current_user.roles] confNodeId = nextFlowList[0].id array = [] if flow_approval.approvalFlow is not None: array = json.loads(flow_approval.approvalFlow) if operate.operateType == 'success': nextFlow = [] for flowConf in flowConfList: parent = json.loads(flowConf.parent) if confNodeId in parent: nextFlow.append(flowConf) if len(nextFlow) > 0: edit.status = 'pending' # 有下一步执行人，则设置为 pending edit.nextStep = json.dumps([item.id for item in nextFlow]) edit.nextStepRole = json.dumps([item.code for item in nextFlow if item.type == 'Role']) edit.nextStepUser = json.dumps([item.code for item in nextFlow if item.type == 'User']) edit.currentNodeId = confNodeId else: edit.status = 'succeed' edit.currentNodeId = confNodeId edit.nextStep = '[]' edit.nextStepRole = '[]' edit.nextStepUser = '[]' if operate.operateType == 'reject': edit.status = 'rejected' edit.currentNodeId = confNodeId edit.nextStep = '[]' edit.nextStepRole = '[]' edit.nextStepUser = '[]' array.append({'operator': current_user.user.user_name, 'confFlowId': confNodeId, 'operateTime': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'operate': operate.operateType, 'operateComment': operate.operateComment, }) edit.approvalFlow = json.dumps(array) if flow_approval.businessType == 'metaDataInfo': await cls.syncSuppInfo(result_db, flow_approval.businessId, edit.status) # 数据标准审批 elif flow_approval.businessType == 'dataStdMain': await cls.syncStdMainInfo(result_db, flow_approval.businessId, edit.status) await ApprovalDao.edit_flow_approval(result_db, edit.model_dump(exclude_unset=True)) await result_db.commit() return CrudResponseModel(is_success=True, message='操作成功') @classmethod async def syncSuppInfo(cls, result_db: AsyncSession, suppId: str, operateType: str): table = await MetaDao.get_supp_table_vett_by_id(suppId, result_db) if table is None: raise ServiceException(message='所查询的业务数据不存在') if table.apply_status == 'succeed' or table.apply_status == 'rejected': raise ServiceException(message='所改业务已审核完毕') if table.apply_status == 'canceled': raise ServiceException(message='所改业务已撤销申请') column_list = await MetaDao.get_meta_col_supp_vett(table, result_db) if column_list is not None and len(column_list) > 0: for column in column_list: column.apply_status = operateType suppColumn = await MetaDao.get_supp_column_by_vett(column, result_db) await MetaDao.updateMetadataFldSuppInfoVett(column.onum, operateType, result_db) if operateType != 'canceled': if suppColumn is None: await MetaDao.insertMetadataFldSuppInfo(column, result_db) else: await MetaDao.updateMetadataFldSuppInfo(suppColumn.onum, column, result_db) suppTable = await MetaDao.get_supp_table_by_vett(table.ssys_cd, table.mdl_name, table.tab_eng_name, result_db) await MetaDao.updateMetadataSuppInfoVett(table.onum, operateType, result_db) if operateType != 'canceled': if suppTable is None: await MetaDao.insertMetadataSuppInfo(table, result_db) else: await MetaDao.updateMetadataSuppInfo(suppTable.onum, table, result_db) return CrudResponseModel(is_success=True, message='操作成功') @classmethod async def syncStdMainInfo(cls, result_db: AsyncSession, suppId: str, operateType: str): apprList = await DataStdDao.get_std_main_appr_list(suppId, result_db) for appr in apprList: # 将 SQLAlchemy 模型实例转换为 Pydantic 模型实例 appr_model = DataStdMainApprModel.model_validate(appr) change_type = appr_model.changeType if operateType =='succeed': if change_type == "add": # 将 DataStdMainApprModel 转换为 DataStdMainModel main_model = DataStdMainModel(**appr_model.model_dump(exclude_unset=True, by_alias=True)) await DataStdDao.add_std_main(result_db, main_model) elif change_type == "edit": # 使用 oldInstId 作为主键 main_model = DataStdMainModel(**appr_model.model_dump(exclude_unset=True, by_alias=True)) main_model.id = appr_model.oldInstId await DataStdDao.update_std_main(result_db, main_model) elif change_type == "delete": await DataStdDao.delete_std_main(result_db, appr_model.oldInstId) # 更新 approStatus 状态 appr_model.approStatus = operateType await DataStdDao.update_std_main_appr(result_db, appr_model) return CrudResponseModel(is_success=True, message='操作成功') @classmethod async def get_flow_list_services(cls, query_db: AsyncSession, query_param: ApprovalQueryObject, current_user: CurrentUserModel): if query_param.status == 'waiting': result = await ApprovalDao.get_flow_list(query_db, query_param, current_user) return result if query_param.status == 'over': result = await ApprovalDao.get_owner_flow_list(query_db, query_param, current_user) return result if query_param.status == 'applying': result = await ApprovalDao.get_my_flow_list(query_db, query_param, current_user) return result return None @classmethod async def get_waiting_total_services(cls, query_db: AsyncSession, current_user: CurrentUserModel): approval_list = await ApprovalDao.get_all_waitingOrPendingFlows(query_db) count = 0 for item in approval_list: nextRoles = json.loads(item.nextStepRole) nextUsers = json.loads(item.nextStepUser) if len(nextRoles)>0 and len(nextUsers) > 0: if current_user.user.user_name in nextUsers: count = count+1 break if any(element in set(current_user.roles) for element in nextRoles): count = count+1 break if len(nextRoles) > 0 and len(nextUsers) == 0: if any(element in set(current_user.roles) for element in nextRoles): count = count+1 break if len(nextRoles) == 0 and len(nextUsers) > 0: if current_user.user.user_name in nextUsers: count = count+1 break return count @classmethod async def cancel_apply_services(cls, query_db: AsyncSession, flow_id: str, current_user: CurrentUserModel): flow = await ApprovalDao.get_flow_by_idAndUser(flow_id, current_user.user.user_name) if flow is None: raise ServiceException(message=f'所操作的流程不存在') if flow.status == 'succeed' or flow.status == 'rejected': raise ServiceException(message='所操作的流程已结束，无需撤回') if flow.status == 'canceled': raise ServiceException(message='所操作的流程已撤回,无需重复操作') flow.status = 'canceled' edit = EditObjectModel() edit.id = flow.id edit.status = 'canceled' edit.nextStep = '[]' edit.nextStepRole = '[]' edit.nextStepUser = '[]' await ApprovalDao.edit_flow_approval(query_db, edit.model_dump(exclude_unset=True)) if flow.businessType == 'metaDataInfo': await cls.syncSuppInfo(query_db, flow.businessId, edit.status) await query_db.commit() return CrudResponseModel(is_success=True, message='操作成功') @classmethod async def get_conf_list_services(cls, module: str, query_db: AsyncSession, current_user: CurrentUserModel): result = await ApprovalDao.get_conf_list(query_db, module) return result @classmethod async def save_flow_config_services(cls, query_db: AsyncSession, saveConfModel: SaveConfModel): await ApprovalDao.delete_flow_by_module(saveConfModel.module, query_db) for item in saveConfModel.confList: confDict = item.model_dump(exclude_unset=True) await ApprovalDao.add_flow_config(query_db, confDict) await query_db.commit() return CrudResponseModel(is_success=True, message='操作成功')