package com.xzzn.quartz.task;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.serotonin.modbus4j.ModbusMaster;
import com.xzzn.common.constant.RedisKeyConstants;
import com.xzzn.common.core.modbus.ModbusProcessor;
import com.xzzn.common.core.modbus.domain.DeviceConfig;
import com.xzzn.common.core.modbus.domain.TagConfig;
import com.xzzn.common.core.redis.RedisCache;
import com.xzzn.common.enums.DeviceRunningStatus;
import com.xzzn.common.utils.StringUtils;
import com.xzzn.ems.domain.EmsDevicesSetting;
import com.xzzn.ems.domain.EmsPointConfig;
import com.xzzn.ems.mapper.EmsDevicesSettingMapper;
import com.xzzn.ems.mapper.EmsPointConfigMapper;
import com.xzzn.ems.service.IEmsAlarmRecordsService;
import com.xzzn.ems.service.impl.DeviceDataProcessServiceImpl;
import com.xzzn.framework.web.service.MqttPublisher;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import javax.annotation.Resource;

import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

/**
 * 轮询设备-通过modbus协议读取数据
 */
@Component("modbusPoller")
public class ModbusPoller {
    private static final Logger log = LoggerFactory.getLogger(ModbusPoller.class);
    private static final int SITE_DEVICE_OFFLINE_THRESHOLD = 6;

    private final Map<String, Integer> deviceFailureCounts = new ConcurrentHashMap<>();
    private final AtomicBoolean polling = new AtomicBoolean(false);

    @Resource(name = "modbusExecutor")
    private ExecutorService modbusExecutor;

    @Autowired
    private ModbusProcessor modbusProcessor;
    @Autowired
    private IEmsAlarmRecordsService iEmsAlarmRecordsService;

    @Autowired
    private DeviceDataProcessServiceImpl deviceDataProcessServiceImpl;
    @Autowired
    private EmsDevicesSettingMapper emsDevicesSettingMapper;
    @Autowired
    private EmsPointConfigMapper emsPointConfigMapper;
    @Autowired
    private RedisCache redisCache;
    @Autowired
    private MqttPublisher mqttPublisher;

    @Value("${mqtt.topic}")
    private String topic;

    public void pollAllDevices() {
        if (!polling.compareAndSet(false, true)) {
            log.warn("上一次轮询尚未完成，本次轮询跳过");
            return;
        }
        List<PollingTask> pollingTasks = buildPollingTasks();
        if (CollectionUtils.isEmpty(pollingTasks)) {
            log.warn("未查询到可用的Modbus采集点位配置，跳过本轮轮询");
            polling.set(false);
            return;
        }

        // 按主机IP分组，同一网关串行访问，避免连接抖动
        Map<String, List<PollingTask>> groupedByHost = pollingTasks.stream()
                .collect(Collectors.groupingBy(
                        task -> task.getDeviceConfig().getHost(),
                        HashMap::new,
                        Collectors.toList()));

        Future<?> future = modbusExecutor.submit(() -> {
            for (Map.Entry<String, List<PollingTask>> entry : groupedByHost.entrySet()) {
                String hostKey = entry.getKey();
                List<PollingTask> tasks = entry.getValue();
                for (PollingTask task : tasks) {
                    try {
                        scheduledStart(task.getSiteId(), task.getDeviceConfig());
                        // 每次读取后等待200ms，给Modbus网关足够的处理时间
                        Thread.sleep(200);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        log.warn("Modbus轮询被中断");
                        return;
                    } catch (Exception e) {
                        log.error("采集设备数据异常: siteId={}, deviceId={}",
                                task.getSiteId(), task.getDeviceConfig().getDeviceNumber(), e);
                    }
                }
                log.info("采集设备数据{}轮询任务执行完成", hostKey);
            }
        });
        try {
            future.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("Modbus轮询任务等待中断");
        } catch (Exception e) {
            log.error("Modbus轮询任务执行异常", e);
        } finally {
            polling.set(false);
        }
    }

