package com.gyee.wisdom.dao.taoscz;

import com.gyee.wisdom.common.data.timeseries.*;
import com.gyee.wisdom.common.exception.WisdomException;
import com.gyee.wisdom.dao.timeseries.IHistoryDao;
import com.gyee.wisdom.dao.timeseries.TaosczDao;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.InputStream;
import java.sql.Blob;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

@Slf4j
@Component
@TaosczDao
public class TaosHistoryDao implements IHistoryDao {

    private long year_time = 8640000000L;

    @Autowired
    TaosConfig taosConfig;
    @Autowired
    ThreadPoolTaskConfig taskConfig;

    @Override
    public List<TsData> getTsDataHistory(TsQuery tsQuery) throws Exception {
        if (tsQuery.getTsPoint().getTsDataType() == TsDataType.DOUBLE) {
            return getDoubleTsDataHistory(tsQuery);
        } else if (tsQuery.getTsPoint().getTsDataType() == TsDataType.BOOLEAN) {
            return getBooleanTsDataHistory(tsQuery);
        } else if (tsQuery.getTsPoint().getTsDataType() == TsDataType.LONG) {
            return getLongTsDataHistory(tsQuery);
        } else if (tsQuery.getTsPoint().getTsDataType() == TsDataType.BLOB) {
            return getBlobTsDataHistory(tsQuery);
        } else if (tsQuery.getTsPoint().getTsDataType() == TsDataType.STRING) {
            return getStringTsDataHistory(tsQuery);
        }
        return null;
    }

    @Override
    public List<DoubleStatData> getDoubleStatDataHistory(TsQuery tsQuery) throws Exception {
        if (tsQuery.getTsPoint().getTsDataType() != TsDataType.DOUBLE) {
            throw new WisdomException("无效的数据类型：" + tsQuery.getTsPoint().getTsDataType());
        }

        List<DoubleStatData> result = new ArrayList<>();

        Statement st = taosConfig.getInstance().createStatement();
        String point = tsQuery.getTsPoint().getId();
        StringBuilder sb = new StringBuilder();
        sb.append("select avg(val),max(val),min(val) from ");
        sb.append(taosConfig.getTableName(tsQuery.getTsPoint().getId()));
        sb.append(" where ts between ").append(tsQuery.getStartTs());
        sb.append(" and ").append(tsQuery.getEndTs());
        //sb.append(" interval(").append(tsQuery.getInterval()).append("s)");
        ResultSet rs = st.executeQuery(sb.toString());
        while (rs.next()) {
            DoubleTsData avgData = new DoubleTsData(tsQuery.getStartTs(), (short) 1, rs.getDouble(1));
            DoubleTsData maxData = new DoubleTsData(tsQuery.getStartTs(), (short) 1, rs.getDouble(2));
            DoubleTsData minData = new DoubleTsData(tsQuery.getStartTs(), (short) 1, rs.getDouble(3));
            DoubleStatData statData = new DoubleStatData(avgData, maxData, minData);
            result.add(statData);
        }

        return result;
    }

    @Override
    public boolean writeHistoryValue(List<TsPointData> dataList) throws Exception {
        Statement st = taosConfig.getInstance().createStatement();
        StringBuilder sb = new StringBuilder();
        sb.append("insert into ");
        for (TsPointData obj : dataList) {
            long time = obj.getTsData().getTs();
            String point = obj.getTagName();
            double value = obj.getTsData().getDoubleValue().get();
            // sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
            sb.append(point).append(" values (");
            sb.append(time).append(",").append(value).append(")");
        }
        boolean flag = st.execute(sb.toString());

        return flag;
    }

    @Override
    public Map<String, TsData> getHistorySection(List<TsPoint> tsPoints, Long ts) throws Exception {
        Map<String, TsData> result = new HashMap<>();
        //存储线程的返回值
        List<Future<Map<String, TsData>>> results = new LinkedList<>();

        Map<TsDataType, List<TsPoint>> pointGroup = tsPoints.stream().collect(Collectors.groupingBy(TsPoint::getTsDataType));
        for (Map.Entry<TsDataType, List<TsPoint>> entry : pointGroup.entrySet()) {
            String[] tagNames = entry.getValue().stream().map(TsPoint::getId).toArray(String[]::new);
            if (entry.getKey() == TsDataType.DOUBLE)
                for (String tag : tagNames) {
                    TaskCallable task = new TaskCallable(taosConfig, ts, tag, TsDataType.DOUBLE);
                    Future<Map<String, TsData>> submit = taskConfig.getInstance().submit(task);
                    results.add(submit);
                }
            if (entry.getKey() == TsDataType.BOOLEAN) {
                for (String tag : tagNames) {
                    TaskCallable task = new TaskCallable(taosConfig, ts, tag, TsDataType.BOOLEAN);
                    Future<Map<String, TsData>> submit = taskConfig.getInstance().submit(task);
                    results.add(submit);
                }
            }
        }
        //返回结果
        for (int i = 0; i < results.size(); i++) {
            Map<String, TsData> map = results.get(i).get();
            result.putAll(map);
        }
//        for (Map.Entry<TsDataType, List<TsPoint>> entry : pointGroup.entrySet()) {
//            String[] tagNames = tsPoints.stream().map(TsPoint::getId).toArray(String[]::new);
//            for (String tag : tagNames) {
//                TsData tsData = null;
//                String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
//                String sql = "select last_row(*) from " + point + " where point_time>='"
//                        + DateFormatUtils.format(ts - year_time, "yyyy-MM-dd HH:mm:ss:SSS") + "'"
//                        + " and point_time <='" + DateFormatUtils.format(ts, "yyyy-MM-dd HH:mm:ss:SSS") + "'";
//                ResultSet rs = taosConfig.getInstance().createStatement().executeQuery(sql);
//                while (rs.next()) {
//                    if (entry.getKey() == TsDataType.DOUBLE)
//                        tsData = new DoubleTsData(ts, (short) 0, rs.getDouble(2));
//                    if (entry.getKey() == TsDataType.BOOLEAN)
//                        tsData = new BooleanTsData(ts, (short) 0, rs.getBoolean(2));
//                    result.put(tag, tsData);
//                }
//            }
//        }

        return result;
    }

