package com.hw.ems.record.service.impl;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hw.common.core.constant.SecurityConstants;
import com.hw.common.core.constant.TdEngineConstants;
import com.hw.common.core.domain.R;
import com.hw.common.core.utils.DateUtils;
import com.hw.common.core.utils.StringUtils;
import com.hw.ems.base.domain.*;
import com.hw.ems.base.mapper.BaseCollectDeviceInfoMapper;
import com.hw.ems.base.mapper.BaseDnbThresholdMapper;
import com.hw.ems.base.mapper.BaseWaterThresholdMapper;
import com.hw.ems.base.service.IBaseMonitorInfo1Service;
import com.hw.ems.base.service.ISysDatasourceService;
import com.hw.ems.record.domain.RecordAlarmData;
import com.hw.ems.record.domain.RecordAlarmRule;
import com.hw.ems.record.domain.RecordDnbInstant;
import com.hw.ems.record.mapper.RecordAlarmRuleMapper;
import com.hw.ems.record.mapper.RecordDnbInstantMapper;
import com.hw.ems.record.mapper.RecordWaterInstantMapper;
import com.hw.ems.record.service.IRecordAlarmDataService;
import com.hw.ems.report.service.IDataAnalysisService;
import com.hw.tdengine.api.RemoteTdEngineService;
import com.hw.tdengine.api.domain.TdField;
import com.hw.tdengine.api.domain.TdSelectDto;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.hw.ems.record.mapper.RecordAlarmDataMapper;


/**
 * 异常数据记录Service业务层处理
 *
 * @author sf
 * @date 2023-04-21
 */

@Service("recordAlarmDataService")
public class RecordAlarmDataServiceImpl implements IRecordAlarmDataService {
    protected Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private RecordAlarmDataMapper recordAlarmDataMapper;

    @Autowired
    private RecordDnbInstantMapper recordDnbInstantMapper;

    @Autowired
    private RecordWaterInstantMapper recordWaterInstantMapper;

    @Autowired
    private BaseDnbThresholdMapper baseDnbThresholdMapper;

    @Autowired
    private BaseWaterThresholdMapper baseWaterThresholdMapper;

    @Autowired
    private BaseCollectDeviceInfoMapper baseCollectDeviceInfoMapper;

    @Autowired
    private RecordAlarmRuleMapper recordAlarmRuleMapper;

    @Autowired
    private ISysDatasourceService datasourceService;

    @Autowired
    private IDataAnalysisService dataAnalysisService;
    @Autowired
    private RemoteTdEngineService remoteTdEngineService;
    @Autowired
    private IBaseMonitorInfo1Service baseMonitorInfo1Service;
    /**
     * 查询异常数据记录
     *
     * @param objId 异常数据记录主键
     * @return 异常数据记录
     */
    @Override
    //@DS("#header.poolName")
    public RecordAlarmData selectRecordAlarmDataByObjId(Long objId) {
        return recordAlarmDataMapper.selectRecordAlarmDataByObjId(objId);
    }

    /**
     * 查询异常数据记录列表
     *
     * @param recordAlarmData 异常数据记录
     * @return 异常数据记录
     */
    @Override
    //@DS("#header.poolName")
    public List<RecordAlarmData> selectRecordAlarmDataList(RecordAlarmData recordAlarmData) {
        return recordAlarmDataMapper.selectRecordAlarmDataList(recordAlarmData);
    }

    /**
     * 新增异常数据记录
     *
     * @param recordAlarmData 异常数据记录
     * @return 结果
     */
    @Override
    //@DS("#header.poolName")
    public int insertRecordAlarmData(RecordAlarmData recordAlarmData) {
        recordAlarmData.setCreateTime(DateUtils.getNowDate());
        return recordAlarmDataMapper.insertRecordAlarmData(recordAlarmData);
    }

    /**
     * 修改异常数据记录
     *
     * @param recordAlarmData 异常数据记录
     * @return 结果
     */
    @Override
    //@DS("#header.poolName")
    public int updateRecordAlarmData(RecordAlarmData recordAlarmData) {
        recordAlarmData.setUpdateTime(DateUtils.getNowDate());
        return recordAlarmDataMapper.updateRecordAlarmData(recordAlarmData);
    }

