package com.usky.cdi.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.usky.cdi.domain.DmpDevice;
import com.usky.cdi.domain.DmpProduct;
import com.usky.cdi.mapper.DmpDeviceMapper;
import com.usky.cdi.mapper.DmpProductMapper;
import com.usky.cdi.service.config.mqtt.MqttBaseConfig;
import com.usky.cdi.service.config.mqtt.MqttOutConfig;
import com.usky.cdi.service.enums.EnvMonitorMqttTopic;
import com.usky.cdi.service.util.DeviceDataQuery;
import com.usky.cdi.service.util.SnowflakeIdGenerator;
import com.usky.cdi.service.vo.IotDataTransferVO;
import com.usky.cdi.service.vo.base.*;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Service;
import org.springframework.web.context.ContextLoader;

import javax.annotation.PostConstruct;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author fyc
 * @email yuchuan.fu@chinausky.com
 * @date 2025/11/20
 */
@Slf4j
@Service
public class IotDataTransferService {

    private MqttOutConfig.MqttGateway mqttGateway;

    // 注入ApplicationContext，确保总是能获取到
    @Autowired
    private ApplicationContext context;

    // MQTT连接相关配置
    private static final String MQTT_URL = "ssl://114.80.201.143:8883";
    private static final String MQTT_TOPIC = "iotInfo/+";
    private static final int KEEP_ALIVE_INTERVAL = 60;
    private static final int COMPLETION_TIMEOUT = 5000;

    private SnowflakeIdGenerator idGenerator;

    @Autowired
    private DeviceDataQuery deviceDataQuery;

    @Autowired
    private DmpDeviceMapper dmpDeviceMapper;

    @Value("${device.data.simulation}")
    private boolean simulation;

    @Autowired
    private DmpProductMapper dmpProductMapper;

    // 从配置文件读取Snowflake参数，默认值为2
    @Value("${snowflake.worker-id:2}")
    private long workerId;

    @Value("${snowflake.data-center-id:2}")
    private long dataCenterId;

    @PostConstruct
    public void init() {
        this.idGenerator = new SnowflakeIdGenerator(workerId, dataCenterId);
    }

    /**
     * 获取当前时间
     */
    private LocalDateTime getCurrentTime() {
        return LocalDateTime.now();
    }

    /**
     * 生成数据包ID
     */
    private Long generateDataPacketID() {
        return idGenerator.nextPacketId();
    }

