package com.ruoshui.flink.streaming.web.common.util;


import com.ruoshui.flink.streaming.web.common.FlinkConstants;
import com.ruoshui.flink.streaming.web.common.RestResult;
import com.ruoshui.flink.streaming.web.common.SystemConstants;
import com.ruoshui.flink.streaming.web.enums.CheckPointParameterEnums;
import com.ruoshui.flink.streaming.web.exceptions.BizException;
import com.ruoshui.common.flink.enums.StateBackendEnum;
import com.ruoshui.common.flink.model.CheckPointParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.cli.*;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * @author xinjingruoshui
 * @Description:
 * @date 2022-09-11
 * @time 00:00
 */
@Slf4j
public class CliConfigUtil {

    /**
     * 检查flink运行启动参数
     *
     * @author xinjingruoshui
     * @date 2022-09-11
     * @time 00:04
     */
    public static RestResult checkFlinkRunConfigForYarn(String flinkRunConfig) {
        try {
            CommandLine cl = getFlinkRunByCli(flinkRunConfig);
            if (!cl.hasOption(SystemConstants.YQU)) {
                return RestResult.error("任务必须包含队列参数 -yqu ");
            }
        } catch (UnrecognizedOptionException e) {
            log.error("checkFlinkRunConfig is error", e);
            return RestResult.error("flink运行配置参数校验通不过,不允许使用参数：" + e.getOption() + " 参数只支持 -p -yjm -yn -ytm -ys -yqu -yD");
        } catch (Exception e) {
            log.error("checkFlinkRunConfig is error", e);
            return RestResult.error("flink运行配置参数校验通不过");
        }
        return null;
    }


    /**
     * 检查CheckPoint参数
     *
     * @author xinjingruoshui
     * @date 2022-09-11
     * @time 00:04
     */
    public static CheckPointParam checkFlinkCheckPoint(String flinkCheckpointConfig) {
        try {
            String[] config = trim(flinkCheckpointConfig);

            ParameterTool parameterTool = ParameterTool.fromArgs(config);
            if (parameterTool == null || parameterTool.getUnrequestedParameters() == null) {
                throw new BizException("parameterTool or parameterTool.getUnrequestedParameters() is null ");
            }

            CheckPointParameterEnums.isExits(parameterTool.getUnrequestedParameters());

            String checkpointDir = parameterTool.get(CheckPointParameterEnums.checkpointDir.name());

            String checkpointingMode = parameterTool.get(CheckPointParameterEnums.checkpointingMode.name(),
                    FlinkConstants.EXACTLY_ONCE);
            String tolerableCheckpointFailureNumber = parameterTool.get(CheckPointParameterEnums.tolerableCheckpointFailureNumber.name());
            String asynchronousSnapshots = parameterTool.get(CheckPointParameterEnums.asynchronousSnapshots.name());
            String checkpointInterval = parameterTool.get(CheckPointParameterEnums.checkpointInterval.name());
            String checkpointTimeout = parameterTool.get(CheckPointParameterEnums.checkpointTimeout.name());
            String externalizedCheckpointCleanup = parameterTool.get(CheckPointParameterEnums.externalizedCheckpointCleanup.name());
            String stateBackendType = parameterTool.get(CheckPointParameterEnums.stateBackendType.name());
            String enableIncremental = parameterTool.get(CheckPointParameterEnums.enableIncremental.name());

            CheckPointParam checkPointParam = new CheckPointParam();
            if (StringUtils.isNotEmpty(asynchronousSnapshots)) {
                if (Boolean.FALSE.toString().equals(asynchronousSnapshots.toLowerCase())
                        || Boolean.TRUE.toString().equals(asynchronousSnapshots.toLowerCase())) {
                    checkPointParam.setAsynchronousSnapshots(Boolean.valueOf(asynchronousSnapshots));
                } else {
                    throw new BizException("asynchronousSnapshots 参数必须是 Boolean 类型或者为空 ");
                }
            }
            if (StringUtils.isNotEmpty(checkpointTimeout)) {
                checkPointParam.setCheckpointTimeout(Long.parseLong(checkpointTimeout));
            }

            checkPointParam.setCheckpointingMode(checkpointingMode);
            if (StringUtils.isNotEmpty(checkpointInterval)) {
                checkPointParam.setCheckpointInterval(Long.parseLong(checkpointInterval));
            }

            if (StringUtils.isNotEmpty(tolerableCheckpointFailureNumber)) {
                checkPointParam.setTolerableCheckpointFailureNumber(Integer.parseInt(tolerableCheckpointFailureNumber));
            }
            if (StringUtils.isNotEmpty(externalizedCheckpointCleanup)) {
                checkPointParam.setExternalizedCheckpointCleanup(externalizedCheckpointCleanup);
            }

            //内存模式下不需要填写checkpointDir
            if (!StateBackendEnum.MEMORY.getType().equalsIgnoreCase(stateBackendType)
                    && StringUtils.isEmpty(checkpointDir)) {
                throw new BizException("checkpointDir不存在或者没有对应的值");
            }
            checkPointParam.setCheckpointDir(checkpointDir);

            checkPointParam.setStateBackendEnum(StateBackendEnum.getStateBackend(stateBackendType));

            if (StringUtils.isNotEmpty(enableIncremental)) {
                if (Boolean.FALSE.toString().equals(enableIncremental.toLowerCase())
                        || Boolean.TRUE.toString().equals(enableIncremental.toLowerCase())) {
                    checkPointParam.setEnableIncremental(Boolean.getBoolean(enableIncremental.trim()));
                } else {
                    throw new BizException("enableIncremental 参数必须是 Boolean 类型或者为空 ");
                }
            }

            log.info("checkPointParam ={}", checkPointParam);

            return checkPointParam;
        } catch (BizException e) {
            log.error("checkFlinkCheckPoint is error", e);
            throw e;
        } catch (Exception e) {
            log.error("checkFlinkCheckPoint is error", e);
            throw new BizException("Checkpoint参数校验不通过:" + e.getMessage());
        }
    }


    public static CommandLine getFlinkRunByCli(String flinkRunConfig) throws ParseException {
        String[] config = trim(flinkRunConfig);
        Options options = new Options();
        options.addOption("p", false, "");
        options.addOption("yjm", false, "");
        options.addOption("yn", false, "");
        options.addOption("ytm", false, "");
        options.addOption("ys", false, "");
        options.addOption("yD", false, "");
        options.addOption(SystemConstants.YQU, true, "");
        CommandLineParser parser = new DefaultParser();
        return parser.parse(options, config);
    }


    private static String[] trim(String cliConfig) {

        List<String> list = new ArrayList<>();
        String[] config = cliConfig.split(" ");
        for (String str : config) {
            if (StringUtils.isNotEmpty(str)) {
                list.add(str);
            }
        }
        return list.toArray(new String[list.size()]);
    }

}