    /**
     * 批量删除异常数据记录
     *
     * @param objIds 需要删除的异常数据记录主键
     * @return 结果
     */
    @Override
    //@DS("#header.poolName")
    public int deleteRecordAlarmDataByObjIds(Long[] objIds) {
        return recordAlarmDataMapper.deleteRecordAlarmDataByObjIds(objIds);
    }

    /**
     * 删除异常数据记录信息
     *
     * @param objId 异常数据记录主键
     * @return 结果
     */
    @Override
    //@DS("#header.poolName")
    public int deleteRecordAlarmDataByObjId(Long objId) {
        return recordAlarmDataMapper.deleteRecordAlarmDataByObjId(objId);
    }

    /**
     * 超过电阈值 对比记录
     * @param baseDnbThresholds
     * @param recordDnbInstants
     * @return
     */
    public Map<String, String> compareThresholdAndRecord(List<BaseDnbThreshold> baseDnbThresholds, List<RecordDnbInstant> recordDnbInstants) {
        Map<String, String> resultMap = new HashMap<>();

        for (RecordDnbInstant recordDnbInstant : recordDnbInstants) {
            String monitorId = recordDnbInstant.getMonitorId();
            BaseDnbThreshold baseDnbThreshold = getThresholdByMonitorId(monitorId, baseDnbThresholds);
            if (baseDnbThreshold != null) {
                String reason = compare(recordDnbInstant, baseDnbThreshold);
                if (reason != null) {
                    resultMap.put(monitorId, reason);
                }
            }
        }

        return resultMap;
    }

    private BaseDnbThreshold getThresholdByMonitorId(String monitorId, List<BaseDnbThreshold> baseDnbThresholds) {
        for (BaseDnbThreshold baseDnbThreshold : baseDnbThresholds) {
            if (baseDnbThreshold.getMonitorId().equals(monitorId)) {
                return baseDnbThreshold;
            }
        }
        return null;
    }

    private String compare(RecordDnbInstant recordDnbInstant, BaseDnbThreshold baseDnbThreshold) {
        if (StringUtils.isNull(recordDnbInstant.getIA())){
            return null;
        }
        BigDecimal zero = new BigDecimal("0.00");
        if (recordDnbInstant.getVA().compareTo(zero) == 0 || recordDnbInstant.getVB().compareTo(zero) == 0
                || recordDnbInstant.getVC().compareTo(zero) == 0) {
            return "缺项报警";
        }
        if (recordDnbInstant.getIA().compareTo(baseDnbThreshold.getiAMax()) > 0) {
            return "超过A相电流最大值";
        }
        if (recordDnbInstant.getIB().compareTo(baseDnbThreshold.getiBMax()) > 0) {
            return "超过B相电流最大值";
        }
        if (recordDnbInstant.getIC().compareTo(baseDnbThreshold.getiCMax()) > 0) {
            return "超过C相电流最大值";
        }
        if (recordDnbInstant.getVA().compareTo(baseDnbThreshold.getvAMax()) > 0) {
            return "超过A相电压最大值";
        }
        if (recordDnbInstant.getVB().compareTo(baseDnbThreshold.getvBMax()) > 0) {
            return "超过B相电压最大值";
        }
        if (recordDnbInstant.getVC().compareTo(baseDnbThreshold.getvCMax()) > 0) {
            return "超过C相电压最大值";
        }
        if (recordDnbInstant.getIA().compareTo(baseDnbThreshold.getiAMin()) < 0) {
            return "小于A相电流最小值";
        }
        if (recordDnbInstant.getIB().compareTo(baseDnbThreshold.getiBMin()) < 0) {
            return "小于B相电流最小值";
        }
        if (recordDnbInstant.getIC().compareTo(baseDnbThreshold.getiCMin()) < 0) {
            return "小于C相电流最小值";
        }
        if (recordDnbInstant.getVA().compareTo(baseDnbThreshold.getvAMin()) < 0) {
            return "小于A相电压最小值";
        }
        if (recordDnbInstant.getVB().compareTo(baseDnbThreshold.getvBMin()) < 0) {
            return "小于B相电压最小值";
        }
        if (recordDnbInstant.getVC().compareTo(baseDnbThreshold.getvCMin()) < 0) {
            return "小于C相电压最小值";
        }
        return null;
    }