    private List<PollingTask> buildPollingTasks() {
        List<EmsPointConfig> pointConfigs = emsPointConfigMapper.selectModbusCollectPointConfigs(null);
        if (CollectionUtils.isEmpty(pointConfigs)) {
            return Collections.emptyList();
        }

        List<EmsDevicesSetting> allDevices = emsDevicesSettingMapper.selectEmsDevicesSettingList(null);
        Map<String, EmsDevicesSetting> deviceMap = allDevices.stream()
                .filter(Objects::nonNull)
                .filter(device -> StringUtils.isNoneBlank(device.getSiteId(), device.getDeviceId()))
                .collect(Collectors.toMap(
                        this::buildSiteDeviceKey,
                        device -> device,
                        (left, right) -> left));

        Map<String, List<EmsPointConfig>> pointsByDevice = pointConfigs.stream()
                .filter(Objects::nonNull)
                .filter(point -> StringUtils.isNoneBlank(point.getSiteId(), point.getDeviceId()))
                .collect(Collectors.groupingBy(
                        point -> point.getSiteId() + "_" + point.getDeviceId(),
                        HashMap::new,
                        Collectors.toList()));

        List<PollingTask> tasks = new ArrayList<>();
        for (Map.Entry<String, List<EmsPointConfig>> entry : pointsByDevice.entrySet()) {
            String siteDeviceKey = entry.getKey();
            EmsDevicesSetting device = deviceMap.get(siteDeviceKey);
            if (device == null) {
                log.warn("未找到设备连接配置，跳过采集: key={}", siteDeviceKey);
                continue;
            }
            DeviceConfig deviceConfig = buildDeviceConfig(device, entry.getValue());
            if (deviceConfig == null) {
                continue;
            }
            tasks.add(new PollingTask(device.getSiteId(), deviceConfig));
        }
        return tasks;
    }

    private DeviceConfig buildDeviceConfig(EmsDevicesSetting device, List<EmsPointConfig> pointConfigs) {
        if (device == null || CollectionUtils.isEmpty(pointConfigs)) {
            return null;
        }
        if (StringUtils.isBlank(device.getIpAddress()) || device.getIpPort() == null || device.getSlaveId() == null) {
            log.warn("设备连接参数不完整，跳过采集: siteId={}, deviceId={}", device.getSiteId(), device.getDeviceId());
            return null;
        }
        List<TagConfig> tags = pointConfigs.stream()
                .sorted(Comparator.comparing(point -> point.getModbusReadOrder() == null ? 0 : point.getModbusReadOrder()))
                .map(this::toTagConfig)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(tags)) {
            log.warn("设备无有效Modbus点位配置，跳过采集: siteId={}, deviceId={}", device.getSiteId(), device.getDeviceId());
            return null;
        }

