package com.ruoyi.platform.service.impl;

import com.ruoyi.platform.constant.Constant;
import com.ruoyi.platform.domain.Ray;
import com.ruoyi.platform.domain.RayIns;
import com.ruoyi.platform.mapper.RayDao;
import com.ruoyi.platform.mapper.RayInsDao;
import com.ruoyi.platform.service.RayInsService;
import com.ruoyi.platform.utils.DateUtils;
import com.ruoyi.platform.utils.HttpUtils;
import com.ruoyi.platform.utils.JsonUtils;
import com.ruoyi.platform.utils.MinioUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.PageRequest;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.stream.Collectors;

@Service("rayInsService")
public class RayInsServiceImpl implements RayInsService {
    @Value("${argo.url}")
    private String argoUrl;
    @Value("${argo.workflowStatus}")
    private String argoWorkflowStatus;
    @Value("${argo.workflowTermination}")
    private String argoWorkflowTermination;
    @Value("${minio.endpoint}")
    String endpoint;
    @Value("${minio.dataReleaseBucketName}")
    private String bucketName;

    @Resource
    private RayInsDao rayInsDao;

    @Resource
    private RayDao rayDao;

    @Resource
    private MinioUtil minioUtil;

    @Override
    public Page<RayIns> queryByPage(Long rayId, PageRequest pageRequest) throws IOException {
        long total = this.rayInsDao.count(rayId);
        List<RayIns> rayInsList = this.rayInsDao.queryAllByLimit(rayId, pageRequest);
        return new PageImpl<>(rayInsList, pageRequest, total);
    }

    @Override
    public RayIns insert(RayIns rayIns) {
        this.rayInsDao.insert(rayIns);
        return rayIns;
    }

    @Override
    public String deleteById(Long id) {
        RayIns rayIns = rayInsDao.queryById(id);
        if (rayIns == null) {
            return "实验实例不存在";
        }
        if (StringUtils.isEmpty(rayIns.getStatus())) {
            rayIns = queryStatusFromArgo(rayIns);
        }
        if (StringUtils.equals(rayIns.getStatus(), Constant.Running)) {
            return "实验实例正在运行，不可删除";
        }

        rayIns.setState(Constant.State_invalid);
        int update = rayInsDao.update(rayIns);
        if (update > 0) {
            updateRayStatus(rayIns.getRayId());
            return "删除成功";
        } else {
            return "删除失败";
        }
    }

    @Override
    public String batchDelete(List<Long> ids) {
        for (Long id : ids) {
            String result = deleteById(id);
            if (!"删除成功".equals(result)) {
                return result;
            }
        }
        return "删除成功";
    }