    public List<TsData> getDoubleTsDataHistory(TsQuery tsQuery) throws Exception {
        List<TsData> tsDataList = new ArrayList<>();

        Statement st = taosConfig.getInstance().createStatement();
        String sql = getHistory(tsQuery);
        ResultSet rs = st.executeQuery(sql);
        while (rs.next()) {
            tsDataList.add(new DoubleTsData(rs.getTimestamp(1).getTime(), (short) 0, rs.getDouble(2)));
        }

        return tsDataList;
    }


    public List<TsData> getLongTsDataHistory(TsQuery tsQuery) throws Exception {
        List<TsData> tsDataList = new ArrayList<>();

        Statement st = taosConfig.getInstance().createStatement();
        String sql = getHistory(tsQuery);
        ResultSet rs = st.executeQuery(sql);
        while (rs.next()) {
            tsDataList.add(new LongTsData(rs.getTimestamp(1).getTime(), (short) 0, rs.getLong(2)));
        }

        return tsDataList;
    }

    public List<TsData> getBooleanTsDataHistory(TsQuery tsQuery) throws Exception {
        List<TsData> tsDataList = new ArrayList<>();

        Statement st = taosConfig.getInstance().createStatement();
        String sql = getHistory(tsQuery);
        ResultSet rs = st.executeQuery(sql);
        while (rs.next()) {
            tsDataList.add(new BooleanTsData(rs.getTimestamp(1).getTime(), (short) 0, rs.getBoolean(2)));
        }

        return tsDataList;
    }

    public List<TsData> getBlobTsDataHistory(TsQuery tsQuery) throws Exception {
        List<TsData> tsDataList = new ArrayList<>();

        Statement st = taosConfig.getInstance().createStatement();
        String sql = getHistory(tsQuery);
        ResultSet rs = st.executeQuery(sql);
        while (rs.next()) {
            tsDataList.add(new BlobTsData(rs.getTimestamp(1).getTime(), (short) 0, getBytes(rs.getBlob(2))));
        }

        return tsDataList;
    }

    public List<TsData> getStringTsDataHistory(TsQuery tsQuery) throws Exception {
        List<TsData> tsDataList = new ArrayList<>();

        Statement st = taosConfig.getInstance().createStatement();
        String sql = getHistory(tsQuery);
        ResultSet rs = st.executeQuery(sql);
        while (rs.next()) {
            tsDataList.add(new StringTsData(rs.getLong(1), (short) 0, rs.getString(2)));
        }

        return tsDataList;
    }


    /**
     * 拼接sql
     *
     * @param tsQuery
     * @return
     * @throws WisdomException
     */
    private String getHistory(TsQuery tsQuery) throws WisdomException {
        String point = tsQuery.getTsPoint().getId();
        StringBuilder sb = new StringBuilder();
        if (tsQuery.getInterpolation() == Interpolation.RAW) {
            sb.append("select * from ");
            sb.append(taosConfig.getTableName(point));
            sb.append(" where ts between ").append(tsQuery.getStartTs());
            sb.append(" and ").append(tsQuery.getEndTs());
        } else if (tsQuery.getInterpolation() == Interpolation.SNAP) {
            if (tsQuery.getDateArray() != null && tsQuery.getDateArray().length > 0) {
                sb.append("select _irowts,interp(val) from ");
                sb.append(taosConfig.getTableName(point));
                sb.append(" range(").append(tsQuery.getStartTs());
                sb.append(",").append(tsQuery.getEndTs());
                sb.append(") every(").append(tsQuery.getInterval()).append("s) fill(prev)");
            } else {
                throw new WisdomException("无效的查询条件！");
            }
        } else {
            throw new WisdomException("Golden不支持历史数据插值方法：" + tsQuery.getInterpolation());
        }

        return sb.toString();
    }


    /**
     * 将blob转化为byte[]
     *
     * @param blob
     * @return
     */
    private byte[] getBytes(Blob blob) {
        try {
            InputStream ins = blob.getBinaryStream();
            byte[] b = new byte[1024];
            int num = 0;
            String res = "";
            while ((num = ins.read(b)) != -1) {
                res += new String(b, 0, num);
            }
            return res.getBytes();
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}
