vue-fastapi-admin/vue-fastapi-backend/module_admin/service/approval_service.py

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.dao.dataast_dao import DataAstDao
from module_admin.entity.vo.dataast_vo import DataAstApprModel, DataAstInfoModel
from module_admin.entity.vo.datastd_vo import DataStdCodeModel, DataStdDictModel, DataStdMainModel, \
    DataStdMainApprModel, DataStdDictApprModel, DataStdDictModel, DataStdCodeApprModel


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)
            # 数据字典审批
        elif flow_approval.businessType == 'dataStdDict':
            await cls.syncStdDictInfo(result_db, flow_approval.businessId, edit.status)
        # 标准代码审批                  
        elif flow_approval.businessType == 'dataStdCode':
            await cls.syncStdCodeInfo(result_db, flow_approval.businessId, edit.status)
            # 数据资产审批
        elif flow_approval.businessType == 'dataAssetMain':
            await cls.syncDataAstInfo(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, businessId: str, operateType: str):
        tableList = await MetaDao.get_supp_table_vett_by_businessId(businessId, result_db)
        if tableList is None or len(tableList) == 0:
            raise ServiceException(message='所查询的业务数据不存在')
        if tableList[0].apply_status == 'succeed' or tableList[0].apply_status == 'rejected':
            raise ServiceException(message='所改业务已审核完毕')
        if tableList[0].apply_status == 'canceled':
            raise ServiceException(message='所改业务已撤销申请')
        for table in tableList:
            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_id, 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.onum = 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 syncStdDictInfo(cls, result_db: AsyncSession, suppId: str, operateType: str):
        apprList = await DataStdDao.get_std_dict_appr_list(suppId, result_db)

        for appr in apprList:
            # 将 SQLAlchemy 模型实例转换为 Pydantic 模型实例
            appr_model = DataStdDictApprModel.model_validate(appr)

            change_type = appr_model.changeType
            if operateType == 'succeed':
                if change_type == "add":
                    # 将 DataStdDictApprModel 转换为 DataStdDictModel
                    main_model = DataStdDictModel(**appr_model.model_dump(exclude_unset=True, by_alias=True))

                    await DataStdDao.add_std_dict(result_db, main_model)

                elif change_type == "edit":
                    # 使用 oldInstId 作为主键
                    main_model = DataStdDictModel(**appr_model.model_dump(exclude_unset=True, by_alias=True))

                    main_model.onum = appr_model.oldInstId
                    await DataStdDao.update_std_dict(result_db, main_model.model_dump(exclude_unset=True))

                elif change_type == "delete":
                    await DataStdDao.delete_std_dict(result_db, appr_model.oldInstId)

            # 更新 approStatus 状态
            appr_model.approStatus = operateType
            await DataStdDao.update_std_dict_appr(result_db, appr_model)

        return CrudResponseModel(is_success=True, message='操作成功')

    @classmethod
    async def syncStdCodeInfo(cls, result_db: AsyncSession, suppId: str, operateType: str):
        apprList = await DataStdDao.get_std_code_appr_list_Flow(suppId, result_db)

        for appr in apprList:
            # 将 SQLAlchemy 模型实例转换为 Pydantic 模型实例
            appr_model = DataStdCodeApprModel.model_validate(appr)

            change_type = appr_model.changeType
            if operateType == 'succeed':
                if change_type == "add":
                    main_model = DataStdCodeModel(**appr_model.model_dump(exclude_unset=True, by_alias=True))

                    await DataStdDao.add_std_code(result_db, main_model)

                elif change_type == "update":
                    # 使用 oldInstId 作为主键
                    main_model = DataStdCodeModel(**appr_model.model_dump(exclude_unset=True, by_alias=True))

                    main_model.onum = appr_model.oldInstId
                    await DataStdDao.update_std_code(result_db, main_model.model_dump(exclude_unset=True))

                elif change_type == "delete":
                    await DataStdDao.delete_std_code(result_db, appr_model.oldInstId)

            # 更新 approStatus 状态
            appr_model.approStatus = operateType
            main_appro_model = DataStdCodeApprModel(**appr_model.model_dump(exclude_unset=True, by_alias=True))

            await DataStdDao.update_std_code_appr(result_db, main_appro_model.model_dump(exclude_unset=True))

        return CrudResponseModel(is_success=True, message='操作成功')

    @classmethod
    async def syncDataAstInfo(cls, result_db: AsyncSession, suppId: str, operateType: str):
        apprList = await DataAstDao.get_dataast_appr_list_Flow(suppId, result_db)

        for appr in apprList:
            # 将 SQLAlchemy 模型实例转换为 Pydantic 模型实例
            appr_model = DataAstApprModel.model_validate(appr)
            change_type = appr_model.changeType
            if operateType == 'succeed':
                if change_type == "add":
                    # logger.info(f"发现新增类型变更，准备添加数据资产，AST_NO: {appr_model.ast_no}")
                    main_model = DataAstInfoModel(**appr_model.model_dump(exclude_unset=True, by_alias=True))
                    await DataAstDao.add_dataast_data(result_db, main_model)
                    # logger.info(f"数据资产添加成功，AST_NO: {main_model.ast_no}")

            # 更新 approStatus 状态
            appr_model.approStatus = operateType
            main_appro_model = DataStdCodeApprModel(**appr_model.model_dump(exclude_unset=True, by_alias=True))

            await DataAstDao.update_data_ast_appr(result_db, main_appro_model.model_dump(exclude_unset=True))

        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(query_db, 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='操作成功')