virtual-patient/virtual-patient-rasa/src/main/java/com/supervision/rasa/service/RasaModelManager.java

package com.supervision.rasa.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.StrUtil;
import com.supervision.model.RasaModelInfo;
import com.supervision.rasa.constant.RasaConstant;
import com.supervision.rasa.pojo.dto.RasaRunParam;
import com.supervision.rasa.util.PortUtil;
import com.supervision.service.RasaModeService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.FileFilter;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

@Slf4j
@Component
@RequiredArgsConstructor
public class RasaModelManager {

    @Value("${rasa.models-path}")
    private String modelsPath;

    private final RasaModeService rasaModeService;

    private final RasaCmdService rasaCmdService;

    private  boolean wakeUpInterruptServerRunning = false;

    public void wakeUpInterruptServer(){

        // 1. 查找出记录表中存活的服务
        List<RasaModelInfo> rasaModelInfos = rasaModeService.listActive();
        List<RasaModelInfo> activeRasaList = rasaModelInfos.stream().filter(info -> CollectionUtil.isNotEmpty(info.getRunCmd()) && null != info.getPort()).collect(Collectors.toList());

        if (CollectionUtil.isEmpty(activeRasaList)){
            log.info("wakeUpInterruptService: no rasa service need wake up ...");
            return;
        }

        // 2. 重新启动中断的服务
          for (RasaModelInfo rasaModelInfo : activeRasaList) {
            if (!PortUtil.portIsActive(rasaModelInfo.getPort())){
                try {
                    RasaRunParam rasaRunParam = RasaRunParam.build(rasaModelInfo.getRunCmd());
                    rasaRunParam.setPort(String.valueOf(rasaModelInfo.getPort()));
                    rasaRunParam.setShellPath(rasaCmdService.getShellPath(RasaConstant.RUN_SHELL));
                    String rasaModelPath = rasaRunParam.getRasaModelPath();
                    if (StrUtil.isEmpty(rasaModelPath) || !FileUtil.exist(rasaModelPath)){
                        log.info("wakeUpInterruptServer: rasa model path {} not exist,attempt find last ...",rasaModelPath);
                        String modeParentPath = replaceDuplicateSeparator(String.join(File.separator, modelsPath, rasaModelInfo.getModelId()));
                        String fixedModePath = listLastFilePath(modeParentPath, f -> f.getName().matches("-?\\d+(\\.\\d+)?.tar.gz"));
                        Assert.notEmpty(fixedModePath,"wakeUpInterruptService: no rasa model in path {} ",modeParentPath);
                        rasaRunParam.setRasaModelPath(fixedModePath);
                    }
                    log.info("wakeUpInterruptServer : use fixedModePath :{}",rasaRunParam.getRasaModelPath());
                    List<String> outMessageList = rasaCmdService.execCmd(rasaRunParam.toList(),
                            s -> StrUtil.isNotBlank(s) && s.contains(RasaConstant.RUN_SUCCESS_MESSAGE), 300);

                    rasaModelInfo.setRunLog(String.join("\r\n",outMessageList));
                    rasaModelInfo.setRunCmd(rasaRunParam.toList());
                    rasaModeService.updateById(rasaModelInfo);

                    if (!runIsSuccess(outMessageList)){
                        log.info("wakeUpInterruptServer: restart server port for {} failed,details info : {}",rasaModelInfo.getPort(),String.join("\r\n",outMessageList));
                    }
                } catch (InterruptedException | ExecutionException | TimeoutException e ) {
                    log.info("wakeUpInterruptServer: restart server port for {} failed",rasaModelInfo.getPort());
                    throw new RuntimeException(e);
                }
                log.info("wakeUpInterruptServer: restart server port for {} success ",rasaModelInfo.getPort());
            }else {
                log.info("wakeUpInterruptServer: port:{} is run..",rasaModelInfo.getPort());
            }
        }
    }



    //默认每十分钟执行一次
    @Scheduled(cron = "${rasa.wakeup.cron:0 */10 * * * ?}")
    public void wakeUpInterruptServerScheduled(){
        try {
            log.info("wakeUpInterruptServerScheduled: Scheduled is run .... wakeUpInterruptServerRunning is :{}",wakeUpInterruptServerRunning);
            if (!wakeUpInterruptServerRunning){
                wakeUpInterruptServerRunning = true;
                wakeUpInterruptServer();

            }
        }  catch (Exception e){
            log.error("wakeUpInterruptServerScheduled: Scheduled is run failed....",e);
            wakeUpInterruptServerRunning = false;
        }
    }

    private  boolean runIsSuccess(List<String> messageList){

        return containKey(messageList,RasaConstant.RUN_SUCCESS_MESSAGE);
    }

    private boolean containKey(List<String> messageList,String keyWord){

        if (CollectionUtil.isEmpty(messageList)){
            return false;
        }
        if (StrUtil.isEmpty(keyWord)){
            return false;
        }
        return messageList.stream().anyMatch(s->StrUtil.isNotEmpty(s) && s.contains(keyWord));
    }

    private String replaceDuplicateSeparator(String path){

        if (StrUtil.isEmpty(path)){
            return path;
        }

        return path.replace(File.separator + File.separator, File.separator);
    }

    private String listLastFilePath(String path, FileFilter filter){
        File file = listLastFile(path, filter);
        if (null == file){
            return null;
        }
        return file.getPath();
    }

    private File listLastFile(String path,FileFilter filter){
        File file = new File(path);
        File[] files = file.listFiles(filter);
        if (null == files){
            return null;
        }

        return Arrays.stream(files).max(Comparator.comparing(File::getName)).orElse(null);
    }
}