        DeviceConfig deviceConfig = new DeviceConfig();
        deviceConfig.setEnabled(true);
        deviceConfig.setDeviceName(device.getDeviceName());
        deviceConfig.setDeviceNumber(device.getDeviceId());
        deviceConfig.setHost(device.getIpAddress());
        deviceConfig.setPort(device.getIpPort().intValue());
        deviceConfig.setSlaveId(device.getSlaveId().intValue());
        deviceConfig.setTags(tags);
        return deviceConfig;
    }

    private TagConfig toTagConfig(EmsPointConfig pointConfig) {
        if (pointConfig == null) {
            return null;
        }
        if (StringUtils.isBlank(pointConfig.getDataKey())) {
            return null;
        }
        String address = normalizeAddress(pointConfig.getRegisterAddress(), pointConfig.getModbusRegisterType());
        if (StringUtils.isBlank(address)) {
            return null;
        }
        if (StringUtils.isBlank(pointConfig.getModbusDataType())) {
            return null;
        }

        TagConfig tag = new TagConfig();
        tag.setKey(pointConfig.getDataKey().trim());
        tag.setAddress(address);
        tag.setDataType(pointConfig.getModbusDataType().trim());
        tag.setA(pointConfig.getDataA() == null ? 0F : pointConfig.getDataA().floatValue());
        tag.setK(pointConfig.getDataK() == null ? 1F : pointConfig.getDataK().floatValue());
        tag.setB(pointConfig.getDataB() == null ? 0F : pointConfig.getDataB().floatValue());
        tag.setBit(pointConfig.getDataBit());
        return tag;
    }

    private String normalizeAddress(String registerAddress, String registerType) {
        if (StringUtils.isBlank(registerAddress)) {
            return null;
        }
        String normalizedAddress = registerAddress.trim();
        if (!normalizedAddress.chars().allMatch(Character::isDigit)) {
            log.warn("寄存器地址必须为数字，当前值: {}", normalizedAddress);
            return null;
        }
        if (normalizedAddress.length() > 1) {
            char first = normalizedAddress.charAt(0);
            if (first >= '0' && first <= '4') {
                return normalizedAddress;
            }
        }
        return getRegisterPrefix(registerType) + normalizedAddress;
    }

    private String getRegisterPrefix(String registerType) {
        String normalized = StringUtils.defaultString(registerType).trim().toUpperCase();
        switch (normalized) {
            case "COIL":
                return "0";
            case "DISCRETE_INPUT":
                return "1";
            case "INPUT_REGISTER":
                return "3";
            case "HOLDING_REGISTER":
            default:
                return "4";
        }
    }

    private String buildSiteDeviceKey(EmsDevicesSetting device) {
        return device.getSiteId() + "_" + device.getDeviceId();
    }

    public void scheduledStart(String siteId, DeviceConfig config) {
        if (config.isEnabled()) {
            log.info("Reading data from devices: {}", config.getDeviceName());
            
            // 带重试的读取，最多重试2次
            Map<String, Object> data = readWithRetry(config, 2);
            
            List<String> rawValuEmptyList = new ArrayList<>();
            // 在这里处理采集到的数据空
            config.getTags().forEach(tag -> {
                Object rawValue = data.get(tag.getKey());
                if (rawValue != null) {
                    float value = 0;
                    if (rawValue instanceof Number) {
                        value = ((Number) rawValue).floatValue(); // 安全地转换为 float
                    } else {
                        log.error("tag:{},无法将数据转换为数字: {}", tag.getKey(), rawValue);
                    }
                    value = tag.getA() * value * value + tag.getK() * value + tag.getB();

                    int intValue = (int) value;
                    if (tag.getBit() != null) {
                        log.info("tag:{},bit:{},value:{}", tag.getKey(), tag.getBit(), value);
                        String binary = Integer.toBinaryString(intValue);
                        data.put(tag.getKey(), binary);
                    } else {
                        data.put(tag.getKey(), value);
                    }
                } else {
//                    data.put(tag.getKey(), rawValue);
//                    log.warn("tag:{},数据为空: {}", tag.getKey(), rawValue);
                    rawValuEmptyList.add("tag: " + tag.getKey() + ",数据为空: " + rawValue);
                }
            });
            if (!rawValuEmptyList.isEmpty()) {
                log.warn("设备 {} 数据为空: {}", config.getDeviceName(), JSON.toJSONString(rawValuEmptyList));
            }
            log.info("Data from {}: {}", config.getDeviceName(), data);
            String deviceNumber = config.getDeviceNumber();
            //处理数据并发送MQTT消息、保存Redis数据和数据入库
            processingData(siteId, data, deviceNumber);
        }
    }

    /**
     * 带重试的读取方法
     */
    private Map<String, Object> readWithRetry(DeviceConfig config, int maxRetries) {
        Map<String, Object> data = new HashMap<>();
        
        for (int attempt = 0; attempt <= maxRetries; attempt++) {
            try {
                ModbusMaster master = modbusProcessor.borrowMaster(config);
                data = modbusProcessor.readDataFromDevice(config, master);
                
                // 如果读取成功（有数据），直接返回
                if (!data.isEmpty()) {
                    if (attempt > 0) {
                        log.info("设备 {} 第 {} 次重试成功", config.getDeviceName(), attempt);
                    }
                    return data;
                }
                
                // 读取返回空数据，等待后重试
                if (attempt < maxRetries) {
                    log.warn("设备 {} 读取返回空数据，等待1秒后重试 ({}/{})", 
                            config.getDeviceName(), attempt + 1, maxRetries);
                    Thread.sleep(1000);
                }
            } catch (Exception e) {
                log.error("设备 {} 读取异常 ({}/{}): {}", 
                        config.getDeviceName(), attempt + 1, maxRetries, e.getMessage());
                
                if (attempt < maxRetries) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }
        
        // 所有重试都失败
        log.error("设备 {} 读取失败，已重试 {} 次", config.getDeviceName(), maxRetries);
        return data;
    }

    private void processingData(String siteId, Map<String, Object> data, String deviceNumber) {
        String siteDeviceKey = siteId + "_" + deviceNumber;
        if (CollectionUtils.isEmpty(data)) {
            // 增加失败计数
            int failureCount = deviceFailureCounts.getOrDefault(siteDeviceKey, 0) + 1;
            deviceFailureCounts.put(siteDeviceKey, failureCount);

            log.warn("设备 {} 数据读取失败，当前连续失败次数: {}", siteDeviceKey, failureCount);

            // 连续6次失败触发报警
            if (failureCount >= SITE_DEVICE_OFFLINE_THRESHOLD) {
                addDeviceOfflineRecord(siteId, deviceNumber);
                log.error("设备 {} 连续 {} 次未读取到数据，触发报警", siteDeviceKey, failureCount);
            }
            return;
        }

        // 数据读取成功，重置计数器
        deviceFailureCounts.remove(siteDeviceKey);
        // 读取到数据后告警自恢复
        deleteDeviceOfflineRecord(siteId, deviceNumber);

        // 发送MQTT消息、保存Redis数据和数据入库
        Long timestamp = System.currentTimeMillis();
        JSONObject json = new JSONObject();
        json.put("Data", data);
        json.put("timestamp", timestamp);
        json.put("Device", deviceNumber);
        if (shouldSendMqttOnChange(siteId, deviceNumber, data)) {
            sendMqttMsg(json);
        } else {
            sendMqttHeartbeat(deviceNumber, timestamp);
            log.info("设备 {} 数据无变化，已发送心跳MQTT", siteDeviceKey);
        }
        saveRedisData(siteId, json, deviceNumber);
        saveDataToDatabase(siteId, data, deviceNumber, timestamp);
    }

    /**
     * 逢变上送：仅当Data发生变化时才发送MQTT
     */
    private boolean shouldSendMqttOnChange(String siteId, String deviceNumber, Map<String, Object> currentData) {
        JSONObject lastPayload = redisCache.getCacheObject(RedisKeyConstants.ORIGINAL_MQTT_DATA + siteId + "_" + deviceNumber);
        if (lastPayload == null) {
            return true;
        }

        Object lastDataObj = lastPayload.get("Data");
        if (lastDataObj == null) {
            return true;
        }

        JSONObject lastData = JSON.parseObject(JSON.toJSONString(lastDataObj));
        JSONObject currentDataJson = JSON.parseObject(JSON.toJSONString(currentData));
        return !Objects.equals(lastData, currentDataJson);
    }

    private void sendMqttHeartbeat(String deviceNumber, Long timestamp) {
        JSONObject heartbeat = new JSONObject();
        heartbeat.put("Device", deviceNumber);
        heartbeat.put("timestamp", timestamp);
        heartbeat.put("Heartbeat", 1);
        heartbeat.put("Data", new JSONObject());
        sendMqttMsg(heartbeat);
    }

    public void sendMqttMsg(JSONObject json) {
        try {
            mqttPublisher.publish(topic, Collections.singletonList(json).toString(), 0);
        } catch (MqttException e) {
            log.error("MQTT消息发布失败: {}, reason code: {}", json.toJSONString(), e.getReasonCode() ,e);
        }
        log.info("已发送数据: {}", json.toJSONString());
    }


    public void saveRedisData(String siteId, JSONObject obj, String deviceNumber) {
        try {
            // 存放mqtt原始每个设备最晚一次数据，便于后面点位获取数据
            redisCache.setCacheObject(RedisKeyConstants.ORIGINAL_MQTT_DATA + siteId + "_" + deviceNumber, obj);
            // 存放每次同步数据，失效时间（同同步时间）-用于判断是否正常同步数据和保护策略查询
            redisCache.setCacheObject(RedisKeyConstants.SYNC_DATA + siteId + "_" + deviceNumber, obj, 1, TimeUnit.MINUTES);
            log.info("数据已成功存储在Redis: {}", deviceNumber);
        } catch (Exception e) {
            log.error("无法在设备的Redis中存储数据: {}", deviceNumber, e);
        }
    }

    private void saveDataToDatabase(String siteId, Map<String, Object> data, String deviceNumber, Long timestamp) {
        JSONObject payload = new JSONObject();
        payload.put("Device", deviceNumber);
        payload.put("Data", JSON.toJSONString(data));
        payload.put("timestamp", timestamp);
        deviceDataProcessServiceImpl.handleDeviceData(Collections.singletonList(payload).toString(), siteId);
    }

    //处理设备连接失败的情况，更新设备状态为离线，添加报警记录
    private void addDeviceOfflineRecord(String siteId, String deviceNumber) {
        updateDeviceStatus(siteId, deviceNumber, DeviceRunningStatus.OFFLINE.getCode());
        iEmsAlarmRecordsService.addDeviceOfflineRecord(siteId, deviceNumber);
    }

    //处理设备读取到数据的情况，更新设备状态为在线，报警记录自恢复
    private void deleteDeviceOfflineRecord(String siteId, String deviceNumber) {
        updateDeviceStatus(siteId, deviceNumber, DeviceRunningStatus.ONLINE.getCode());
        iEmsAlarmRecordsService.deleteDeviceOfflineRecord(siteId, deviceNumber);
    }

    // 更新设备状态为在线或离线
    private void updateDeviceStatus(String siteId, String deviceNumber, String deviceStatus) {
        EmsDevicesSetting emsDevicesSetting = emsDevicesSettingMapper.getDeviceBySiteAndDeviceId(deviceNumber,  siteId);
        if (emsDevicesSetting != null && !Objects.equals(emsDevicesSetting.getDeviceStatus(), deviceStatus)) {
            emsDevicesSetting.setDeviceStatus(deviceStatus);
            emsDevicesSettingMapper.updateEmsDevicesSetting(emsDevicesSetting);
        }
    }

    private static class PollingTask {
        private final String siteId;
        private final DeviceConfig deviceConfig;

        private PollingTask(String siteId, DeviceConfig deviceConfig) {
            this.siteId = siteId;
            this.deviceConfig = deviceConfig;
        }

        public String getSiteId() {
            return siteId;
        }

        public DeviceConfig getDeviceConfig() {
            return deviceConfig;
        }
    }

}
