package com.njcn.energy.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.util.CharUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.github.tocrhz.mqtt.annotation.MqttSubscribe;
import com.github.tocrhz.mqtt.annotation.Payload;
import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.njcn.common.pojo.enums.common.DataStateEnum;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.db.constant.DbConstant;
import com.njcn.energy.mapper.AirStrategyMapper;
import com.njcn.energy.mapper.LogicDeviceLineMapper;
import com.njcn.energy.pojo.dto.AirStrategyDTO;
import com.njcn.energy.pojo.dto.NetAndDevByLineDTO;
import com.njcn.energy.pojo.enums.EnergyResponseEnum;
import com.njcn.energy.pojo.param.AirStrategyParam;
import com.njcn.energy.pojo.po.AirSet;
import com.njcn.energy.pojo.po.AirStrategy;
import com.njcn.energy.pojo.vo.AirStrategyVO;
import com.njcn.energy.service.EleAirStrategyService;
import com.njcn.energy.service.IEleSetService;
import com.njcn.influx.utils.InfluxDbUtils;
//import com.njcn.job.api.JobFeignClient;
//import com.njcn.job.model.XxlJobInfo;
import com.njcn.web.factory.PageFactory;
import com.njcn.web.utils.RequestUtil;
//import com.xxl.job.core.biz.model.ReturnT;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.influxdb.dto.QueryResult;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * pqs
 *
 * @author cdf
 * @date 2022/4/21
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class EleAirStrategyServiceImpl extends ServiceImpl<AirStrategyMapper, AirStrategy> implements EleAirStrategyService {

    // 存储设备响应状态的共享Map（线程安全）
    private final Map<String, Boolean> deviceResponseMap = new ConcurrentHashMap<>();


    private final IEleSetService iEleSetService;

    //private final JobFeignClient jobFeignClient;

    private final LogicDeviceLineMapper logicDeviceLineMapper;

    private final MqttPublisher publisher;

    private final AirStrategyMapper airStrategyMapper;


    private final InfluxDbUtils influxDbUtils;

    @Value("${mqtt.client-id}")
    private String clientId;

    //徐扬执行器 5 陈道飞执行器 6   18数据库测试执行器 3
    private final Integer xxlGroup = 5;


    @Override
    public Page<AirStrategy> listAirStrategy(AirStrategyParam.AirStrategyQueryParam queryParam) {
        QueryWrapper<AirStrategy> queryWrapper = new QueryWrapper<>();
        if (Objects.nonNull(queryParam)) {
            //查询参数不为空，进行条件填充
            if (StrUtil.isNotBlank(queryParam.getSearchValue())) {
                //字典类型表，仅提供名称、编码模糊查询
                queryWrapper
                        .and(param -> param.like("ele_air_strategy.name", queryParam.getSearchValue()));
            }
            if (queryParam.getType() == 0) {
                queryWrapper.eq("ele_air_strategy.type", queryParam.getType());
            } else {
                queryWrapper.ne("ele_air_strategy.type", 0);
            }

            //排序
            if (ObjectUtil.isAllNotEmpty(queryParam.getSortBy(), queryParam.getOrderBy())) {
                queryWrapper.orderBy(true, queryParam.getOrderBy().equals(DbConstant.ASC), StrUtil.toUnderlineCase(queryParam.getSortBy()));
            } else {
                //没有排序参数，默认根据sort字段排序，没有排序字段的，根据updateTime更新时间排序
                queryWrapper.orderBy(true, true, "ele_air_strategy.sort");
            }


        }
        queryWrapper.eq("ele_air_strategy.state", DataStateEnum.ENABLE.getCode());
        return this.baseMapper.selectPage(new Page<>(PageFactory.getPageNum(queryParam), PageFactory.getPageSize(queryParam)), queryWrapper);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean addAirStrategy(AirStrategyParam airStrategyParam) {


        checkDicTypeName(airStrategyParam, false);

        if (airStrategyParam.getType() == 0) {
            //默认策略
            //校验参数
            LambdaQueryWrapper<AirStrategy> tem = new LambdaQueryWrapper<>();
            tem.eq(AirStrategy::getState, DataStateEnum.ENABLE.getCode());
            tem.eq(AirStrategy::getType,0);
            List<AirStrategy> temList = this.list(tem);
            if (CollectionUtil.isNotEmpty(temList)) {
                for (AirStrategy airStrategy : temList) {
                    LocalDateTime start = LocalDateTimeUtil.parse(airStrategy.getStartTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                    LocalDateTime startParam = LocalDateTimeUtil.parse(airStrategyParam.getStartTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                    long rangeStart = LocalDateTimeUtil.between(start, startParam).toMinutes();
                    LocalDateTime end = LocalDateTimeUtil.parse(airStrategy.getStartTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                    LocalDateTime endParam = LocalDateTimeUtil.parse(airStrategyParam.getStartTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                    long rangeEnd = LocalDateTimeUtil.between(end, endParam).toMinutes();

                    if (rangeStart <= 5 || rangeEnd <= 5) {
                        throw new BusinessException(EnergyResponseEnum.TASK_TIME_ERROR);
                    }
                }

            }
            AirStrategy airStrategy = assAirStrategy(airStrategyParam);

            List<AirSet> list = new ArrayList<>();
            airStrategyParam.getLineIds().forEach((item) -> {
                AirSet airSet = new AirSet(airStrategy.getId(), item);
                list.add(airSet);
            });
            iEleSetService.saveBatch(list);


            //时间表达式需要处理成 秒 分 时 日 月 ？ 年，标志指定时间执行一次
            LocalDateTime startTime = LocalDateTimeUtil.parse(airStrategy.getStartTime(), DatePattern.NORM_DATETIME_PATTERN);
            String cronOn = "" + startTime.getSecond() + CharUtil.SPACE + startTime.getMinute() + CharUtil.SPACE + startTime.getHour() + CharUtil.SPACE + startTime.getDayOfMonth() + CharUtil.SPACE + startTime.getMonthValue() + CharUtil.SPACE + "?" + CharUtil.SPACE + startTime.getYear();
            String paramOn = "open" + StrUtil.C_COMMA + airStrategy.getId();

        /*    XxlJobInfo xxlJobInfo = assScheduling("测试定时开空调",cronOn,paramOn);
            HttpResult<String> result = jobFeignClient.addJob(xxlJobInfo);
            if (result.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }


            //时间表达式需要处理成 秒 分 时 日 月 ？ 年，标志指定时间执行一次
            LocalDateTime startTimeEnd = LocalDateTimeUtil.parse(airStrategy.getEndTime(), DatePattern.NORM_DATETIME_PATTERN);
            String cronOff = "" + startTimeEnd.getSecond() + CharUtil.SPACE + startTimeEnd.getMinute() + CharUtil.SPACE + startTimeEnd.getHour() + CharUtil.SPACE + startTimeEnd.getDayOfMonth() + CharUtil.SPACE + startTimeEnd.getMonthValue() + CharUtil.SPACE + "?" + CharUtil.SPACE + startTimeEnd.getYear();
            String paramOff = "close" + StrUtil.C_COMMA + airStrategy.getId();
            XxlJobInfo airStrategyClose  = assScheduling("测试定时关空调",cronOff,paramOff);
            HttpResult<String> result2 = jobFeignClient.addJob(airStrategyClose);
            if (result2.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }

            String ids = result.getData() + StrUtil.COMMA + result2.getData();
            airStrategy.setXxlId(ids);*/
            this.updateById(airStrategy);
        } else if(airStrategyParam.getType() == 1) {


            //开关机策略
            AirStrategy airStrategyOnOff = assAirStrategy(airStrategyParam);

            List<AirSet> list = new ArrayList<>();
            airStrategyParam.getLineIds().forEach((item) -> {
                AirSet airSet = new AirSet(airStrategyOnOff.getId(), item);
                list.add(airSet);
            });
            iEleSetService.saveBatch(list);
            /*
             * 添加开机调度任务
             */
            //时间表达式需要处理成 秒 分 时 日 月 ？ 年，标志指定时间执行一次
      /*      String startTime = airStrategyOnOff.getStartTime();
            String[] temTime = startTime.split(":");
            String cron = "0" + CharUtil.SPACE + temTime[1] + CharUtil.SPACE + temTime[0] + CharUtil.SPACE + "* * ?";
            String param = "close" + StrUtil.C_COMMA + airStrategyOnOff.getId();
            XxlJobInfo xOn = assScheduling("每日定时关空调",cron,param);
            HttpResult<String> result = jobFeignClient.addJob(xOn);
            if (result.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }
            airStrategyOnOff.setXxlId(result.getData());*/
            this.updateById(airStrategyOnOff);
        } else if(airStrategyParam.getType() == 2) {
            //开机策略
            AirStrategy airStrategyOnOff = assAirStrategy(airStrategyParam);
            List<AirSet> list = new ArrayList<>();
            airStrategyParam.getLineIds().forEach((item) -> {
                AirSet airSet = new AirSet(airStrategyOnOff.getId(), item);
                list.add(airSet);
            });
            iEleSetService.saveBatch(list);
            /*
             * 添加开机调度任务
             */
     /*       String startTime = airStrategyOnOff.getStartTime();
            String[] temTime = startTime.split(":");
            String cron = "0" + CharUtil.SPACE + temTime[1] + CharUtil.SPACE + temTime[0] + CharUtil.SPACE + "* * ?";
            String param = "open" + StrUtil.C_COMMA + airStrategyOnOff.getId();
            XxlJobInfo xOff = assScheduling("每日定时开空调",cron,param);
            HttpResult<String> result = jobFeignClient.addJob(xOff);
            if (result.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }
            airStrategyOnOff.setXxlId(result.getData());*/
            this.updateById(airStrategyOnOff);
        }

        return true;
    }


    /*组装数据库数据*/
    private AirStrategy  assAirStrategy(AirStrategyParam airStrategyParam){
        AirStrategy airStrategyOnOff = new AirStrategy();
        BeanUtils.copyProperties(airStrategyParam, airStrategyOnOff);
        airStrategyOnOff.setState(DataStateEnum.ENABLE.getCode());
        airStrategyOnOff.setStatus(0);
        airStrategyOnOff.setCreateBy(RequestUtil.getUserIndex());
        airStrategyOnOff.setCreateTime(LocalDateTime.now());
        airStrategyOnOff.setUpdateTime(LocalDateTime.now());
        this.save(airStrategyOnOff);
        return airStrategyOnOff;
    }

    /*组装调度任务*/
/*
    private XxlJobInfo assScheduling(String desc,String confTime,String param){
        XxlJobInfo xxlJobInfo = new XxlJobInfo();
        //执行器ID，手动指定，此处设置为3，对应xxl_job_group中的executor--测试执行器
        xxlJobInfo.setJobGroup(xxlGroup);
        //任务描述
        xxlJobInfo.setJobDesc(desc);
        xxlJobInfo.setAuthor("cdf");
        xxlJobInfo.setScheduleType("CRON");
        xxlJobInfo.setScheduleConf(confTime);
        xxlJobInfo.setMisfireStrategy("DO_NOTHING");
        xxlJobInfo.setExecutorRouteStrategy("FIRST");
        xxlJobInfo.setExecutorBlockStrategy("SERIAL_EXECUTION");
        //指定处理器
        xxlJobInfo.setExecutorHandler("airStrategyJob");
        //传递的参数
        xxlJobInfo.setExecutorParam(param);
        xxlJobInfo.setGlueType("BEAN");
        xxlJobInfo.setGlueRemark("GLUE代码初始化");
        return xxlJobInfo;
    }
*/

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean updateAirStrategy(AirStrategyParam.AirUpdateParam updateParam) {
        AirStrategy airStrategyRes = this.getById(updateParam.getId());
        if (Objects.isNull(airStrategyRes)) {
            throw new BusinessException(CommonResponseEnum.FAIL);
        }
        checkDicTypeName(updateParam, true);

        AirStrategy airStrategy = new AirStrategy();
        BeanUtils.copyProperties(updateParam, airStrategy);
        airStrategy.setUpdateBy(RequestUtil.getUserIndex());
        airStrategy.setUpdateTime(LocalDateTime.now());
        this.updateById(airStrategy);


        iEleSetService.removeById(airStrategy.getId());
        List<AirSet> list = new ArrayList<>();
        updateParam.getLineIds().forEach((item) -> {
            AirSet airSet = new AirSet(airStrategy.getId(), item);
            list.add(airSet);
        });
        iEleSetService.saveBatch(list);

/*
        String[] jobIds = airStrategyRes.getXxlId().split(StrUtil.COMMA);
        if(updateParam.getType() == 0) {
            //时间表达式需要处理成 秒 分 时 日 月 ？ 年，标志指定时间执行一次
            LocalDateTime startTime = LocalDateTimeUtil.parse(airStrategy.getStartTime(), DatePattern.NORM_DATETIME_PATTERN);
            String cronOn = "" + startTime.getSecond() + CharUtil.SPACE + startTime.getMinute() + CharUtil.SPACE + startTime.getHour() + CharUtil.SPACE + startTime.getDayOfMonth() + CharUtil.SPACE + startTime.getMonthValue() + CharUtil.SPACE + "?" + CharUtil.SPACE + startTime.getYear();
            String paramOn = "open" + StrUtil.C_COMMA + airStrategy.getId();

            XxlJobInfo xxlJobInfo = assScheduling("测试开空调",cronOn,paramOn);
            xxlJobInfo.setId(Integer.parseInt(jobIds[0]));
            HttpResult<String> result = jobFeignClient.updateJob(xxlJobInfo);
            if (result.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }
            LocalDateTime startTimeEnd = LocalDateTimeUtil.parse(airStrategy.getEndTime(), DatePattern.NORM_DATETIME_PATTERN);
            String cronOff = "" + startTimeEnd.getSecond() + CharUtil.SPACE + startTimeEnd.getMinute() + CharUtil.SPACE + startTimeEnd.getHour() + CharUtil.SPACE + startTimeEnd.getDayOfMonth() + CharUtil.SPACE + startTimeEnd.getMonthValue() + CharUtil.SPACE + "?" + CharUtil.SPACE + startTimeEnd.getYear();
            String paramOff = "close" + StrUtil.C_COMMA + airStrategy.getId();

            XxlJobInfo xxlJobInfoOff = assScheduling("测试关空调",cronOff,paramOff);
            xxlJobInfoOff.setId(Integer.parseInt(jobIds[1]));
            HttpResult<String> result2 = jobFeignClient.updateJob(xxlJobInfoOff);
            if (result.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }

            if (result2.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }
        }else if(updateParam.getType() == 1){
            String[] temTime = updateParam.getStartTime().split(":");
            String cron = "0" + CharUtil.SPACE + temTime[1] + CharUtil.SPACE + temTime[0] + CharUtil.SPACE + "* * ?";
            String param = "close" + StrUtil.C_COMMA + airStrategy.getId();
            XxlJobInfo xOff = assScheduling("每日定时关空调",cron,param);
            xOff.setId(Integer.parseInt(jobIds[0]));
            HttpResult<String> result = jobFeignClient.updateJob(xOff);
            if (result.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }
        }else if(updateParam.getType() == 2){
            String[] temTime = updateParam.getStartTime().split(":");
            String cron = "0" + CharUtil.SPACE + temTime[1] + CharUtil.SPACE + temTime[0] + CharUtil.SPACE + "* * ?";
            String param = "open" + StrUtil.C_COMMA + airStrategy.getId();
            XxlJobInfo xOff = assScheduling("每日定时开空调",cron,param);
            xOff.setId(Integer.parseInt(jobIds[0]));
            HttpResult<String> result = jobFeignClient.updateJob(xOff);
            if (result.getCode().equals(String.valueOf(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_CREATE_ERROR);
            }
        }*/
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean deleteAirStrategy(String id) {
        AirStrategy airStrategy = this.getById(id);
        if (Objects.isNull(airStrategy)) {
            return true;
        }

        if (airStrategy.getStatus() == 1) {
            throw new BusinessException(EnergyResponseEnum.TASK_RUNING);
        }

        this.lambdaUpdate()
                .set(AirStrategy::getState, DataStateEnum.DELETED.getCode())
                .eq(AirStrategy::getId, id)
                .update();

        LambdaQueryWrapper<AirSet> lambdaQueryWrapper = new LambdaQueryWrapper<>();
        lambdaQueryWrapper.eq(AirSet::getId, id);
        iEleSetService.remove(lambdaQueryWrapper);

    /*    String[] arr = airStrategy.getXxlId().split(StrUtil.COMMA);

        for (String xxlId : arr) {
            HttpResult<String> result = jobFeignClient.removeJob(Integer.parseInt(xxlId));
            if (result.getCode().equals(StrUtil.toString(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_DEL_ERROR);
            }
        }*/

        return true;
    }


    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean stopAirStrategy(String id) {
        AirStrategy airStrategy = this.getById(id);
        if (Objects.isNull(airStrategy)) {
            throw new BusinessException(CommonResponseEnum.FAIL);
        }
        if (airStrategy.getStatus() != 1) {
            throw new BusinessException(EnergyResponseEnum.TASK_NO_RUN);
        }
    /*    String[] arr = airStrategy.getXxlId().split(StrUtil.COMMA);
        for (String xxlId : arr) {
            HttpResult<String> result = jobFeignClient.stopJob(Integer.parseInt(xxlId));
            if (result.getCode().equals(StrUtil.toString(ReturnT.FAIL_CODE))) {
                throw new BusinessException(EnergyResponseEnum.TASK_STOP_ERROR);
            }
        }*/

        this.lambdaUpdate()
                .set(AirStrategy::getStatus, 4)
                .eq(AirStrategy::getId, id)
                .update();

        return true;
    }


    @Override
    public AirStrategyVO findById(String id) {
        AirStrategy airStrategy = this.lambdaQuery().eq(AirStrategy::getId, id).one();
        if (Objects.isNull(airStrategy)) {
            throw new BusinessException(CommonResponseEnum.NO_DATA);
        }
        List<AirSet> list = iEleSetService.lambdaQuery().eq(AirSet::getId, id).list();
        List<String> lineIds = list.stream().map(AirSet::getLineId).collect(Collectors.toList());

        AirStrategyVO airStrategyVO = new AirStrategyVO();
        BeanUtils.copyProperties(airStrategy, airStrategyVO);
        airStrategyVO.setLineIds(lineIds);
        return airStrategyVO;
    }

    @Override
    public boolean startAirStrategy(String id) {
        AirStrategy airStrategy = getById(id);
        if (Objects.isNull(airStrategy)) {
            throw new BusinessException(CommonResponseEnum.FAIL);
        }
        String[] ids = airStrategy.getXxlId().split(StrUtil.COMMA);
        if (airStrategy.getStatus() == 0) {
            //可以启动
           /* for (String xxlId : ids) {
                HttpResult<String> result = jobFeignClient.startTask(Integer.parseInt(xxlId));
                if (result.getCode().equals(StrUtil.toString(ReturnT.FAIL_CODE))) {
                    throw new BusinessException(EnergyResponseEnum.TASK_RUNING);
                }
            }*/
            this.lambdaUpdate().set(AirStrategy::getStatus, 1).eq(AirStrategy::getId, id).update();
        } else if (airStrategy.getStatus() == 1) {
            //已经执行中
          /*  for (String xxlId : ids) {
                HttpResult<String> result = jobFeignClient.stopJob(Integer.parseInt(xxlId));
                if (result.getCode().equals(StrUtil.toString(ReturnT.FAIL_CODE))) {
                    throw new BusinessException(EnergyResponseEnum.TASK_DEL_ERROR);
                }
            }*/
            this.lambdaUpdate().set(AirStrategy::getStatus, 0).eq(AirStrategy::getId, id).update();
        } else {
            //已经完成
            throw new BusinessException(EnergyResponseEnum.TASK_HAS_START);
        }
        return true;
    }


    @Override
    public void dealAirStrategyId(String operation) {
        ExecutorService executor = Executors.newSingleThreadExecutor(); // 顺序执行
        try {
            AirStrategy airStrategy = this.list(new LambdaQueryWrapper<>()).get(0);
            if (Objects.isNull(airStrategy)) {
                throw new BusinessException(CommonResponseEnum.FAIL);
            }

            LambdaQueryWrapper<AirSet> lambdaQueryWrapper = new LambdaQueryWrapper<>();
            lambdaQueryWrapper.eq(AirSet::getId, airStrategy.getId());
            List<AirSet> list = iEleSetService.list(lambdaQueryWrapper);

            if (CollectionUtil.isNotEmpty(list)) {
                List<String> lineIds = list.stream().map(AirSet::getLineId).collect(Collectors.toList());

                // 如果是关闭操作，过滤出当前开启的空调
                if ("close".equals(operation)) {
                    StringBuilder temSql = assToInfluxParam(lineIds);
                    String sql = "select LineId,ACInStat from air_data_Real where " + temSql + " and ACInStat = 'On'";
                    QueryResult result = influxDbUtils.query(sql);
                    List<QueryResult.Series> listSeries = result.getResults().get(0).getSeries();
                    if (!CollectionUtils.isEmpty(listSeries)) {
                        lineIds = listSeries.get(0).getValues().stream()
                                .map(item -> item.get(1).toString())
                                .collect(Collectors.toList());
                    }
                }

                List<NetAndDevByLineDTO> netDevList = logicDeviceLineMapper.getNetAndDevByLineIds(lineIds);
                if (CollectionUtil.isNotEmpty(netDevList)) {
                    // 提交任务到线程池，按顺序执行
                    executor.submit(() -> {
                        for (NetAndDevByLineDTO device : netDevList) {
                            boolean success = false;
                            int retryCount = 0;

                            // 每个设备独立重试3次
                            while (retryCount < 3 && !success) {
                                try {
                                    sendAirControlCommand(device, operation, airStrategy);
                                    success = waitForDeviceResponse(device.getDevId()); // 等待响应
                                    retryCount++;
                                } catch (Exception e) {
                                    retryCount++;
                                    if (retryCount >= 3) {
                                        System.err.println("设备 " + device.getDevId() + " 控制失败，已达最大重试次数");
                                    }
                                }
                            }

                            if (!success) {

                                log.error("跳过设备: " + device.getDevId() + "（控制失败或无响应）");
                            }
                        }


                        // 所有设备控制完成后更新状态
                        if ("close".equals(operation) && airStrategy.getType() == 0) {
                            AirStrategy update = new AirStrategy();
                            update.setId(airStrategy.getId());
                            update.setStatus(2);
                            airStrategyMapper.updateById(update);
                        }


                    });
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executor.shutdown(); // 关闭线程池
            deviceResponseMap.clear();
        }
    }


    private void sendAirControlCommand(NetAndDevByLineDTO device, String operation, AirStrategy airStrategy) {
        Gson gson = new Gson();
        String power = "open".equals(operation) ? "On" : "Off";
        String array = "ACInStat";

        AirStrategyDTO airStrategyDTO = new AirStrategyDTO();
        airStrategyDTO.setMid(1);
        airStrategyDTO.setDeviceId(device.getDevId());
        airStrategyDTO.setUserId(clientId);
        airStrategyDTO.setTimestamp(Calendar.getInstance().getTimeInMillis());
        airStrategyDTO.setExpire(-1);
        airStrategyDTO.setType("CMD_DEV_CTRL");

        AirStrategyDTO.Param param = new AirStrategyDTO.Param();
        param.setLineId(device.getLineId());
        param.setAction(array);
        param.setParm(power);
        airStrategyDTO.setParam(param);

        String json = gson.toJson(airStrategyDTO);
        String topic = "/platform/devcmd/v1/" + device.getNdid();
        log.info("开始往{}发送控制命令:{}",topic,json);
        publisher.send(topic, json, 1, false);
    }

    private boolean waitForDeviceResponse(String deviceId) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        long timeout = 15000; // 15秒超时

        while (System.currentTimeMillis() - startTime < timeout) {
            // 检查是否收到响应（可通过共享变量或消息队列实现）
            if (isDeviceResponded(deviceId)) { // 需要实现此方法
                return true;
            }
            Thread.sleep(1000); // 每秒检查一次
        }
        return false; // 超时未响应
    }

    // 示例：检查设备是否响应（需根据实际MQTT订阅逻辑实现）
    private boolean isDeviceResponded(String deviceId) {
        // 例如：检查某个共享Map或数据库标记
        if(deviceResponseMap.containsKey(deviceId) && deviceResponseMap.get(deviceId)){
            System.out.println("匹配到已经响应完成"+deviceId);
            return true;
        }
        return false;
    }




    @MqttSubscribe(value = "/platform/devack/#", qos = 1)
    public void airOperation(String topic, MqttMessage message, @Payload String payload) {
        System.out.println("收到网关反馈控制"+LocalDateTime.now());
        System.out.println(message.toString());
        JSONObject jsonObject = new JSONObject(payload);
        String userId = jsonObject.getStr("userId");

        if (clientId.equals(userId)) {
            String code = jsonObject.getStr("code");
            String devId = jsonObject.get("deviceId").toString();
            if ("200".equals(code)) {
                markDeviceAsResponded(devId); // 标记设备已响应
            } else {
                System.err.println("设备 " + devId + " 返回错误: " + payload);
            }
        }
    }

    // 示例：标记设备响应（需实现具体逻辑）
    private void markDeviceAsResponded(String deviceId) {
        // 1. 更新内存中的状态（适用于单机应用）
        deviceResponseMap.put(deviceId, true);
    }



    /**
     * 校验参数，检查是否存在相同名称的空调策略
     */
    private void checkDicTypeName(AirStrategyParam airStrategyParam, boolean isExcludeSelf) {
        LambdaQueryWrapper<AirStrategy> netDevLambdaQueryWrapper = new LambdaQueryWrapper<>();
        netDevLambdaQueryWrapper
                .eq(AirStrategy::getName, airStrategyParam.getName())
                .eq(AirStrategy::getState, DataStateEnum.ENABLE.getCode());
        //更新的时候，需排除当前记录
        if (isExcludeSelf) {
            if (airStrategyParam instanceof AirStrategyParam.AirUpdateParam) {
                netDevLambdaQueryWrapper.ne(AirStrategy::getId, ((AirStrategyParam.AirUpdateParam) airStrategyParam).getId());
            }
        }
        int countByAccount = this.count(netDevLambdaQueryWrapper);
        //大于等于1个则表示重复
        if (countByAccount >= 1) {
            throw new BusinessException(EnergyResponseEnum.STRATEGY_REPEAT);
        }
    }

    public static StringBuilder assToInfluxParam(List<String> lineIds) {
        StringBuilder lineIdsStr = new StringBuilder("(");
        for (int i = 0; i < lineIds.size(); i++) {
            if (lineIds.size() - i != 1) {
                lineIdsStr.append("LineId ='").append(lineIds.get(i)).append("' or ");
            } else {
                lineIdsStr.append("LineId ='").append(lineIds.get(i)).append("') ");
            }
        }
        return lineIdsStr;
    }

}