    /**
     * 超过阈值告警
     */
    @Override
    public R<Boolean>  thresholdTimingTask() {

//        // 加载sf-cloud库的sys_datasource
//        List<SysDatasource> dateSources = datasourceService.selectSysDatasource();
//
//        ExecutorService executorService = Executors.newFixedThreadPool(dateSources.size());
//        try{
//            dateSources.forEach(dateSource ->{
//                logger.info("++++++++++++"+dateSource.getPoolName() +"++++开始++++++++++");
//                Runnable run = () -> thresholdTimingTaskDetail(dateSource.getPoolName());
//                executorService.execute(run);
//            });
//        }catch (Exception e){
//            logger.error("service == thresholdTimingTask == exception", e);
//            return R.fail("service == thresholdTimingTask == exception");
//        }finally {
//            executorService.shutdown();
//        }

        this.thresholdTimingTaskDetail("1");
        return R.ok(true);
    }

    public void thresholdTimingTaskDetail(String poolName){
//        DynamicDataSourceContextHolder.push(poolName);// 这是数据源的key
        //电阈值
        BaseDnbThreshold baseDnbThreshold = new BaseDnbThreshold();
        baseDnbThreshold.setThresholdStatus(0L);
        List<BaseDnbThreshold> baseDnbThresholds = baseDnbThresholdMapper.selectBaseDnbThresholdList(baseDnbThreshold);
        //查询最新的电实时数据对象
        //todo 更改为通过tde查询电实时数据对象
//        List<RecordDnbInstant> recordDnbInstants = recordDnbInstantMapper.instantList(new RecordDnbInstant());
        List<RecordDnbInstant> recordDnbInstants = this.getDnbInstantByTde();
        //通过电实时数据与阈值信息对象做对比
        if (recordDnbInstants.size() == 0){
            return;
        }
        //todo 获取到数据后与报警值对比
        Map<String, String> thresholdMap = compareThresholdAndRecord(baseDnbThresholds, recordDnbInstants);
        //防止多次存入异常数据
        RecordAlarmData recordAlarmData = new RecordAlarmData();
        recordAlarmData.setAlarmType(0L);
        recordAlarmData.setAlarmStatus(1L);
        List<RecordAlarmData> alarmDataList = recordAlarmDataMapper.selectRecordAlarmDataList(recordAlarmData);
        List<String> monitorIdList = alarmDataList.stream().map(RecordAlarmData::getMonitorId).collect(Collectors.toList());
        for (String monitorId : thresholdMap.keySet()) {
            if (monitorIdList.contains(monitorId)) {
                continue;
            }
            RecordAlarmData alarmData = new RecordAlarmData();
            alarmData.setMonitorId(monitorId);
            alarmData.setCollectTime(new Date());
            alarmData.setAlarmType(0L);
            alarmData.setAlarmStatus(1L);
            alarmData.setAlarmData(thresholdMap.get(monitorId));
            recordAlarmDataMapper.insertRecordAlarmData(alarmData);
        }

//        水 瞬时报警
//        try {
//            List<RecordWaterInstant> recordWaterInstants = recordWaterInstantMapper.fluxFlowAlarm(new RecordWaterInstant());
//            if (recordWaterInstants.size() > 0){
//                for (RecordWaterInstant waterInstant : recordWaterInstants) {
//                    String monitorId = waterInstant.getMonitorId();
//                    RecordAlarmData alarmData = new RecordAlarmData();
//                    alarmData.setMonitorId(monitorId);
//                    alarmData.setCollectTime(new Date());
//                    alarmData.setAlarmType(2L);
//                    alarmData.setAlarmStatus(1L);
//                    alarmData.setAlarmData("瞬时流量异常");
//                    recordAlarmDataMapper.insertRecordAlarmData(alarmData);
//                }
//            }
//        } catch (Exception e) {
//            logger.error("service == 水瞬时报警 == exception", e);
//        }
    }
    /**
     * 通过tde获取实时数据
     * */
    public List<RecordDnbInstant> getDnbInstantByTde(){
        List<RecordDnbInstant> list = new ArrayList<>();
        List<TdField> listTdfields = new ArrayList<>();
         TdField tdField1 = new TdField();
         TdField tdField2 = new TdField();
         TdField tdField3 = new TdField();
         TdField tdField4 = new TdField();
         TdField tdField5 = new TdField();
         TdField tdField6 = new TdField();
         TdField tdField7 = new TdField();
         TdField tdField8 = new TdField();
         TdField tdField9 = new TdField();
         TdField tdField10 = new TdField();
         TdField tdField11 = new TdField();
         TdField tdField12 = new TdField();
         tdField1.setFieldName("volatagea");
         tdField2.setFieldName("volatageb");
         tdField3.setFieldName("volatagec");
         tdField4.setFieldName("currenta");
         tdField5.setFieldName("currentb");
         tdField6.setFieldName("currentc");
         tdField7.setFieldName("glys");
         tdField8.setFieldName("zxyg");
         tdField9.setFieldName("activepower");
         tdField10.setFieldName("reactivepower");
         tdField11.setFieldName("consumption");
         tdField12.setFieldName("collecttype");
        listTdfields.add(tdField1);
        listTdfields.add(tdField2);
        listTdfields.add(tdField3);
        listTdfields.add(tdField4);
        listTdfields.add(tdField5);
        listTdfields.add(tdField6);
        listTdfields.add(tdField7);
        listTdfields.add(tdField8);
        listTdfields.add(tdField9);
        listTdfields.add(tdField10);
        listTdfields.add(tdField11);
        listTdfields.add(tdField12);
        TdSelectDto tdSelectDto = new TdSelectDto();
        tdSelectDto.setDatabaseName(TdEngineConstants.EMS_DATABASE_NAME);
        tdSelectDto.setSuperTableName("st_electricity");
        tdSelectDto.setSchemaFieldValues(listTdfields);
        List<BaseMonitorInfo> baseMonitorInfos = baseMonitorInfo1Service.selectBaseMonitorInfoList(new BaseMonitorInfo());
        for (BaseMonitorInfo baseMonitorInfo : baseMonitorInfos){
            tdSelectDto.setTableName(TdEngineConstants.getEmsTableName(baseMonitorInfo.getObjid()));
//            tdSelectDto.setTableName(TdEngineConstants.getEmsTableName("ceshi7"));
            tdSelectDto.setTagsName("monitorobjid");
            //查出每个计量设备最新的一条实时数据

            List<Map<String, Object>> data = (ArrayList)remoteTdEngineService.getLatestData(tdSelectDto, SecurityConstants.INNER).getData();

            //转换成对象
            //因为tde中字段名与对象字段名不同，只能手动转换
            RecordDnbInstant recordDnbInstant = new RecordDnbInstant();
            recordDnbInstant.setMonitorId(baseMonitorInfo.getMonitorId());
            recordDnbInstant.setMonitorName(baseMonitorInfo.getMonitorName());
            recordDnbInstant.setObjid(baseMonitorInfo.getObjid());
            recordDnbInstant.setVA(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("volatagea")))));
            recordDnbInstant.setVB(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("volatageb")))));
            recordDnbInstant.setVC(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("volatagec")))));
            recordDnbInstant.setIA(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("currenta")))));
            recordDnbInstant.setIB(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("currentb")))));
            recordDnbInstant.setIC(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("currentc")))));
            recordDnbInstant.setGlys(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("glys")))));
            recordDnbInstant.setZxyg(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("zxyg")))));
            recordDnbInstant.setActivePower(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("activepower")))));
            recordDnbInstant.setReactivePower(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("reactivepower")))));