    @Override
    public boolean terminateRayIns(Long id) throws Exception {
        RayIns rayIns = rayInsDao.queryById(id);
        if (rayIns == null) {
            throw new IllegalStateException("实验实例未查询到，id: " + id);
        }

        String currentStatus = rayIns.getStatus();
        String name = rayIns.getArgoInsName();
        String namespace = rayIns.getArgoInsNs();

        // 获取当前状态，如果为空，则从Argo查询
        if (StringUtils.isEmpty(currentStatus)) {
            currentStatus = queryStatusFromArgo(rayIns).getStatus();
        }

        // 只有状态是"Running"时才能终止实例
        if (!currentStatus.equalsIgnoreCase(Constant.Running)) {
            throw new Exception("终止错误，只有运行状态的实例才能终止"); // 如果不是"Running"状态，则不执行终止操作
        }

        // 创建请求数据map
        Map<String, Object> requestData = new HashMap<>();
        requestData.put("namespace", namespace);
        requestData.put("name", name);
        // 创建发送数据map，将请求数据作为"data"键的值
        Map<String, Object> res = new HashMap<>();
        res.put("data", requestData);

        try {
            // 发送POST请求到Argo工作流状态查询接口，并将请求数据转换为JSON
            String req = HttpUtils.sendPost(argoUrl + argoWorkflowTermination, null, JsonUtils.mapToJson(res));
            // 检查响应是否为空或无内容
            if (StringUtils.isEmpty(req)) {
                throw new RuntimeException("终止响应内容为空。");

            }
            // 将响应的JSON字符串转换为Map对象
            Map<String, Object> runResMap = JsonUtils.jsonToMap(req);
            // 从响应Map中直接获取"errCode"的值
            Integer errCode = (Integer) runResMap.get("errCode");
            if (errCode != null && errCode == 0) {
                //更新autoMlIns,确保状态更新被保存到数据库
                RayIns ins = queryStatusFromArgo(rayIns);
                String nodeStatus = ins.getNodeStatus();
                Map<String, Object> nodeMap = JsonUtils.jsonToMap(nodeStatus);

                // 遍历 map
                for (Map.Entry<String, Object> entry : nodeMap.entrySet()) {
                    // 获取每个 Map 中的值并强制转换为 Map
                    Map<String, Object> innerMap = (Map<String, Object>) entry.getValue();
                    // 检查 phase 的值
                    if (innerMap.containsKey("phase")) {
                        String phaseValue = (String) innerMap.get("phase");
                        // 如果值不等于 Succeeded，则赋值为 Failed
                        if (!StringUtils.equals("Succeeded", phaseValue)) {
                            innerMap.put("phase", "Failed");
                        }
                    }
                }
                ins.setNodeStatus(JsonUtils.mapToJson(nodeMap));
                ins.setStatus(Constant.Terminated);
                ins.setUpdateTime(new Date());
                rayInsDao.update(ins);
                updateRayStatus(rayIns.getRayId());
                return true;
            } else {
                return false;
            }
        } catch (Exception e) {
            throw new RuntimeException("终止实例错误: " + e.getMessage(), e);
        }
    }

    @Override
    public RayIns getDetailById(Long id) throws Exception {
        RayIns rayIns = rayInsDao.queryById(id);
        if (Constant.Running.equals(rayIns.getStatus()) || Constant.Pending.equals(rayIns.getStatus())) {
            rayIns = queryStatusFromArgo(rayIns);
        }
        getTrialList(rayIns);

//        rayIns.setTrialList(getTrialList(rayIns.getResultPath()));
        return rayIns;
    }

    @Override
    public void updateRayStatus(Long rayId) {
        List<RayIns> insList = rayInsDao.getByRayId(rayId);
        List<String> statusList = new ArrayList<>();
        // 更新实验状态列表
        for (int i = 0; i < insList.size(); i++) {
            statusList.add(insList.get(i).getStatus());
        }
        String subStatus = statusList.toString().substring(1, statusList.toString().length() - 1);
        Ray ray = rayDao.getRayById(rayId);
        if (!StringUtils.equals(ray.getStatusList(), subStatus)) {
            ray.setStatusList(subStatus);
            rayDao.edit(ray);
        }
    }

