package com.purui.prims.dscheduler.service;

import com.alibaba.fastjson.JSON;
import com.purui.prims.common.config.DolphinSchedulerConfig;
import com.purui.prims.common.util.DateUtils;
import com.purui.prims.dscheduler.model.ProcessDefinition;
import com.purui.prims.dscheduler.model.TaskDefinition;
import com.purui.prims.dscheduler.model.TaskDefinitionLog;
import com.purui.prims.dscheduler.model.TaskDefinitionVO;
import com.purui.prims.dscheduler.model.TaskInstance;
import com.purui.prims.dscheduler.model.TaskLog;
import com.purui.prims.dscheduler.model.TaskMainInfo;
import com.purui.prims.dscheduler.result.PageInfo;
import com.purui.prims.dscheduler.utils.ParamUtil;
import com.purui.prims.headless.api.pojo.enums.DolphineSchedulerType;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 任务服务
 *
 * @author charlie
 * @date 2024/7/10
 */
@Service
@Slf4j
public class TaskService extends AbstractSchedulerService {
    public TaskService(DolphinSchedulerConfig config, RestTemplate restTemplate) {
        super(config, restTemplate);
    }

    /**
     * 创建任务定义
     *
     * @param projectCode 项目编号
     * @param processCode 工作流定义编号
     * @return {@link TaskDefinitionLog}
     */
    public TaskDefinitionLog createTaskDefinition(
        Long projectCode, Long processCode, List<String> upstreamCodes, String taskDefinitionJsonObj) {
        Map<String, Object> params = new HashMap<>();
        params.put("processDefinitionCode", processCode);
        if (CollectionUtils.isNotEmpty(upstreamCodes)) {
            params.put("upstreamCodes", StringUtils.join(upstreamCodes, ","));
        }

        params.put("taskDefinitionJsonObj", taskDefinitionJsonObj);

        return runSchedulerX(projectCode, 0L, DolphineSchedulerType.TASK_CREATE, params, TaskDefinitionLog.class);
    }

    /**
     * 查询任务定义集合
     *
     * @param projectCode 项目编号
     * @param processName 工作流定义名称
     * @param taskName    任务定义名称
     * @return {@link List<TaskMainInfo>}
     */
    public List<TaskMainInfo> getTaskMainInfos(Long projectCode, String processName, String taskName, String taskType) {
        Map<String, Object> pageParams = ParamUtil.getPageParams();
        pageParams.put("searchTaskName", taskName);
        pageParams.put("searchWorkflowName", processName);
        pageParams.put("taskType", taskType);

        PageInfo data = runSchedulerX(projectCode, 0L, DolphineSchedulerType.TASK_LIST, pageParams, PageInfo.class);

        List<TaskMainInfo> lists = new ArrayList<>();
        if (data == null || data.getTotalList() == null) {
            return lists;
        }
        lists = JSON.parseArray(JSON.toJSONString(data.getTotalList()), TaskMainInfo.class);
        return lists;
    }

    /**
     * 根据编号查询
     *
     * @param projectCode 项目编号
     * @param taskCode    任务编号
     * @return {@link TaskDefinition}
     */
    public TaskDefinitionVO getTaskDefinition(Long projectCode, Long taskCode) {
        return runSchedulerX(projectCode, taskCode, DolphineSchedulerType.TASK_DETAIL, null, TaskDefinitionVO.class);

    }

    // 查询任务代码
    public Long getTaskCode(Long projectCode, String taskName) {
        List<TaskMainInfo> lists = getTaskMainInfos(projectCode, "", taskName, "");
        if (CollectionUtils.isNotEmpty(lists)) {
            return lists.get(0).getTaskCode();
        }
        return 0L;
    }