//            recordDnbInstant.setConsumption(BigDecimal.valueOf(Double.valueOf(String.valueOf(data.get(0).get("consuption")))));
            recordDnbInstant.setCollectType(Long.valueOf(String.valueOf(data.get(0).get("collecttype"))));
                    list.add(recordDnbInstant);
        }

        return list;
    }

    /**
     * 设备离线告警(定时任务)
     */
    @Override
    public R<Boolean> deviceOfflineTimingTask() {
//        // 加载sf-cloud库的sys_datasource
//        List<SysDatasource> dateSources = datasourceService.selectSysDatasource();
//
//        ExecutorService executorService = Executors.newFixedThreadPool(dateSources.size());
//        try{
//            dateSources.forEach(dateSource ->{
//                logger.info("++++++++++++"+dateSource.getPoolName() +"++++开始++++++++++");
//                Runnable run = () -> deviceOfflineTimingTaskDetail(dateSource.getPoolName());
//                executorService.execute(run);
//            });
//        }catch (Exception e){
//            logger.error("service == deviceOfflineTimingTask == exception", e);
//            return R.fail("service == deviceOfflineTimingTask == exception");
//        }finally {
//            executorService.shutdown();
//        }
        this.deviceOfflineTimingTaskDetail("1");
        return R.ok(true);
    }

    public void deviceOfflineTimingTaskDetail(String poolName) {
//        DynamicDataSourceContextHolder.push(poolName);// 这是数据源的key
        Date date = new Date();
        RecordAlarmRule alarmRule = new RecordAlarmRule();
        alarmRule.setTriggerRule(1L);
        List<RecordAlarmRule> recordAlarmRules = recordAlarmRuleMapper.selectRecordAlarmRuleList(alarmRule);
        Long minuteValue = 1000 * 60L;
        if (recordAlarmRules.size() > 0) {
            if (recordAlarmRules.get(0).getDeviceOfflineTime() != null) {
                minuteValue = minuteValue * recordAlarmRules.get(0).getDeviceOfflineTime();
            }
        } else {
            return;
        }
        BaseCollectDeviceInfo collectDeviceInfo = new BaseCollectDeviceInfo();
        collectDeviceInfo.setOnlineState(0L);
        //查询终端在线状态为0的采集设备信息
        List<BaseCollectDeviceInfo> baseCollectDeviceInfos = baseCollectDeviceInfoMapper.selectBaseCollectDeviceInfoList(collectDeviceInfo);
        if (baseCollectDeviceInfos.size() > 0) {
            RecordAlarmData recordAlarmData = new RecordAlarmData();
            recordAlarmData.setAlarmType(1L);
            recordAlarmData.setAlarmStatus(1L);
            List<RecordAlarmData> alarmDataList = recordAlarmDataMapper.selectRecordAlarmDataList(recordAlarmData);
            List<String> collectDeviceIdList = alarmDataList.stream().map(RecordAlarmData::getCollectDeviceId).collect(Collectors.toList());
            for (BaseCollectDeviceInfo deviceInfo : baseCollectDeviceInfos) {
                if (collectDeviceIdList.contains(deviceInfo.getCollectDeviceId())) {
                    continue;
                }
                if ((deviceInfo.getUpdateTime().getTime() + minuteValue) < date.getTime()) {
                    RecordAlarmData alarmData = new RecordAlarmData();
                    alarmData.setCollectDeviceId(deviceInfo.getCollectDeviceId());
                    alarmData.setCollectTime(new Date());
                    alarmData.setAlarmType(1L);
                    alarmData.setAlarmStatus(1L);
                    alarmData.setAlarmData("设备离线已超过" + recordAlarmRules.get(0).getDeviceOfflineTime() + "分钟");
                    recordAlarmDataMapper.insertRecordAlarmData(alarmData);
                }
            }
        }
    }

    /**
     * 小时线损告警
     * @return
     */
    @Override
    public R<Boolean> hourLineLossTimingTask() {
        // 加载sf-cloud库的sys_datasource
        List<SysDatasource> dateSources = datasourceService.selectSysDatasource();

        ExecutorService executorService = Executors.newFixedThreadPool(dateSources.size());
        try{
            dateSources.forEach(dateSource ->{
                logger.info("++++++++++++"+dateSource.getPoolName() +"++++开始++++++++++");
                Runnable run = () -> hourLineLossTimingTask(dateSource.getPoolName());
                executorService.execute(run);
            });
        }catch (Exception e){
            logger.error("service == thresholdTimingTask == exception", e);
            return R.fail("service == thresholdTimingTask == exception");
        }finally {
            executorService.shutdown();
        }
        return R.ok(true);
    }

    /**
     * 水电 线损告警
     * @param poolName
     */
    public void hourLineLossTimingTask(String poolName){
        DynamicDataSourceContextHolder.push(poolName);// 这是数据源的key
        BaseLineLoss baseLineLoss = new BaseLineLoss();
        Date nowDate = DateUtils.getNowDate();
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(nowDate);
        calendar.add(Calendar.HOUR_OF_DAY, -1);
        Date hourAgoTime = calendar.getTime();
        Map<String, Object>  lineLossMap = new HashMap<>();
        lineLossMap.put("beginCollectTime", DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss", hourAgoTime));
        lineLossMap.put("endCollectTime", DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss", nowDate));
        baseLineLoss.setParams(lineLossMap);
        //电告警
        baseLineLoss.setMonitorType(2L);
        List<Map<String, String>> lineLossList = dataAnalysisService.lineLossAnalysisDataTimedTasks(baseLineLoss);
        //电阈值
        BaseDnbThreshold baseDnbThreshold = new BaseDnbThreshold();
        baseDnbThreshold.setThresholdStatus(0L);
        List<BaseDnbThreshold> baseDnbThresholds = baseDnbThresholdMapper.selectBaseDnbThresholdList(baseDnbThreshold);
        if (lineLossList.size() > 0){
            for (Map<String, String> map : lineLossList) {
                String monitorId = String.valueOf(map.get("monitorId"));
                for (BaseDnbThreshold dnbThreshold : baseDnbThresholds) {
                    if (monitorId.equals(dnbThreshold.getMonitorId())){
                        BigDecimal lineLossRate = dnbThreshold.getLineLossRate();
                        if (StringUtils.isNull(lineLossRate)){
                            continue;
                        }
                        String substring = String.valueOf(map.get("lossRate"));
                        BigDecimal lossRate = new BigDecimal(substring);
                        if (lineLossRate.compareTo(lossRate) < 0){
                            RecordAlarmData alarmData = new RecordAlarmData();
                            alarmData.setMonitorId(monitorId);
                            alarmData.setCollectTime(new Date());
                            alarmData.setAlarmType(3L);
                            alarmData.setAlarmStatus(1L);
                            alarmData.setAlarmData("超过线损电阈值");
                            recordAlarmDataMapper.insertRecordAlarmData(alarmData);
                        }
                    }
                }
            }
        }


        //水线损
        baseLineLoss.setMonitorType(3L);
        List<Map<String, String>> waterLossList = dataAnalysisService.lineLossAnalysisDataTimedTasks(baseLineLoss);
        //水阈值
        BaseWaterThreshold baseWaterThreshold = new BaseWaterThreshold();
        baseWaterThreshold.setThresholdStatus(0L);
        List<BaseWaterThreshold> baseWaterThresholds = baseWaterThresholdMapper.selectBaseWaterThresholdList(baseWaterThreshold);
        if (waterLossList.size() > 0){
            for (Map<String, String> map : waterLossList) {
                String monitorId = String.valueOf(map.get("monitorId"));
                for (BaseWaterThreshold waterThreshold : baseWaterThresholds) {
                    if (monitorId.equals(waterThreshold.getMonitorId())){
                        BigDecimal lineLossRate = waterThreshold.getLineLossRate();
                        if (StringUtils.isNull(lineLossRate)){
                            continue;
                        }
                        String substring = String.valueOf(map.get("lossRate"));
                        BigDecimal lossRate = new BigDecimal(substring);
                        if (lineLossRate.compareTo(lossRate) < 0){
                            RecordAlarmData alarmData = new RecordAlarmData();
                            alarmData.setMonitorId(monitorId);
                            alarmData.setCollectTime(new Date());
                            alarmData.setAlarmType(3L);
                            alarmData.setAlarmStatus(1L);
                            alarmData.setAlarmData("超过线损水阈值");
                            recordAlarmDataMapper.insertRecordAlarmData(alarmData);
                        }
                    }
                }
            }
        }
    }

    /**
     * 日耗量告警
     * @return
     */
    @Override
    public R<Boolean> dailyConsumptionTimingTask() {
        // 加载sf-cloud库的sys_datasource
        List<SysDatasource> dateSources = datasourceService.selectSysDatasource();

        ExecutorService executorService = Executors.newFixedThreadPool(dateSources.size());
        try{
            dateSources.forEach(dateSource ->{
                logger.info("++++++++++++"+dateSource.getPoolName() +"++++开始++++++++++");
                Runnable run = () -> dailyConsumptionTimingTask(dateSource.getPoolName());
                executorService.execute(run);
            });
        }catch (Exception e){
            logger.error("service == dailyConsumptionTimingTask == exception", e);
            return R.fail("service == dailyConsumptionTimingTask == exception");
        }finally {
            executorService.shutdown();
        }
        return R.ok(true);
    }


    /**
     * 水电日耗量
     * @param poolName
     */
    public void dailyConsumptionTimingTask(String poolName){
        DynamicDataSourceContextHolder.push(poolName);// 这是数据源的key
        HashMap<String, Object> map = new HashMap<>();
        map.put("timeSub", null);
        Date today = new Date(System.currentTimeMillis() - 1000 * 60 * 60 * 24);
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
        String yesterday = simpleDateFormat.format(today);//获取昨天日期
        map.put("startDay", yesterday);
        map.put("endDay", yesterday);

        //电阈值
        BaseDnbThreshold baseDnbThreshold = new BaseDnbThreshold();
        baseDnbThreshold.setThresholdStatus(0L);
        List<BaseDnbThreshold> baseDnbThresholds = baseDnbThresholdMapper.selectBaseDnbThresholdList(baseDnbThreshold);
        if (baseDnbThresholds.size() > 0){
            for (BaseDnbThreshold dnbThreshold : baseDnbThresholds) {
                if (StringUtils.isNull(dnbThreshold.getDayConsumption())){
                    continue;
                }
                map.put("energyType", 2L);
                map.put("monitorId", dnbThreshold.getMonitorId());
                List<Map<String, String>> dnbConsumptionList = dataAnalysisService.energyConsumptionByTime(map);
                if (dnbConsumptionList.size() == 0){
                    continue;
                }
                BigDecimal expend = new BigDecimal(String.valueOf(dnbConsumptionList.get(0).get("expend")));
                if (expend.compareTo(dnbThreshold.getDayConsumption()) > 0){
                    RecordAlarmData alarmData = new RecordAlarmData();
                    alarmData.setMonitorId(dnbThreshold.getMonitorId());
                    alarmData.setCollectTime(new Date());
                    alarmData.setAlarmType(4L);
                    alarmData.setAlarmStatus(1L);
                    alarmData.setAlarmData("超过日耗量电阈值");
                    //插入异常数据记录
                    recordAlarmDataMapper.insertRecordAlarmData(alarmData);
                }
            }
        }
        //水阈值
        BaseWaterThreshold baseWaterThreshold = new BaseWaterThreshold();
        baseDnbThreshold.setThresholdStatus(0L);
        List<BaseWaterThreshold> baseWaterThresholds = baseWaterThresholdMapper.selectBaseWaterThresholdList(baseWaterThreshold);
        if (baseWaterThresholds.size() > 0){
            for (BaseWaterThreshold waterThreshold : baseWaterThresholds) {
                if (StringUtils.isNull(waterThreshold.getDayConsumption())){
                    continue;
                }
                map.put("energyType", 3L);
                map.put("monitorId", waterThreshold.getMonitorId());
                List<Map<String, String>> dnbConsumptionList = dataAnalysisService.energyConsumptionByTime(map);
                if (dnbConsumptionList.size() == 0){
                    continue;
                }
                BigDecimal expend = new BigDecimal(String.valueOf(dnbConsumptionList.get(0).get("expend")));
                if (expend.compareTo(waterThreshold.getDayConsumption()) > 0){
                    RecordAlarmData alarmData = new RecordAlarmData();
                    alarmData.setMonitorId(waterThreshold.getMonitorId());
                    alarmData.setCollectTime(new Date());
                    alarmData.setAlarmType(4L);
                    alarmData.setAlarmStatus(1L);
                    alarmData.setAlarmData("超过日耗量水阈值");
                    recordAlarmDataMapper.insertRecordAlarmData(alarmData);
                }
            }
        }

    }




}