    @Override
    public RayIns queryStatusFromArgo(RayIns ins) {
        String namespace = ins.getArgoInsNs();
        String name = ins.getArgoInsName();

        // 创建请求数据map
        Map<String, Object> requestData = new HashMap<>();
        requestData.put("namespace", namespace);
        requestData.put("name", name);

        // 创建发送数据map，将请求数据作为"data"键的值
        Map<String, Object> res = new HashMap<>();
        res.put("data", requestData);

        try {
            // 发送POST请求到Argo工作流状态查询接口，并将请求数据转换为JSON
            String req = HttpUtils.sendPost(argoUrl + argoWorkflowStatus, null, JsonUtils.mapToJson(res));
            // 检查响应是否为空或无内容
            if (req == null || StringUtils.isEmpty(req)) {
                throw new RuntimeException("工作流状态响应为空。");
            }
            // 将响应的JSON字符串转换为Map对象
            Map<String, Object> runResMap = JsonUtils.jsonToMap(req);
            // 从响应Map中获取"data"部分
            Map<String, Object> data = (Map<String, Object>) runResMap.get("data");
            if (data == null || data.isEmpty()) {
                throw new RuntimeException("工作流数据为空.");
            }
            // 从"data"中获取"status"部分，并返回"phase"的值
            Map<String, Object> status = (Map<String, Object>) data.get("status");
            if (status == null || status.isEmpty()) {
                throw new RuntimeException("工作流状态为空。");
            }

            //解析流水线结束时间
            String finishedAtString = (String) status.get("finishedAt");
            if (finishedAtString != null && !finishedAtString.isEmpty()) {
                Date finishTime = DateUtils.convertUTCtoShanghaiDate(finishedAtString);
                ins.setFinishTime(finishTime);
            }

            // 解析nodes字段,提取节点状态并转换为JSON字符串
            Map<String, Object> nodes = (Map<String, Object>) status.get("nodes");
            Map<String, Object> modifiedNodes = new LinkedHashMap<>();
            if (nodes != null) {
                for (Map.Entry<String, Object> nodeEntry : nodes.entrySet()) {
                    Map<String, Object> nodeDetails = (Map<String, Object>) nodeEntry.getValue();
                    String templateName = (String) nodeDetails.get("displayName");
                    modifiedNodes.put(templateName, nodeDetails);
                }
            }


            String nodeStatusJson = JsonUtils.mapToJson(modifiedNodes);
            ins.setNodeStatus(nodeStatusJson);

            //终止态为终止不改
            if (!StringUtils.equals(ins.getStatus(), Constant.Terminated)) {
                ins.setStatus(StringUtils.isNotEmpty((String) status.get("phase")) ? (String) status.get("phase") : Constant.Pending);
            }
            if (StringUtils.equals(ins.getStatus(), "Error")) {
                ins.setStatus(Constant.Failed);
            }
            return ins;
        } catch (Exception e) {
            throw new RuntimeException("查询状态失败: " + e.getMessage(), e);
        }
    }

    @Override
    public List<RayIns> queryByRayInsIsNotTerminated() {
        return rayInsDao.queryByRayInsIsNotTerminated();
    }

    public void getTrialList(RayIns rayIns) throws Exception {
        // 获取指定路径下的所有文件
        String directoryPath = rayIns.getResultPath();

        rayIns.setResultPath(endpoint + "/" + directoryPath);
        rayIns.setResultTxt(endpoint + "/" + directoryPath + "/result.txt");

        String prefix = directoryPath.substring(directoryPath.indexOf("/") + 1, directoryPath.length()) + "/";
        List<Map> maps = minioUtil.listFilesInDirectory(bucketName, prefix);

        if (!maps.isEmpty()) {
            rayIns.setFileList(maps);
            List<Map> collect = maps.stream().filter(map -> map.get("name").toString().startsWith("experiment_state")).collect(Collectors.toList());
            if (!collect.isEmpty()) {
                Path experimentState = Paths.get(collect.get(0).get("name").toString());
                String content = minioUtil.readObjectAsString(bucketName, prefix + "/" + experimentState);

                Map<String, Object> result = JsonUtils.jsonToMap(content);
                ArrayList<ArrayList> trial_data_list = (ArrayList<ArrayList>) result.get("trial_data");
                ArrayList<Map<String, Object>> trialList = new ArrayList<>();

                for (ArrayList trial_data : trial_data_list) {
                    Map<String, Object> trial_data_0 = JsonUtils.jsonToMap((String) trial_data.get(0));
                    Map<String, Object> trial_data_1 = JsonUtils.jsonToMap((String) trial_data.get(1));

                    Map<String, Object> trial = new HashMap<>();
                    trial.put("trial_id", trial_data_0.get("trial_id"));
                    trial.put("config", trial_data_0.get("config"));
                    trial.put("status", trial_data_0.get("status"));

                    Map<String, Object> last_result = (Map<String, Object>) trial_data_1.get("last_result");
                    Map<String, Object> metric_analysis = (Map<String, Object>) trial_data_1.get("metric_analysis");
                    Map<String, Object> time_total_s = (Map<String, Object>) metric_analysis.get("time_total_s");

                    trial.put("training_iteration", last_result.get("training_iteration"));
                    trial.put("time", time_total_s.get("avg"));

                    trialList.add(trial);
                }
                rayIns.setTrialList(trialList);
            }
        }
    }
}