    /**
     * 根据任务名称查询任务实例
     *
     * @param projectCode
     * @return
     */
    public List<TaskInstance> getTaskInstances(Long projectCode, String startDate, String endDate, String key) {
        Map<String, Object> pageParams = ParamUtil.getPageParams(key, 1, 20);
        pageParams.put("taskCode", "");
        pageParams.put("startDate", startDate);
        pageParams.put("endDate", endDate);
        pageParams.put("processInstanceId", "");
        pageParams.put("processInstanceName", "");
        pageParams.put("host", "");
        pageParams.put("stateType", "");
        pageParams.put("executorName", "");

        PageInfo data = runSchedulerX(projectCode, 0L, DolphineSchedulerType.TASK_INSTANCE_LIST,
            pageParams, PageInfo.class);
        if (data != null && data.getTotalList() != null) {
            return JSON.parseArray(JSON.toJSONString(data.getTotalList()), TaskInstance.class);
        }
        return null;
    }

    /**
     * 查询指标任务执行日志
     *
     * @param projectCode
     * @param instanceId
     * @return
     */
    public TaskLog queryTaskLog(Long projectCode, int instanceId) {
        if (instanceId > 0) {
            Map<String, Object> params = new HashMap<>(4);
            params.put("taskInstanceId", instanceId);
            params.put("skipLineNum", 0);
            params.put("limit", 2000);
            return runSchedulerX(projectCode, 0L, DolphineSchedulerType.TASK_LOG,
                params, TaskLog.class);
        }
        return null;
    }

    /**
     * 修改任务定义
     *
     * @param projectCode           项目编号
     * @param taskCode              任务定义编号
     * @param taskDefinitionJsonObj 修改参数
     * @return {@link Long}
     */
    public Long updateTaskDefinition(
        long projectCode, long taskCode, List<String> upstreamCodes, String taskDefinitionJsonObj) {
        Map<String, Object> params = new HashMap<>();
        if (CollectionUtils.isNotEmpty(upstreamCodes)) {
            params.put("upstreamCodes", StringUtils.join(upstreamCodes, ","));
        }
        params.put("taskDefinitionJsonObj", taskDefinitionJsonObj);

        return runSchedulerX(projectCode, taskCode, DolphineSchedulerType.TASK_UPDATE, params, Long.class);
    }

    /**
     * 生成任务定义编号
     *
     * @param projectCode 项目编号
     * @param genNum      生成个数
     * @return {@link List}
     */
    public List<Long> genTaskCodes(Long projectCode, int genNum) {
        Map<String, Object> params = new HashMap<>();
        params.put("genNum", genNum);
        return runSchedulerX(projectCode, 0L, DolphineSchedulerType.TASK_GEN_CODE, params, List.class);
    }

    /**
     * 运行任务
     *
     * @param projectCode
     * @param taskCode
     * @return
     */
    public boolean runTaskNow(Long projectCode, ProcessDefinition process, Long taskCode) {
        if (taskCode == null || taskCode.longValue() < 1L) {
            return false;
        }
        Map<String, Object> params = new HashMap<>();

        params.put("processDefinitionCode", process.getCode());

        params.put("version", process.getVersion());
        // 失败策略
        params.put("failureStrategy", "END");
        // 告警策略
        params.put("warningType", "NONE");

        params.put("execType", "START_PROCESS");

        params.put("startNodeList", taskCode);

        params.put("taskDependType", "TASK_POST");

        params.put("complementDependentMode", "OFF_MODE");

        params.put("runMode", "RUN_MODE_SERIAL");

        params.put("processInstancePriority", "MEDIUM");

        params.put("workerGroup", "default");

        params.put("tenantCode", "default");
        params.put("environmentCode", "");
        params.put("startParams", "");
        params.put("allLevelDependent", false);
        params.put("dryRun", 0);
        params.put("testFlag", 0);
        params.put("executionOrder", "DESC_ORDER");

        Map<String, Object> scheduleTime = new HashMap<>();
        scheduleTime.put("complementStartDate", DateUtils.format(new Date(), DateUtils.TIME_FORMAT));
        scheduleTime.put("endTime", DateUtils.format(new Date(), DateUtils.TIME_FORMAT));
        params.put("scheduleTime", JSON.toJSONString(scheduleTime));


        return runSchedulerX(projectCode, taskCode, DolphineSchedulerType.PROCESS_RUN_NOW, params, Boolean.class);
    }

}
