package com.iailab.module.model.influxdb.service.impl;

import com.iailab.module.model.influxdb.common.config.InfluxDBInstance;
import com.iailab.module.model.influxdb.common.utils.MeasurementUtils;
import com.iailab.module.model.influxdb.pojo.InfluxModelResultByOutPutIdsPOJO;
import com.iailab.module.model.influxdb.pojo.InfluxModelResultPOJO;
import com.iailab.module.model.influxdb.service.InfluxDBService;
import com.iailab.module.model.influxdb.vo.InfluxModelResultVO;
import com.influxdb.client.QueryApi;
import com.influxdb.client.WriteApi;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * InfluxDB操作类
 */
@Slf4j
@Service
public class InfluxDBServiceImpl implements InfluxDBService {
    @Resource
    private InfluxDBInstance influxDBInstance;
    private WriteApi writeApi;
    private QueryApi queryApi;

    @PostConstruct
    private void init() {
        writeApi = influxDBInstance.getClient().makeWriteApi();
        queryApi = influxDBInstance.getClient().getQueryApi();
    }

    @Override
    public void asyncWriteModelResults(List<InfluxModelResultPOJO> pointValues) {
        if (!CollectionUtils.isEmpty(pointValues)) {
            pointValues.forEach(item -> {
                String bucket = influxDBInstance.getBucket();
                writeApi.writeMeasurement(bucket, influxDBInstance.org, WritePrecision.MS, item);
            });
        }
        writeApi.flush();
    }

    @Override
    public List<InfluxModelResultVO> queryModelResults(InfluxModelResultPOJO pojo, Date startTime, Date endTime) {
        Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.MILLISECOND, 0);
        String start = startTime.toInstant().toString();

        if (startTime.getTime() == endTime.getTime()) {
            // 如果相等，则engTime加1毫秒，否则influxdb报错(因为influxdb的range函数是左闭右开区间，所以将engTime加一毫秒，才可以查到startTime时间点的数据)
            endTime.setTime(endTime.getTime() + 1);
        }
        String stop = endTime.toInstant().toString();

        List<InfluxModelResultVO> dataList = new ArrayList<>();
        String measurement = MeasurementUtils.getMeasurement(pojo.getType());
        StringBuilder sb = new StringBuilder();
        sb.append("from(bucket:\"" + influxDBInstance.getBucket() + "\") ");
        sb.append("|> range(start: ").append(start).append(", stop: ").append(stop).append(") ");
        sb.append("|> filter(fn: (r) => r[\"_measurement\"] == \"" + measurement + "\")");
        sb.append("|> filter(fn: (r) => r[\"_field\"] == \"value\")");
        sb.append("|> filter(fn: (r) => r[\"outPutId\"] == \"" + pojo.getOutPutId() + "\")");
        sb.append("|> sort(columns: [\"_time\"]) ");
        sb.append("|> yield(name: \"mean\")");
        log.info("influxdbSql===============" + sb);
        List<FluxTable> tables = queryApi.query(sb.toString(), influxDBInstance.org);

        for (FluxTable table : tables) {
            List<FluxRecord> records = table.getRecords();
            for (FluxRecord record : records) {
                InfluxModelResultVO vo = new InfluxModelResultVO(record.getValueByKey("_value"),record.getTime());
                dataList.add(vo);
            }
        }
        return dataList;
    }

    @Override
    public Map<String, List<InfluxModelResultVO>> queryModelResultsByOutPutIds(InfluxModelResultByOutPutIdsPOJO pojo, Date startTime, Date endTime) {
        Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.MILLISECOND, 0);
        String start = startTime.toInstant().toString();

        if (startTime.getTime() == endTime.getTime()) {
            // 如果相等，则engTime加1毫秒，否则influxdb报错(因为influxdb的range函数是左闭右开区间，所以将engTime加一毫秒，才可以查到startTime时间点的数据)
            endTime.setTime(endTime.getTime() + 1);
        }
        String stop = endTime.toInstant().toString();

        String measurement = MeasurementUtils.getMeasurement(pojo.getType());
        // 拼接OutPutIds
        String outPutIdsFilter = pojo.getOutPutIds().stream().map(id -> "r[\"outPutId\"] == \"" + id + "\"").collect(Collectors.joining(" or "));
        StringBuilder sb = new StringBuilder();
        sb.append("from(bucket:\"" + influxDBInstance.getBucket() + "\") ");
        sb.append("|> range(start: ").append(start).append(", stop: ").append(stop).append(") ");
        sb.append("|> filter(fn: (r) => r[\"_measurement\"] == \"" + measurement + "\")");
        sb.append("|> filter(fn: (r) => r[\"_field\"] == \"value\")");
        sb.append("|> filter(fn: (r) => " + outPutIdsFilter + ")");
        sb.append("|> sort(columns: [\"_time\"]) ");
        sb.append("|> yield(name: \"mean\")");
        log.info("influxdbSql===============" + sb);
        List<FluxTable> tables = queryApi.query(sb.toString(), influxDBInstance.org);

        Map<String, List<InfluxModelResultVO>> result = new HashMap<>(pojo.getOutPutIds().size());
        for (FluxTable table : tables) {
            List<FluxRecord> records = table.getRecords();
            for (FluxRecord record : records) {
                String outPutId = record.getValueByKey("outPutId").toString();
                if (result.containsKey(outPutId)) {
                    result.get(outPutId).add( new InfluxModelResultVO(record.getValueByKey("_value"),record.getTime()));
                } else {
                    List<InfluxModelResultVO> dataList = new ArrayList<>();
                    InfluxModelResultVO vo = new InfluxModelResultVO(record.getValueByKey("_value"),record.getTime());
                    dataList.add(vo);
                    result.put(outPutId,dataList);
                }
            }
        }
        return result;
    }
}