    /**
     * 发送水浸状态（deviceType：702）
     * Topic: iotInfo/flooded
     *
     * @return 推送结果，包含成功数和失败数
     */
    public Map<String, Integer> sendWaterLeak(IotDataTransferVO transferVO) {
        Map<String, Integer> result = new HashMap<>();
        result.put("successCount", 0);
        result.put("failureCount", 0);

        if (!validateMqttGateway()) {
            return result;
        }

        try {
            List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
            log.warn("获取到的数据：{}", deviceData);
            Integer deviceType = transferVO.getDeviceType();
            Integer totalDevices = transferVO.getDevices().size();

            log.info("开始推送水浸状态数据，设备类型：{}，设备数量：{}，获取到的数据条数：{}",
                    deviceType, totalDevices, deviceData.size());

            if (deviceData.isEmpty()) {
                log.warn("没有获取到水浸数据！设备类型：{}", deviceType);
                result.put("failureCount", totalDevices);
                return result;
            }

            Long engineeringId = transferVO.getEngineeringId();
            for (JSONObject deviceDataItem : deviceData) {
                LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
                if (dataEndTime == null) {
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                Integer deviceId = deviceDataItem.getIntValue("device_id");
                Integer value = deviceDataItem.getInteger("leach_status");
                if (value == null) {
                    log.warn("设备{}的水浸状态数据为空", deviceId);
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                WaterLeakVO vo = new WaterLeakVO();
                vo.setDataPacketID(generateDataPacketID());
                vo.setSensorID(deviceId);
                vo.setEngineeringID(engineeringId);
                vo.setPublishTime(getCurrentTime());
                vo.setSensorValue(value);
                vo.setDataEndTime(dataEndTime);

                try {
                    sendMqttMessage(EnvMonitorMqttTopic.WATER_LEAK, vo, "水浸状态信息");
                    result.put("successCount", result.get("successCount") + 1);
                } catch (Exception e) {
                    log.warn("设备{}的水浸状态数据推送失败：{}", deviceId, e.getMessage());
                    result.put("failureCount", result.get("failureCount") + 1);
                }
            }

            log.info("水浸状态数据推送完成，设备类型：{}，成功：{}，失败：{}",
                    deviceType, result.get("successCount"), result.get("failureCount"));

            return result;
        } catch (Exception e) {
            log.error("水浸状态数据推送发生异常", e);
            result.put("failureCount", transferVO.getDevices().size());
            return result;
        }
    }

    /**
     * 发送温湿度及气体浓度数据(设备类型:701,707-711)
     * 包含: wd(温度), sd(湿度), o2(氧气), co(一氧化碳), co2(二氧化碳)
     *
     * @return 推送结果，包含成功数和失败数
     */
    public Map<String, Integer> sendEnvData(IotDataTransferVO transferVO) {
        Map<String, Integer> result = new HashMap<>();
        result.put("successCount", 0);
        result.put("failureCount", 0);

        if (!validateMqttGateway()) {
            return result;
        }

        try {
            Integer deviceType = transferVO.getDeviceType();
            List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
            Integer totalDevices = transferVO.getDevices().size();

            // 统计各类型设备数据数量
            int tempCount = 0, humidityCount = 0, coCount = 0, o2Count = 0, co2Count = 0;
            for (JSONObject dataItem : deviceData) {
                if (dataItem.containsKey("wd") && dataItem.getFloat("wd") != null) tempCount++;
                if (dataItem.containsKey("sd") && dataItem.getFloat("sd") != null) humidityCount++;
                if (dataItem.containsKey("co") && dataItem.getFloat("co") != null) coCount++;
                if (dataItem.containsKey("o2") && dataItem.getFloat("o2") != null) o2Count++;
                if (dataItem.containsKey("co2") && dataItem.getFloat("co2") != null) co2Count++;
            }

            log.info("开始推送温湿度及气体浓度数据，设备类型：{}，设备数量：{}，获取到的数据条数：{}",
                    deviceType, totalDevices, deviceData.size());
            log.info("各类型设备数据数量：温度{}条，湿度{}条，一氧化碳{}条，氧气{}条，二氧化碳{}条",
                    tempCount, humidityCount, coCount, o2Count, co2Count);

            if (deviceData.isEmpty()) {
                log.warn("没有获取到空气质量数据！设备类型：{}", deviceType);
                result.put("failureCount", totalDevices);
                return result;
            }

            Long engineeringId = transferVO.getEngineeringId();
            for (JSONObject deviceDataItem : deviceData) {
                LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
                if (dataEndTime == null) {
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                Integer deviceId = deviceDataItem.getIntValue("device_id");
                boolean deviceSuccess = true;

                // 根据设备类型发送对应的数据
                try {
                    switch (deviceType) {
                        case 707:
                            sendTempData(deviceId, dataEndTime, deviceDataItem, engineeringId);
                            break;
                        case 708:
                            sendHumidityData(deviceId, dataEndTime, deviceDataItem, engineeringId);
                            break;
                        case 709:
                            sendOxygenData(deviceId, dataEndTime, deviceDataItem, engineeringId);
                            break;
                        case 710:
                            sendCo2Data(deviceId, dataEndTime, deviceDataItem, engineeringId);
                            break;
                        case 711:
                            sendCoData(deviceId, dataEndTime, deviceDataItem, engineeringId);
                            break;
                    }
                } catch (Exception e) {
                    log.warn("设备{}的环境数据推送失败：{}", deviceId, e.getMessage());
                    deviceSuccess = false;
                }

                // 统计设备推送结果
                if (deviceSuccess) {
                    result.put("successCount", result.get("successCount") + 1);
                } else {
                    result.put("failureCount", result.get("failureCount") + 1);
                }
            }

            log.info("温湿度及气体浓度数据推送完成，设备类型：{}，成功：{}，失败：{}",
                    deviceType, result.get("successCount"), result.get("failureCount"));

            return result;
        } catch (Exception e) {
            log.error("温湿度及气体浓度数据推送发生异常", e);
            result.put("failureCount", transferVO.getDevices().size());
            return result;
        }
    }

    /**
     * 发送人员闯入情况（703）
     *
     * @return 推送结果，包含成功数和失败数
     **/
    public Map<String, Integer> sendPersonPresence(IotDataTransferVO transferVO) {
        Map<String, Integer> result = new HashMap<>();
        result.put("successCount", 0);
        result.put("failureCount", 0);

        if (!validateMqttGateway()) {
            return result;
        }

        try {
            List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
            Integer deviceType = transferVO.getDeviceType();
            Integer totalDevices = transferVO.getDevices().size();

            log.info("开始推送人员闯入情况数据，设备类型：{}，设备数量：{}，获取到的数据条数：{}",
                    deviceType, totalDevices, deviceData.size());

            if (deviceData.isEmpty()) {
                log.warn("没有获取到人员闯入情况数据！设备类型：{}", deviceType);
                result.put("failureCount", totalDevices);
                return result;
            }

            Long engineeringId = transferVO.getEngineeringId();
            for (JSONObject deviceDataItem : deviceData) {
                LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
                if (dataEndTime == null) {
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                Integer deviceId = deviceDataItem.getIntValue("device_id");

                PersonPresenceVO vo = new PersonPresenceVO();
                vo.setDataPacketID(generateDataPacketID());
                vo.setSensorID(deviceId);
                vo.setDataEndTime(dataEndTime);
                vo.setPublishTime(getCurrentTime());
                vo.setEngineeringID(engineeringId);
                // 传感器值固定为0（可能是默认值或占位符）
                vo.setSensorValue(0);

                try {
                    sendMqttMessage(EnvMonitorMqttTopic.PERSON_PRESENCE, vo, "人员闯入情况");
                    result.put("successCount", result.get("successCount") + 1);
                } catch (Exception e) {
                    log.warn("设备{}的人员闯入情况数据推送失败：{}", deviceId, e.getMessage());
                    result.put("failureCount", result.get("failureCount") + 1);
                }
            }

            log.info("人员闯入情况数据推送完成，设备类型：{}，成功：{}，失败：{}",
                    deviceType, result.get("successCount"), result.get("failureCount"));

            return result;
        } catch (Exception e) {
            log.error("人员闯入情况数据推送发生异常", e);
            result.put("failureCount", transferVO.getDevices().size());
            return result;
        }
    }

    /**
     * 发送人防用电负荷情况（704）
     *
     * @return 推送结果，包含成功数和失败数
     **/
    public Map<String, Integer> sendElectricityLoad(IotDataTransferVO transferVO) {
        Map<String, Integer> result = new HashMap<>();
        result.put("successCount", 0);
        result.put("failureCount", 0);

        if (!validateMqttGateway()) {
            return result;
        }

        try {
            List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
            Integer deviceType = transferVO.getDeviceType();
            Integer totalDevices = transferVO.getDevices().size();

            log.info("开始推送人防用电负荷情况数据，设备类型：{}，设备数量：{}，获取到的数据条数：{}",
                    deviceType, totalDevices, deviceData.size());

            if (deviceData.isEmpty()) {
                log.warn("没有获取到人防用电负荷情况数据！设备类型：{}", deviceType);
                result.put("failureCount", totalDevices);
                return result;
            }

            Long engineeringId = transferVO.getEngineeringId();
            for (JSONObject deviceDataItem : deviceData) {
                LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
                if (dataEndTime == null) {
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                Integer deviceId = deviceDataItem.getIntValue("device_id");
                ElectricityLoadVO vo = new ElectricityLoadVO();
                vo.setDataPacketID(generateDataPacketID());
                vo.setSensorID(deviceId);
                vo.setEngineeringID(engineeringId);
                vo.setPublishTime(getCurrentTime());
                vo.setDataEndTime(dataEndTime);
                vo.setAVoltage(deviceDataItem.getFloat("aVoltage"));
                vo.setBVoltage(deviceDataItem.getFloat("bVoltage"));
                vo.setCVoltage(deviceDataItem.getFloat("cVoltage"));
                vo.setAElectricity(deviceDataItem.getFloat("aElectricity"));
                vo.setBElectricity(deviceDataItem.getFloat("bElectricity"));
                vo.setCElectricity(deviceDataItem.getFloat("cElectricity"));
                vo.setLine1TEMP(deviceDataItem.getFloat("line1TEMP"));
                vo.setLine2TEMP(deviceDataItem.getFloat("Line2TEMP"));
                vo.setLine3TEMP(deviceDataItem.getFloat("Line3TEMP"));
                vo.setLeakageCurrent(deviceDataItem.getFloat("leakageCurrent"));

                // 根据模拟模式选择不同的功率字段
                vo.setTotalPower(simulation ?
                        deviceDataItem.getFloat("totalPower") :
                        deviceDataItem.getFloat("active_power"));

                try {
                    sendMqttMessage(EnvMonitorMqttTopic.ELECTRICITY_LOAD, vo, "人防用电负荷情况");
                    result.put("successCount", result.get("successCount") + 1);
                } catch (Exception e) {
                    log.warn("设备{}的人防用电负荷情况数据推送失败：{}", deviceId, e.getMessage());
                    result.put("failureCount", result.get("failureCount") + 1);
                }
            }

            log.info("人防用电负荷情况数据推送完成，设备类型：{}，成功：{}，失败：{}",
                    deviceType, result.get("successCount"), result.get("failureCount"));

            return result;
        } catch (Exception e) {
            log.error("人防用电负荷情况数据推送发生异常", e);
            result.put("failureCount", transferVO.getDevices().size());
            return result;
        }
    }

    /**
     * 推送温度信息（701）
     *
     * @param deviceDataItem 设备数据
     * @param deviceId 设备ID
     * @param dataEndTime 数据结束时间
     * @param engineeringID 工程ID
     **/
    private void sendTempData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
        Float value = deviceDataItem.getFloat("wd");
        if (value == null) {
            log.warn("设备{}的温度数据为空", deviceId);
            return;
        }
        TempVO tempVO = new TempVO();
        tempVO.setDataPacketID(generateDataPacketID());
        tempVO.setSensorID(deviceId);
        tempVO.setEngineeringID(engineeringID);
        tempVO.setPublishTime(getCurrentTime());
        tempVO.setSensorValue(value);
        tempVO.setDataEndTime(dataEndTime);
        sendMqttMessage(EnvMonitorMqttTopic.TEMP, tempVO, "温度信息");
    }

    /**
     * 推送湿度信息（702）
     *
     * @param deviceDataItem 设备数据
     * @param deviceId 设备ID
     * @param dataEndTime 数据结束时间
     * @param engineeringID 工程ID
     **/
    private void sendHumidityData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
        Float value = deviceDataItem.getFloat("sd");
        if (value == null) {
            log.warn("设备{}的湿度数据为空", deviceId);
            return;
        }
        HumidityVO humidityVO = new HumidityVO();
        humidityVO.setDataPacketID(generateDataPacketID());
        humidityVO.setSensorID(deviceId);
        humidityVO.setEngineeringID(engineeringID);
        humidityVO.setPublishTime(getCurrentTime());
        humidityVO.setSensorValue(value);
        humidityVO.setDataEndTime(dataEndTime);
        sendMqttMessage(EnvMonitorMqttTopic.HUMIDITY, humidityVO, "湿度信息");
    }

    /**
     * 推送氧气浓度信息（705）
     *
     * @param deviceDataItem 设备数据
     * @param deviceId 设备ID
     * @param dataEndTime 数据结束时间
     * @param engineeringID 工程ID
     **/
    private void sendOxygenData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
        Float value = deviceDataItem.getFloat("o2");
        if (value == null) {
            log.warn("设备{}的氧气浓度数据为空", deviceId);
            return;
        }
        OxygenVO oxygenVO = new OxygenVO();
        oxygenVO.setDataPacketID(generateDataPacketID());
        oxygenVO.setSensorID(deviceId);
        oxygenVO.setEngineeringID(engineeringID);
        oxygenVO.setPublishTime(getCurrentTime());
        oxygenVO.setSensorValue(value);
        oxygenVO.setDataEndTime(dataEndTime);
        sendMqttMessage(EnvMonitorMqttTopic.OXYGEN, oxygenVO, "氧气浓度信息");
    }

    /**
     * 推送一氧化碳浓度信息（706）
     *
     * @param deviceDataItem 设备数据
     * @param deviceId 设备ID
     * @param dataEndTime 数据结束时间
     * @param engineeringID 工程ID
     **/
    private void sendCoData(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
        Float value = deviceDataItem.getFloat("co");
        if (value == null) {
            log.warn("设备{}的一氧化碳浓度数据为空", deviceId);
            return;
        }
        CoVO coVO = new CoVO();
        coVO.setDataPacketID(generateDataPacketID());
        coVO.setSensorID(deviceId);
        coVO.setEngineeringID(engineeringID);
        coVO.setPublishTime(getCurrentTime());
        coVO.setSensorValue(value);
        coVO.setDataEndTime(dataEndTime);
        sendMqttMessage(EnvMonitorMqttTopic.CO, coVO, "一氧化碳浓度信息");
    }

    /**
     * 推送二氧化碳浓度信息（707）
     *
     * @param deviceDataItem 设备数据
     * @param deviceId 设备ID
     * @param dataEndTime 数据结束时间
     * @param engineeringID 工程ID
     **/
    private void sendCo2Data(Integer deviceId, LocalDateTime dataEndTime, JSONObject deviceDataItem, Long engineeringID) {
        Float value = deviceDataItem.getFloat("co2");
        if (value == null) {
            log.warn("设备{}的二氧化碳浓度数据为空", deviceId);
            return;
        }
        Co2VO co2VO = new Co2VO();
        co2VO.setDataPacketID(generateDataPacketID());
        co2VO.setSensorID(deviceId);
        co2VO.setEngineeringID(engineeringID);
        co2VO.setPublishTime(getCurrentTime());
        co2VO.setSensorValue(value);
        co2VO.setDataEndTime(dataEndTime);
        sendMqttMessage(EnvMonitorMqttTopic.CO2, co2VO, "二氧化碳浓度信息");
    }

    /**
     * 发送倾斜数据（712）
     *
     * @return 推送结果，包含成功数和失败数
     **/
    public Map<String, Integer> sendTiltData(IotDataTransferVO transferVO) {
        Map<String, Integer> result = new HashMap<>();
        result.put("successCount", 0);
        result.put("failureCount", 0);

        if (!validateMqttGateway()) {
            return result;
        }

        try {
            List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
            Integer deviceType = transferVO.getDeviceType();
            Integer totalDevices = transferVO.getDevices().size();

            log.info("开始推送倾斜数据，设备类型：{}，设备数量：{}，获取到的数据条数：{}",
                    deviceType, totalDevices, deviceData.size());

            if (deviceData.isEmpty()) {
                log.warn("没有获取到倾斜数据！设备类型：{}", deviceType);
                result.put("failureCount", totalDevices);
                return result;
            }

            Long engineeringId = transferVO.getEngineeringId();
            for (JSONObject deviceDataItem : deviceData) {
                LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
                if (dataEndTime == null) {
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                Integer deviceId = deviceDataItem.getIntValue("device_id");
                Integer value = deviceDataItem.getInteger("qx");
                if (value == null) {
                    log.warn("设备{}的倾斜数据为空", deviceId);
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                TiltVO vo = new TiltVO();
                vo.setDataPacketID(generateDataPacketID());
                vo.setSensorID(deviceId);
                vo.setEngineeringID(engineeringId);
                vo.setPublishTime(getCurrentTime());
                vo.setDataEndTime(dataEndTime);
                vo.setSensorValue(value == 0 ? 0 : 1);

                try {
                    sendMqttMessage(EnvMonitorMqttTopic.TILT, vo, "倾斜信息");
                    result.put("successCount", result.get("successCount") + 1);
                } catch (Exception e) {
                    log.warn("设备{}的倾斜数据推送失败：{}", deviceId, e.getMessage());
                    result.put("failureCount", result.get("failureCount") + 1);
                }
            }

            log.info("倾斜数据推送完成，设备类型：{}，成功：{}，失败：{}",
                    deviceType, result.get("successCount"), result.get("failureCount"));

            return result;
        } catch (Exception e) {
            log.error("倾斜数据推送发生异常", e);
            result.put("failureCount", transferVO.getDevices().size());
            return result;
        }
    }

    /**
     * 发送裂缝数据（713）
     *
     * @return 推送结果，包含成功数和失败数
     **/
    public Map<String, Integer> sendCrackData(IotDataTransferVO transferVO) {
        Map<String, Integer> result = new HashMap<>();
        result.put("successCount", 0);
        result.put("failureCount", 0);

        if (!validateMqttGateway()) {
            return result;
        }

        try {
            List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
            Integer deviceType = transferVO.getDeviceType();
            Integer totalDevices = transferVO.getDevices().size();

            log.info("开始推送裂缝数据，设备类型：{}，设备数量：{}，获取到的数据条数：{}",
                    deviceType, totalDevices, deviceData.size());

            if (deviceData.isEmpty()) {
                log.warn("没有获取到裂缝数据！设备类型：{}", deviceType);
                result.put("failureCount", totalDevices);
                return result;
            }

            Long engineeringId = transferVO.getEngineeringId();
            for (JSONObject deviceDataItem : deviceData) {
                LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
                if (dataEndTime == null) {
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                Integer deviceId = deviceDataItem.getIntValue("device_id");
                Integer value = deviceDataItem.getInteger("cd");
                if (value == null) {
                    log.warn("设备{}的裂缝数据为空", deviceId);
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                CrackVO vo = new CrackVO();
                vo.setDataPacketID(generateDataPacketID());
                vo.setSensorID(deviceId);
                vo.setEngineeringID(engineeringId);
                vo.setPublishTime(getCurrentTime());
                vo.setDataEndTime(dataEndTime);
                vo.setSensorValue(value == 0 ? 0 : 1);

                try {
                    sendMqttMessage(EnvMonitorMqttTopic.CRACK, vo, "裂缝信息");
                    result.put("successCount", result.get("successCount") + 1);
                } catch (Exception e) {
                    log.warn("设备{}的裂缝数据推送失败：{}", deviceId, e.getMessage());
                    result.put("failureCount", result.get("failureCount") + 1);
                }
            }

            log.info("裂缝数据推送完成，设备类型：{}，成功：{}，失败：{}",
                    deviceType, result.get("successCount"), result.get("failureCount"));

            return result;
        } catch (Exception e) {
            log.error("裂缝数据推送发生异常", e);
            result.put("failureCount", transferVO.getDevices().size());
            return result;
        }
    }

    /**
     * 发送位移数据（714）
     *
     * @return 推送结果，包含成功数和失败数
     **/
    public Map<String, Integer> sendDeviationData(IotDataTransferVO transferVO) {
        Map<String, Integer> result = new HashMap<>();
        result.put("successCount", 0);
        result.put("failureCount", 0);

        if (!validateMqttGateway()) {
            return result;
        }

        try {
            List<JSONObject> deviceData = deviceDataQuery.getDeviceData(transferVO);
            Integer deviceType = transferVO.getDeviceType();
            Integer totalDevices = transferVO.getDevices().size();

            log.info("开始推送位移数据，设备类型：{}，设备数量：{}，获取到的数据条数：{}",
                    deviceType, totalDevices, deviceData.size());

            if (deviceData.isEmpty()) {
                log.warn("没有获取到位移数据！设备类型：{}", deviceType);
                result.put("failureCount", totalDevices);
                return result;
            }

            Long engineeringId = transferVO.getEngineeringId();
            for (JSONObject deviceDataItem : deviceData) {
                LocalDateTime dataEndTime = parseDataTime(deviceDataItem);
                if (dataEndTime == null) {
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                Integer deviceId = deviceDataItem.getIntValue("device_id");
                Integer value = deviceDataItem.getInteger("wy");
                if (value == null) {
                    log.warn("设备{}的位移数据为空", deviceId);
                    result.put("failureCount", result.get("failureCount") + 1);
                    continue;
                }

                DeviationVO vo = new DeviationVO();
                vo.setDataPacketID(generateDataPacketID());
                vo.setSensorID(deviceId);
                vo.setEngineeringID(engineeringId);
                vo.setPublishTime(getCurrentTime());
                vo.setDataEndTime(dataEndTime);
                vo.setSensorValue(value == 0 ? 0 : 1);

                try {
                    sendMqttMessage(EnvMonitorMqttTopic.DEVIATION, vo, "位移信息");
                    result.put("successCount", result.get("successCount") + 1);
                } catch (Exception e) {
                    log.warn("设备{}的位移数据推送失败：{}", deviceId, e.getMessage());
                    result.put("failureCount", result.get("failureCount") + 1);
                }
            }

            log.info("位移数据推送完成，设备类型：{}，成功：{}，失败：{}",
                    deviceType, result.get("successCount"), result.get("failureCount"));

            return result;
        } catch (Exception e) {
            log.error("位移数据推送发生异常", e);
            result.put("failureCount", transferVO.getDevices().size());
            return result;
        }
    }

    /**
     * 同步设备数据
     * @param tenantId 租户ID
     * @param engineeringId 工程ID
     * @param username MQTT用户名
     * @param password MQTT密码
     */
    public void synchronizeDeviceData(Integer tenantId, Long engineeringId, String username, String password) {
        // 参数校验
        if (engineeringId == null || username == null || password == null) {
            log.error("工程ID、MQTT用户名或密码不能为空");
            return;
        }

        // 查询租户下的所有产品类型
        List<String> deviceTypeList = getDeviceTypeListByTenant(tenantId);
        if (deviceTypeList.isEmpty()) {
            log.warn("租户{}不存在任何产品", tenantId);
            return;
        }

        // 查询设备列表
        List<DmpDevice> deviceList = getDeviceListByType(deviceTypeList);
        if (deviceList.isEmpty()) {
            log.warn("租户{}不存在任何设备", tenantId);
            return;
        }

        // 按设备类型分组
        Map<Integer, List<DmpDevice>> deviceTypeMap = deviceList.stream()
                .collect(Collectors.groupingBy(DmpDevice::getDeviceType));

        // 构建数据传输对象列表
        List<IotDataTransferVO> transferList = new ArrayList<>();
        deviceTypeMap.forEach((deviceType, devices) -> {
            IotDataTransferVO transferVO = new IotDataTransferVO();
            transferVO.setDeviceType(deviceType);
            transferVO.setDevices(devices);
            transferVO.setEngineeringId(engineeringId);
            transferList.add(transferVO);
        });

        // 按产品代码分组，构建ProductCode到uuid列表的映射，过滤掉产品代码为null的设备
        Map<String, List<String>> codeDeviceUuidsMap = deviceList.stream()
                .filter(device -> device.getProductCode() != null)
                .collect(Collectors.groupingBy(DmpDevice::getProductCode,
                        Collectors.mapping(DmpDevice::getDeviceUuid, Collectors.toList())));

        // 创建MQTT连接
        createMqttConnection(username, password);

        // 任务开始日志
        Integer totalDevices = deviceList.size();
        Integer totalProductTypes = codeDeviceUuidsMap.size();
        log.info("设备数据同步任务开始，租户ID：{}，工程ID：{}", tenantId, engineeringId);
        log.info("总共涉及产品类型数：{}个，产品代码为：{}", totalProductTypes, codeDeviceUuidsMap.keySet());
        log.info("总共需要推送设备数量：{}个，涉及设备类型数：{}个，设备类型为：{}",
                totalDevices, deviceTypeMap.size(), deviceTypeMap.keySet());

        // 记录每种设备类型的设备数量
        deviceTypeMap.forEach((deviceType, devices) -> {
            log.info("设备类型：{}，设备数量：{}", deviceType, devices.size());
        });

        // 按设备类型处理数据同步
        int totalSuccessCount = 0;
        int totalFailureCount = 0;

        for (IotDataTransferVO transferVO : transferList) {
            Integer deviceType = transferVO.getDeviceType();
            Map<String, Integer> result = new HashMap<>();

            switch (deviceType) {
                case 707:
                case 708:
                case 709:
                case 710:
                case 711:
                    result = sendEnvData(transferVO);
                    break;
                case 702:
                    result = sendWaterLeak(transferVO);
                    break;
                case 703:
                    result = sendPersonPresence(transferVO);
                    break;
                case 704:
                    result = sendElectricityLoad(transferVO);
                    break;
                case 712:
                    result = sendTiltData(transferVO);
                    break;
                case 713:
                    result = sendCrackData(transferVO);
                    break;
                case 714:
                    result = sendDeviationData(transferVO);
                    break;

                default:
                    log.debug("不支持的设备类型：{}", deviceType);
                    continue;
            }

            // 累加成功数和失败数
            totalSuccessCount += result.get("successCount");
            totalFailureCount += result.get("failureCount");
        }

        // 任务完成总结
        log.info("设备数据同步任务完成，租户ID：{}，工程ID：{}", tenantId, engineeringId);
        log.info("总共涉及产品类型数：{}个，产品代码为：{}", totalProductTypes, codeDeviceUuidsMap.keySet());
        log.info("总共推送设备数量：{}个，成功：{}个，失败：{}个",
                totalDevices, totalSuccessCount, totalFailureCount);
    }

    /**
     * 查询租户下的设备类型列表
     * @param tenantId 租户ID
     * @return 设备类型列表
     */
    private List<String> getDeviceTypeListByTenant(Integer tenantId) {
        LambdaQueryWrapper<DmpProduct> productQueryWrapper = new LambdaQueryWrapper<>();
        productQueryWrapper.select(DmpProduct::getProductCode)
                .eq(tenantId != null && tenantId > 0, DmpProduct::getTenantId, tenantId)
                .eq(DmpProduct::getDeleteFlag, 0);
        List<DmpProduct> productList = dmpProductMapper.selectList(productQueryWrapper);
        return productList.stream()
                .map(DmpProduct::getProductCode)
                .distinct() // 去重
                .collect(Collectors.toList());
    }

    /**
     * 根据设备类型列表查询设备
     * @param productCodeList 设备类型列表
     * @return 设备列表
     */
    private List<DmpDevice> getDeviceListByType(List<String> productCodeList) {
        LambdaQueryWrapper<DmpDevice> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.select(DmpDevice::getDeviceUuid, DmpDevice::getDeviceType, DmpDevice::getDeviceId, DmpDevice::getProductCode)
                .in(DmpDevice::getProductCode, productCodeList)
                .eq(DmpDevice::getDeleteFlag, 0)
                .notIn(DmpDevice::getServiceStatus, 3)
                .orderByAsc(DmpDevice::getProductCode);
        return dmpDeviceMapper.selectList(queryWrapper);
    }

    /**
     * 手动创建MQTT连接
     * @param username MQTT用户名
     * @param password MQTT密码
     */
    public void createMqttConnection(String username, String password) {
        try {
            // 使用注入的ApplicationContext获取已有的mqttGateway实例
            // 因为我们保留了@MessagingGateway注解，Spring会自动创建这个实例
            if (this.context == null) {
                throw new IllegalStateException("ApplicationContext未注入，无法获取MQTT Gateway");
            }

            // 1. 获取mqttGateway实例
            this.mqttGateway = this.context.getBean(MqttOutConfig.MqttGateway.class);
            if (this.mqttGateway == null) {
                throw new IllegalStateException("MQTT Gateway未找到，无法发送消息");
            }

            // 2. 获取现有的mqttClientFactory实例
            DefaultMqttPahoClientFactory mqttClientFactory = this.context.getBean(DefaultMqttPahoClientFactory.class);
            if (mqttClientFactory == null) {
                throw new IllegalStateException("MQTT Client Factory未找到，无法创建MQTT连接");
            }

            // 3. 创建并配置MqttConnectOptions
            MqttConnectOptions options = new MqttConnectOptions();
            options.setServerURIs(new String[]{MQTT_URL});
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setKeepAliveInterval(KEEP_ALIVE_INTERVAL);

            // 4. 更新mqttClientFactory的连接选项
            mqttClientFactory.setConnectionOptions(options);

            log.info("MQTT Gateway初始化成功，用户名：{}", username);
            log.info("MQTT连接配置完成，服务器地址：{}，客户端ID：mqttx-{}", MQTT_URL, username);
        } catch (Exception e) {
            log.error("初始化MQTT连接失败: {}", e.getMessage(), e);
            throw new RuntimeException("初始化MQTT连接失败", e);
        }
    }

    /**
     * 验证MQTT网关是否初始化
     * @return 是否初始化
     */
    private boolean validateMqttGateway() {
        if (mqttGateway == null) {
            log.warn("MQTT Gateway未初始化，无法发送消息");
            return false;
        }
        return true;
    }

    /**
     * 解析数据时间
     * @param deviceDataItem 设备数据
     * @return 解析后的时间，如果解析失败返回null
     */
    private LocalDateTime parseDataTime(JSONObject deviceDataItem) {
        log.warn("解析的json{}", deviceDataItem.toString());
        Long dataTime = deviceDataItem.getLong("realtime");
        if (dataTime == null) {
            log.warn("设备{}的time为空", deviceDataItem.getString("device_id"));
            return null;
        }
        return LocalDateTime.ofInstant(Instant.ofEpochMilli(dataTime), ZoneId.systemDefault());
    }

    /**
     * 发送MQTT消息
     * @param topicEnum 主题枚举
     * @param vo 消息对象
     * @param messageType 消息类型描述
     */
    private void sendMqttMessage(EnvMonitorMqttTopic topicEnum, Object vo, String messageType) {
        String json = JSON.toJSONString(vo);
        String topic = topicEnum.getTopic();
        // 不再记录每条数据的详情，只记录发送操作
        mqttGateway.sendToMqtt(topic, json);
    }

    public void allData(Long engineeringId, String username, String password) {
        Integer tenantId = 0;
        synchronizeDeviceData(tenantId, engineeringId, username, password);
    }
}