package com.yingxin.prms.service.business;

import com.yingxin.prms.config.filter.TokenInterceptor;
import com.yingxin.prms.domain.*;
import com.yingxin.prms.dto.*;
import com.yingxin.prms.service.asyncTask.QueryHostStatusTask;
import com.yingxin.prms.service.dao.CassandraDaoImpl;
import com.yingxin.prms.service.dao.FlowDaoImpl;
import com.yingxin.prms.utils.CurveFittingMethod;
import com.yingxin.prms.utils.SendHttpUtil;
import com.yingxin.prms.utils.TimeUtil;
import com.yingxin.prms.utils.TokenUtil;
import net.sf.json.JSONObject;
import org.kairosdb.client.HttpClient;
import org.kairosdb.client.builder.*;
import org.kairosdb.client.response.QueryResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


@Service
public class Manager {

    @Value("${ascynTaskThreadPool.clusterStatusPoolSize}")
    private int THREADPOOLSIZE;

    @Value("${tokeTime}")
    private long tokeTime;

    @Value("${pointNumber}")
    private int pointNumber;

    private final AuthModeAndHostList authModeAndHostList;

    private final CassandraDaoImpl cassandraDao;

    @Autowired
    private FlowDaoImpl flowDao;

    private RestTemplate restTemplate;

    private final HttpClient httpClient;

    @Autowired
    private TokenInterceptor tokenInterceptor;


    @Autowired
    public Manager(AuthModeAndHostList authModeAndHostList, CassandraDaoImpl cassandraDao, HttpClient httpClient) {
        restTemplate = new RestTemplate();
        this.authModeAndHostList = authModeAndHostList;
        this.cassandraDao = cassandraDao;
        this.httpClient = httpClient;
    }


    /**
     * 获取二包的服务器ip和port
     *
     * @return 服务器ip和port列表
     */
    public List selectSecondIpAndPort() {
        List<Map> hostInfos = new ArrayList<>();
        for (CTID_Process_Host_Dic processHost : authModeAndHostList.getProcessHostList()) {
            HashMap<String, Object> hostInfo = new HashMap<>();
            hostInfo.put("ip", processHost.getIp());
            hostInfo.put("port", processHost.getPort());
            hostInfos.add(hostInfo);
        }
        return hostInfos;

    }

