package com.ruoyi.platform.service.impl;

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.service.ResourceOccupyService;
import com.ruoyi.platform.utils.*;
import com.ruoyi.system.api.constant.Constant;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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 org.springframework.transaction.annotation.Transactional;

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

@Service("rayInsService")
public class RayInsServiceImpl implements RayInsService {
    private static final Logger logger = LoggerFactory.getLogger(RayInsServiceImpl.class);

    @Value("${argo.url}")
    private String argoUrl;
    @Value("${argo.workflowStatus}")
    private String argoWorkflowStatus;
    @Value("${argo.workflowTermination}")
    private String argoWorkflowTermination;
    @Value("${minio.endpointIp}")
    String endpoint;

    @Value("${aim.url}")
    private String aimUrl;
    @Value("${aim.proxyUrl}")
    private String aimProxyUrl;

    @Resource
    private RayInsDao rayInsDao;

    @Resource
    private RayDao rayDao;

    @Resource
    private ResourceOccupyService resourceOccupyService;

    @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 RayIns update(RayIns rayIns) {
        rayIns.setUpdateTime(new Date());
        rayIns = queryStatusFromArgo(rayIns);
        rayInsDao.update(rayIns);
        updateRayStatus(rayIns.getRayId());
        if (Constant.Failed.equals(rayIns.getStatus()) || Constant.Succeeded.equals(rayIns.getStatus())) {
            resourceOccupyService.endDeduce(Constant.TaskType_Ray, null, rayIns.getId(), null, null);
        }
        return rayIns;
    }

    @Override
    @Transactional
    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) {
            resourceOccupyService.deleteTaskState(Constant.TaskType_Ray, rayIns.getRayId(), id);
            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(Constant.Succeeded, phaseValue)) {
                            innerMap.put("phase", Constant.Failed);
                            Calendar calendar = Calendar.getInstance();
                            calendar.add(Calendar.HOUR_OF_DAY, -8);
                            innerMap.put("finishedAt", new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").format(calendar.getTime()));
                        }
                    }
                }
                ins.setNodeStatus(JsonUtils.mapToJson(nodeMap));
                ins.setStatus(Constant.Terminated);
                ins.setUpdateTime(new Date());
                ins.setFinishTime(new Date());
                rayInsDao.update(ins);
                updateRayStatus(rayIns.getRayId());
                // 结束扣积分
                resourceOccupyService.endDeduce(Constant.TaskType_Ray, null, id, null, null);
                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);
        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(), Constant.Error)) {
                ins.setStatus(Constant.Failed);
            }
            return ins;
        } catch (Exception e) {
            throw new RuntimeException("查询状态失败: " + e.getMessage(), e);
        }
    }

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

    @Override
    public String getExpMetrics(String trailIds) throws Exception {
        String encodedUrlString = URLEncoder.encode("run.trial_id in " + trailIds, "UTF-8");
        String url = aimProxyUrl + "/api/runs/search/run?query=" + encodedUrlString;
        String s = HttpUtils.sendGet(url, null);
        List<Map<String, Object>> responses = JacksonUtil.parseJSONStr2MapList(s);

        List<String> runIds = new ArrayList<>();
        for (Map response : responses) {
            runIds.add((String) response.get("run_hash"));
        }

        String decode = AIM64EncoderUtil.decode(runIds);
        return aimUrl + "/metrics?select=" + decode;
    }

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

        rayIns.setResultPath(endpoint + "/" + directoryPath);
        rayIns.setResultTxt(endpoint + "/" + directoryPath + "/result.txt");
        try {
            String bucketName = directoryPath.substring(0, directoryPath.indexOf("/"));
            String prefix = directoryPath.substring(directoryPath.indexOf("/") + 1, directoryPath.length()) + "/";
            List<Map> fileMaps = minioUtil.listRayFilesInDirectory(bucketName, prefix);

            if (!fileMaps.isEmpty()) {
                List<Map> collect = fileMaps.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);

                    String resultTxt = minioUtil.readObjectAsString(bucketName, prefix + "result.txt");
                    String bestMetrics = getStringBetween(resultTxt, "Best metrics:", "Best result_df");
                    Map<String, Object> bestMetricsMap = JsonUtils.jsonToMap(bestMetrics);
                    String bestTrialId = (String) bestMetricsMap.get("trial_id");

                    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));

                        String trialId = (String) trial_data_0.get("trial_id");
                        Map<String, Object> trial = new HashMap<>();
                        trial.put("trial_id", trialId);
                        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_avg", time_total_s.get("avg"));

                        Map<String, Object> param = JsonUtils.jsonToMap(rayIns.getParam());
                        trial.put("metric_analysis", metric_analysis.get((String) param.get("metric")));
                        trial.put("metric", param.get("metric"));

                        for (Map fileMap : fileMaps) {
                            if (fileMap.get("name").toString().contains(trialId)) {
                                trial.put("file", fileMap);
                            }
                        }

                        if (bestTrialId.equals(trialId)) {
                            trial.put("is_best", true);
                            trialList.add(0, trial);
                        } else {
                            trialList.add(trial);
                        }
                    }
                    rayIns.setTrialList(trialList);
                }
            }
        } catch (Exception e) {
            logger.error("未找到结果文件");
        }
    }

    String getStringBetween(String input, String startMarker, String endMarker) {
        int startIndex = input.indexOf(startMarker);
        if (startIndex == -1) {
            return ""; // 如果未找到起始标记，返回空字符串
        }

        // 跳过起始标记
        startIndex += startMarker.length();

        int endIndex = input.indexOf(endMarker, startIndex);
        if (endIndex == -1) {
            return ""; // 如果未找到结束标记，返回空字符串
        }

        return input.substring(startIndex, endIndex);
    }
}