    public Map<String, Object> getAlarmLog(String json) {
        Map<String,Object> map = new HashMap();
        List<AlarmLog> alarmList = cassandraDao.getAlarmList();
        map.put("alarm",alarmList);
        JSONObject jsonObject  = JSONObject.fromObject(json);
        int pageNum = jsonObject.getInt("pageNum");
        int pageSize = jsonObject.getInt("pageSize");

        int i = 0;
        String flag = "";
        Map total = cassandraDao.getAlarmLogSize();
        List<AlarmLog> logList = null;
        do{
            logList = cassandraDao.getAlarmLog(pageSize,flag);

            if (logList.size()==0){
                break;
            }
            AlarmLog alarmLog = logList.get(logList.size()-1);
            flag = alarmLog.getSerialnum();
            i++;
        }while(i<pageNum);

        map.put("log",logList);
        map.put("logSize",total.get("count"));
        return map;
    }
    public Map selectClusterStatus() {
        /*查询集群监控列表*/
        List<Server_Host_Info> host_info_List = authModeAndHostList.getClusterBeMonitoredList();
        List<SingleHostStatus> accessList = new ArrayList<>();
        List<SingleHostStatus> dataList = new ArrayList<>();
        List<SingleHostStatus> passportList = new ArrayList<>();
        List<SingleHostStatus> coreList = new ArrayList<>();

        List<Future> accessFuture = new ArrayList<>();
        List<Future> dataFuture = new ArrayList<>();
        List<Future> passportFuture = new ArrayList<>();
        List<Future> futureList = new ArrayList<>();
        List<Future> coreFuture = new ArrayList<>();
        List<SingleHostStatus> hostStatusList = new ArrayList<>();
        /*线程池*/
        ExecutorService service = Executors.newFixedThreadPool(THREADPOOLSIZE);

        for (Server_Host_Info hostInfo : host_info_List) {
            Future<SingleHostStatus> result = service.submit(new QueryHostStatusTask(hostInfo, httpClient));
            futureList.add(result);
            if (hostInfo.getLayer().equals("接入区")){
                accessFuture.add(result);
            }else if(hostInfo.getLayer().equals("算法区")){
                dataFuture.add(result);
            }else if(hostInfo.getLayer().equals("护照柜")){
                passportFuture.add(result);
            }else if(hostInfo.getLayer().equals("核心业务区")){
                coreFuture.add(result);
            }
        }
        service.shutdown();
        try {
            if (service.awaitTermination(3, TimeUnit.MINUTES)) {
                for (Future future : futureList) {
                    SingleHostStatus singleHostStatus = (SingleHostStatus) future.get();
                    hostStatusList.add(singleHostStatus);
                }
                for (Future future : accessFuture) {
                    SingleHostStatus singleHostStatus = (SingleHostStatus) future.get();
                    accessList.add(singleHostStatus);
                }
                for (Future future : dataFuture) {
                    SingleHostStatus singleHostStatus = (SingleHostStatus) future.get();
                    dataList.add(singleHostStatus);
                }
                for (Future future : passportFuture) {
                    SingleHostStatus singleHostStatus = (SingleHostStatus) future.get();
                    passportList.add(singleHostStatus);
                }
                for (Future future : coreFuture) {
                    SingleHostStatus singleHostStatus = (SingleHostStatus) future.get();
                    coreList.add(singleHostStatus);
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        Map accessMap = selectClusterStatusBylayer(accessList);
        Map dataMap = selectClusterStatusBylayer(dataList);
        Map passportMap = selectClusterStatusBylayer(passportList);
        Map coreMap = selectClusterStatusBylayer(coreList);
        Map map  = new HashMap();
        map.put("接入区",accessMap);
        map.put("算法区",dataMap);
        map.put("护照柜",passportMap);
        map.put("核心业务区",coreMap);
        return map;
    }
    public Map selectClusterStatusBylayer(List<SingleHostStatus> hostStatusList){
        if (hostStatusList.size() != 0) {
            Map<String, Object> clusterStatus = new HashMap<>();
            Map<String, List<SingleHostStatus>> listMap = hostStatusList.stream().collect(Collectors.groupingBy(SingleHostStatus::getClusterName));
            for (Map.Entry<String, List<SingleHostStatus>> singleClusterStatusEntry : listMap.entrySet()) {
                String clusterName = singleClusterStatusEntry.getKey();
                Map<String, Object> singleClusterStatus = new HashMap<>();
                List<SingleHostStatus> singleHostStatusList = singleClusterStatusEntry.getValue();
                for (SingleHostStatus singleHostStatus : singleHostStatusList) {
                    singleClusterStatus.put(singleHostStatus.getHostName(), singleHostStatus.getHostStatus());
                }
                clusterStatus.put(clusterName, singleClusterStatus);
            }
            return clusterStatus;
        }
        return null;
    }

    public Map login(String json) {
        Map<String,Object>resultMap=new HashMap<>();
        JSONObject jsonObject  = JSONObject.fromObject(json);
        String username = jsonObject.getString("username");
        String password = jsonObject.getString("password");
        Map<String, String> tokenList = tokenInterceptor.getTokenList();
        Admin admin = cassandraDao.findByUsername(username);
        String passwordNew = TimeUtil.getMD5Result(password);
        if(admin!=null && admin.getPassword().equals(passwordNew)) {
            String token = TokenUtil.sign(tokeTime);
            resultMap.put("flag", true);
            resultMap.put("token", token);
            long time = (long) (tokeTime * 0.8);
            if(time>0){
                resultMap.put("time", time);
            }else{
                resultMap.put("time", tokeTime);
            }
            tokenList.put(username,token);
        }else{
            resultMap.put("flag", false);
        }
        return resultMap;

    }

    public Map<String, Object> getToken(String json) {
        Map<String, String> tokenList = tokenInterceptor.getTokenList();
        JSONObject jsonObject = JSONObject.fromObject(json);
        String username = jsonObject.getString("username");
        Map<String,Object>resultMap=new HashMap<>();
        String token = TokenUtil.sign(tokeTime);
        resultMap.put("token",token);
        tokenList.put(username,token);
        return resultMap;
    }

    public Map selectChart(String json) throws Exception{
        JSONObject jsonObject  = JSONObject.fromObject(json);
        String url = jsonObject.getString("url");
        String flag = jsonObject.getString("flag");

        Server_Host_Info hostInfo = cassandraDao.findHostInfo(url);
        return getData(hostInfo.getService_path(),flag);
    }

    private Map getData(String servicePath,String flag) throws Exception {


        /* 获取前六天数据 datapoints 具体时间段 数据 */
        List<DataPoint> dataPoints = getAgodataPoints(servicePath,flag);
        /* 获取今天数据 */
        List<DataPoint> currDataPoint = getdataPoints(servicePath,flag);

        /* 获取当前需要展示的横坐标 */
        double[] xFlag = null;
        String[] xResult = null;
        double[] yResult = null;
        double[] yForecast = null;
        if(currDataPoint.size() > 0 ){
            xFlag = new double[currDataPoint.size()];
            xResult=new String[currDataPoint.size()];
            yResult = new double[currDataPoint.size()];
            yForecast = new double[currDataPoint.size()];
            for (int i=0;i<currDataPoint.size();++i){
                xFlag[i] = currDataPoint.get(i).getTimestamp();
                xResult[i]= TimeUtil.toDate(currDataPoint.get(i).getTimestamp());
                yResult[i] = (int)Double.parseDouble(String.valueOf(currDataPoint.get(i).getValue()));
                yForecast[i] = 0;
            }
        }
        Map map = new HashMap();
        if(dataPoints.size()>1){
            double [] x=new double[dataPoints.size()];
            double [] y=new double[dataPoints.size()];
            for (int i=0;i<dataPoints.size();i++){
                x[i]= dataPoints.get(i).getTimestamp();
            }
            for (int i=0;i<dataPoints.size();i++){
                y[i]=Double.parseDouble(String.valueOf(dataPoints.get(i).getValue()) );
            }
            //阶数为4，固定值
            Map mapForecast = CurveFittingMethod.run(x, y,4,xFlag);
            //横坐标
            map.put("abscissa",xResult);
            //纵坐标
            map.put("ordinate",yResult);
            //纵坐标预测值
            map.put("ordinateForecast",mapForecast.get("y_data_array"));
        }else{
            map.put("abscissa",xResult);
            map.put("ordinate",yResult);
            map.put("ordinateForecast",yForecast);
        }
        return map;


    }

    private List<DataPoint> getdataPoints(String servicePath,String flag) throws Exception {
        QueryBuilder queryBuilder = QueryBuilder.getInstance();
        Calendar calendar = Calendar.getInstance();
        int number = pointNumber+ 1;
        switch (flag){
            case("one"):
                calendar.set(Calendar.MINUTE,
                        calendar.get(Calendar.MINUTE) - number);
                queryBuilder.setStart(calendar.getTime())
                        .setEnd(new Date())
                        .addMetric("respdur")
                        .addTag("host",servicePath)
                        .addAggregator(AggregatorFactory.createAverageAggregator(1, org.kairosdb.client.builder.TimeUnit.MINUTES));
                break;
            case("ten"):
                calendar.set(Calendar.MINUTE,
                        calendar.get(Calendar.MINUTE) - (number*10));
                queryBuilder.setStart(calendar.getTime())
                        .setEnd(new Date())
                        .addMetric("respdur")
                        .addTag("host",servicePath)
                        .addAggregator(AggregatorFactory.createAverageAggregator(10, org.kairosdb.client.builder.TimeUnit.MINUTES));

                break;
            case("thirty"):
                calendar.set(Calendar.MINUTE,
                        calendar.get(Calendar.MINUTE) - (number*30));
                queryBuilder.setStart(calendar.getTime())
                        .setEnd(new Date())
                        .addMetric("respdur")
                        .addTag("host",servicePath)
                        .addAggregator(AggregatorFactory.createAverageAggregator(30, org.kairosdb.client.builder.TimeUnit.MINUTES));
                break;
            case("hour"):
                calendar.set(Calendar.HOUR,
                        calendar.get(Calendar.HOUR) - number);
                queryBuilder.setStart(calendar.getTime())
                        .setEnd(new Date())
                        .addMetric("respdur")
                        .addTag("host",servicePath)
                        .addAggregator(AggregatorFactory.createAverageAggregator(1, org.kairosdb.client.builder.TimeUnit.HOURS));
                break;
        }
        QueryResponse response = httpClient.query(queryBuilder);
        return response.getQueries().get(0).getResults().get(0).getDataPoints();
    }


    private List<DataPoint> getAgodataPoints(String servicePath,String flag) throws IOException {
        QueryBuilder queryBuilder = QueryBuilder.getInstance() ;
        Date currDate = new Date();
        QueryResponse response = null;
        List<DataPoint> dataPoints  = new ArrayList<>();
        for(int i = 0 ;i<6 ;i++){
            currDate = TimeUtil.PreviousDay(currDate);
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(currDate);
            int number = pointNumber + 1;
            switch (flag){
                case("one"):
                    calendar.set(Calendar.MINUTE,
                            calendar.get(Calendar.MINUTE) - number);
                    queryBuilder.setStart(calendar.getTime())
                            .setEnd(currDate)
                            .addMetric("respdur")
                            .addTag("host",servicePath)
                            .addAggregator(AggregatorFactory.createAverageAggregator(1, org.kairosdb.client.builder.TimeUnit.MINUTES));
                    break;
                case("ten"):
                    calendar.set(Calendar.MINUTE,
                            calendar.get(Calendar.MINUTE) - (number*10));
                    queryBuilder.setStart(calendar.getTime())
                            .setEnd(currDate)
                            .addMetric("respdur")
                            .addTag("host",servicePath)
                            .addAggregator(AggregatorFactory.createAverageAggregator(10, org.kairosdb.client.builder.TimeUnit.MINUTES));
                    break;
                case("thirty"):
                    calendar.set(Calendar.MINUTE,
                            calendar.get(Calendar.MINUTE) - (number*30));
                    queryBuilder.setStart(calendar.getTime())
                            .setEnd(currDate)
                            .addMetric("respdur")
                            .addTag("host",servicePath)
                            .addAggregator(AggregatorFactory.createAverageAggregator(30, org.kairosdb.client.builder.TimeUnit.MINUTES));
                    break;
                case("hour"):
                    calendar.set(Calendar.HOUR,
                            calendar.get(Calendar.HOUR) - number);
                    queryBuilder.setStart(calendar.getTime())
                            .setEnd(currDate)
                            .addMetric("respdur")
                            .addTag("host",servicePath)
                            .addAggregator(AggregatorFactory.createAverageAggregator(1, org.kairosdb.client.builder.TimeUnit.HOURS));
                    break;
            }
            response = httpClient.query(queryBuilder);
            if(response.getQueries().size() > 0 ){
                dataPoints.addAll(response.getQueries().get(0).getResults().get(0).getDataPoints());
            }
        }
        return dataPoints;
    }

    public Map selectTopology(){
        Map result = new HashMap();
        List<FlowNode> flowNodes = flowDao.findNodeAll();
        List<FlowLine> flowLines = flowDao.findLineAll();
        List<TcpFlowLineDto> lineDtos = new ArrayList<>();
        List<TcpFlowNodeDto> nodeDtos = new ArrayList<>();
        for(FlowNode node : flowNodes){
            TcpFlowNodeDto nodeDto = new TcpFlowNodeDto();
            long faultCount=0;
            if(node.getServerPath()==null || node.getServerPath().isEmpty()){
                nodeDto.setCategory(node.getMark());
            }else{
                try {
                    faultCount = getFaultCount(node.getServerPath());
                    if(faultCount>0){
                        /* machine浅色图标 clusterFault错误图标 clusterNormal正常图标 meaningless无意义 */
                        nodeDto.setCategory("clusterFault");
                    }else{
                        nodeDto.setCategory("clusterNormal");
                    }
                } catch (Exception e) {
                    faultCount = 0;
                }
            }
            nodeDto.setFaultCount((int) faultCount);
            nodeDto.setKey(node.getCluster_name());
            nodeDto.setDescription("错误数");
            nodeDto.setUrl(node.getServerPath());
            nodeDto.setServerName(node.getCluster_name());
            nodeDto.setText(node.getCluster_name());
            nodeDto.setPos(node.getCoordinate());
            nodeDtos.add(nodeDto);
        }

        for(FlowLine line : flowLines){
            TcpFlowLineDto lineDto = new TcpFlowLineDto();
            lineDto.setFrom(line.getCluster_name_left());
            lineDto.setTo(line.getClusterNameRight());
            lineDto.setFromSpot(line.getCoordinateLeft());
            lineDto.setToSpot(line.getCoordinateRight());
            lineDto.setColor(line.getColor());
            String point = line.getPoints();
            if(point!=null && point.length()>0){
                point = point.substring(1,point.length()-1);
                point =  point.replaceAll("\\\\", "");
                point =  point.replaceAll("\"", "");
                List list = new ArrayList();
                String [] mapArray = point.split(",");
                for (int j = 0; j < mapArray.length; j++) {
                    list.add(Double.valueOf(mapArray[j]));
                }
                lineDto.setPoints(list);
            }else{
                lineDto.setPoints(null);
            }
            lineDtos.add(lineDto);
        }
        result.put("class", "GraphLinksModel");
        result.put("nodeDataArray", nodeDtos);
        result.put("linkDataArray", lineDtos);

       return result;
    }

    private long getFaultCount(String servicePath) throws IOException, DataFormatException {
        Aggregator aggregatorDay = AggregatorFactory.createSumAggregator(1, org.kairosdb.client.builder.TimeUnit.DAYS).withStartTimeAlignment(0);
        QueryBuilder queryBuilder = QueryBuilder.getInstance();
        queryBuilder.setStart(1, org.kairosdb.client.builder.TimeUnit.DAYS).setTimeZone(TimeZone.getDefault())
                .addMetric("heartbeat")
                .addTag("host", servicePath)
                .addAggregator(aggregatorDay);
        QueryResponse response = httpClient.query(queryBuilder);
        List<DataPoint> dataPointsDay = response.getQueries().get(0).getResults().get(0).getDataPoints();
        return dataPointsDay.size() == 0 ? 0 : dataPointsDay.get(dataPointsDay.size() - 1).longValue();
    }

    public Map saveTopology(String json) {
        JSONObject jsonObject = JSONObject.fromObject(json);
        List<JSONObject> nodeList = jsonObject.getJSONArray("nodeDataArray");
        List<JSONObject> linkList = jsonObject.getJSONArray("linkDataArray");
        for(JSONObject js : nodeList){
            if(!js.isNullObject()){
                flowDao.updateNode(js.getString("serverName"),js.getString("pos"), js.getString("category"));
            }
        }
        for(JSONObject js : linkList){
            if(!js.isNullObject()){
                flowDao.updateLine(js.getString("from"),js.getString("to"), js.getString("points"),js.getString("color"));
            }
        }
        Map result = new HashMap();
        result.put("state",true);
        /*线程池*/
        return result;
    }


    public Map getStatistics() {
        Server_Host_Info serverHostInfo = cassandraDao.getStatisticsUri();
        Map<String, String> map = SendHttpUtil.doGet(serverHostInfo.getService_path());
        return map;
    }

    /*
    public Map selectClusterStatus() {
        List<Server_Host_Info> host_info_List = authModeAndHostList.getClusterBeMonitoredList();
        List<Server_Host_Info> accessList = new ArrayList<>();
        List<Server_Host_Info> serviceList = new ArrayList<>();
            for(Server_Host_Info hostInfo : host_info_List){
            if (hostInfo.getLayer().equals("接入区")){
                accessList.add(hostInfo);
            }else {
                serviceList.add(hostInfo);
            }
        }
        Map accessMap = selectClusterStatusBylayer(accessList);
        Map serviceMap = selectClusterStatusBylayer(serviceList);
        Map map  = new HashMap();
            map.put("接入区",accessMap);
            map.put("服务区",serviceMap);
            return map;
    }

    public Map selectClusterStatusBylayer(List<Server_Host_Info> host_info_List){
        List<SingleHostStatus> hostStatusList = new ArrayList<>();
        List<Future> futureList = new ArrayList<>();


        ExecutorService service = Executors.newFixedThreadPool(THREADPOOLSIZE);

        for (Server_Host_Info hostInfo : host_info_List) {
            Future<SingleHostStatus> result = service.submit(new QueryHostStatusTask(hostInfo, httpClient));
            futureList.add(result);
        }
        service.shutdown();
        try {
            if (service.awaitTermination(3, TimeUnit.MINUTES)) {
                for (Future future : futureList) {
                    SingleHostStatus singleHostStatus = (SingleHostStatus) future.get();
                    hostStatusList.add(singleHostStatus);
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        if (hostStatusList.size() != 0) {
            Map<String, Object> clusterStatus = new HashMap<>();
            Map<String, List<SingleHostStatus>> listMap = hostStatusList.stream().collect(Collectors.groupingBy(SingleHostStatus::getClusterName));
            for (Map.Entry<String, List<SingleHostStatus>> singleClusterStatusEntry : listMap.entrySet()) {
                String clusterName = singleClusterStatusEntry.getKey();
                Map<String, Object> singleClusterStatus = new HashMap<>();
                List<SingleHostStatus> singleHostStatusList = singleClusterStatusEntry.getValue();
                for (SingleHostStatus singleHostStatus : singleHostStatusList) {
                    singleClusterStatus.put(singleHostStatus.getHostName(), singleHostStatus.getHostStatus());
                }
                clusterStatus.put(clusterName, singleClusterStatus);
            }
            return clusterStatus;
        }
        return null;
    }
    * */

//    /**
//     * 获取一包基准值
//     * @return 基准值实体类
//     */
//    public List<CTID_Request_Perf_BaseLineValue> getReqBLV() {
//        return cassandraDao.getReqBLV();
//    }
//
//    /**
//     * 获取二包基准值
//     * @return 基准值实体类
//     */
//    public List<CTID_Process_Perf_BaseLineValue> getProBLV() {
//        return cassandraDao.getProBLV();
//    }
//
//    /**
//     * 更新一包性能基准值
//     * @param entity 新的一包基准值
//     * @return 更新成功返回的基准值
//     */
//    public List<CTID_Request_Perf_BaseLineValue> updateReqBLV(CTID_Request_Perf_BaseLineValue entity) {
//        cassandraDao.saveReqBLV(entity);
//        return cassandraDao.getReqBLV();
//    }
//
//    /**
//     * 更新二包性能基准值
//     * @param entity 新的二包基准值
//     * @return 更新成功返回的基准值
//     */
//    public List<CTID_Process_Perf_BaseLineValue> updateProBLV(CTID_Process_Perf_BaseLineValue entity) {
//        cassandraDao.saveProBLV(entity);
//        return cassandraDao.getProBLV();
//    }
/*public static void main(String[] args) {
    long time = 1607304515793L;
    Date date = new Date(time);
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy MM dd hh:mm:ss");
    System.out.println(sdf.format(date));
}*/


}
