package com.tongran.rocketmq.handler;

import com.alibaba.fastjson.JSONObject;
import com.tongran.common.core.constant.SecurityConstants;
import com.tongran.common.core.domain.R;
import com.tongran.common.core.enums.MsgEnum;
import com.tongran.common.core.utils.DateUtils;
import com.tongran.common.core.utils.StringUtils;
import com.tongran.common.security.utils.SecurityUtils;
import com.tongran.rocketmq.domain.*;
import com.tongran.rocketmq.domain.vo.*;
import com.tongran.rocketmq.enums.AlarmTypeEnum;
import com.tongran.rocketmq.enums.PushMethodEnum;
import com.tongran.rocketmq.model.ProducerMode;
import com.tongran.rocketmq.producer.MessageProducer;
import com.tongran.rocketmq.service.*;
import com.tongran.rocketmq.utils.DataProcessUtil;
import com.tongran.rocketmq.utils.JsonDataParser;
import com.tongran.rocketmq.utils.WeChatWorkBot;
import com.tongran.system.api.RemoteRevenueConfigService;
import com.tongran.system.api.domain.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 设备消息处理器
 */
@Slf4j
@Component
@EnableScheduling
public class MessageHandler {

    private final Map<String, Consumer<DeviceMessage>> messageHandlers = new HashMap<>();
    // 硬盘上报状态
    String DISK_COUNT_PREFIX = "disk:count:";
    // 心跳状态
    private static final String HEARTBEAT_STATUS_PREFIX = "heartbeat:status:";
    // 心跳时间
    private static final String HEARTBEAT_TIME_PREFIX = "heartbeat:time:";
    // 心跳告警
    private static final String HEARTBEAT_ALERT_PREFIX = "heartbeat:alert:";
    String HEARTBEAT_RECOVERY_COUNT_PREFIX = "heartbeat:recovery:count:";

    String HEARTBEAT_COUNT_PREFIX = "heartbeat:count:";

    // Redis key定义
    private static final String USED_PORTS_KEY = "frpc:used_ports";
    private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时


    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    private  IInitialBandwidthTrafficService initialBandwidthTrafficService;
    @Autowired
    private  RemoteRevenueConfigService remoteRevenueConfigService;
    @Autowired
    private IInitialDockerInfoService initialDockerInfoService;
    @Autowired
    private IInitialCpuInfoService initialCpuInfoService;
    @Autowired
    private IInitialDiskInfoService initialDiskInfoService;
    @Autowired
    private IInitialMemoryInfoService initialMemoryInfoService;
    @Autowired
    private IInitialMountPointInfoService initialMountPointInfoService;
    @Autowired
    private IInitialHeartbeatListenLogService initialHeartbeatListenLog;
    @Autowired
    private IInitialSystemOtherCollectDataService iInitialSystemOtherCollectDataService;
    @Autowired
    private IRmResourceRemoteService rmResourceRemoteService;
    @Autowired
    private IRmAgentManagementService rmAgentManagementService;
    @Autowired
    private DataProcessUtil dataProcessUtil;
    @Autowired
    private IRmNetworkInterfaceService rmNetworkInterfaceService;
    @Autowired
    private IRmNetworkInterfaceChildService rmNetworkInterfaceChildService;
    @Autowired
    private IRmMonitorPolicyService rmMonitorPolicyService;
    @Autowired
    private IRmDeploymentPolicyService rmDeploymentPolicyService;
    @Autowired
    private IInitialBandwidthTrafficTempService initialBandwidthTrafficTempService;
    @Autowired
    private IRmAlarmLogService rmAlarmLogService;
    @Autowired
    private IRmAlarmPushConfigService rmAlarmPushConfigService;
    @Autowired
    private IRmFrpcConfigManageService rmFrpcConfigManageService;
    @Autowired
    private IRmPppoeConfigSubService rmPppoeConfigSubService;
    @Autowired
    private IAllDiskNameService allDiskNameService;
    @Autowired
    private ProducerMode producerMode;
    @Autowired
    private RedissonClient redissonClient;


    /**
     * 初始化处理器映射
     */
    @PostConstruct
    public void init() {
        registerHandler(MsgEnum.执行脚本策略应答.getValue(), this::handleScriptRspMessage);
        registerHandler(MsgEnum.Agent版本更新应答.getValue(), this::handleAgentUpdateRspMessage);

        // 其他类型消息可以单独注册处理器
        registerHandler(MsgEnum.注册.getValue(), this::handleRegisterMessage);
        registerHandler(MsgEnum.获取最新策略.getValue(), this::handleNewPolicyMessage);
        registerHandler(MsgEnum.CPU上报.getValue(), this::handleCpuMessage);
        registerHandler(MsgEnum.磁盘上报.getValue(), this::handleDiskMessage);
        registerHandler(MsgEnum.容器上报.getValue(), this::handleDockerMessage);
        registerHandler(MsgEnum.内存上报.getValue(), this::handleMemoryMessage);
        registerHandler(MsgEnum.网络上报.getValue(), this::handleNetMessage);
        registerHandler(MsgEnum.网络上报重试.getValue(), this::handleNetRecoverMessage);
        registerHandler(MsgEnum.挂载上报.getValue(), this::handleMountPointMessage);
        registerHandler(MsgEnum.系统其他上报.getValue(), this::handleOtherSystemMessage);
        registerHandler(MsgEnum.心跳上报.getValue(), this::handleHeartbeatMessage);
        registerHandler(MsgEnum.多公网IP探测.getValue(), this::handleNetWorkDelectMessage);
        registerHandler(MsgEnum.修改frp配置文件应答.getValue(), this::handleUpdateFrpMessage);
        registerHandler(MsgEnum.macvlan状态上报.getValue(), this::handleMacvlanMessage);
        registerHandler(MsgEnum.iops结果上报.getValue(), this::handleIopsResultMessage);
        registerHandler(MsgEnum.tcpdump结果上报.getValue(), this::handleTcpdumpResultMessage);
    }

    private void handleTcpdumpResultMessage(DeviceMessage message) {
        List<TcpdumpVo> tcpdumpVoList = JsonDataParser.parseJsonData(message.getData(), TcpdumpVo.class);
        if(tcpdumpVoList != null && !tcpdumpVoList.isEmpty()){
            for (TcpdumpVo tcpdumpVo : tcpdumpVoList) {
                String ip = tcpdumpVo.getIp();
                Integer count = tcpdumpVo.getCount();
                String type = checkIPVersion(ip);
                // 查询IP地址归属地信息
                Map<String, String> locationInfo = queryIpLocation(ip);
                if (locationInfo != null) {
                    String operator = locationInfo.get("operator"); // 运营商
                    String province = locationInfo.get("province"); // 省份

                    System.out.println("IP: " + ip +
                            ", 运营商: " + operator +
                            ", 省份: " + province +
                            ", 抓包数量: " + count);

                    // 设置到对象中（如果TcpdumpVo有这些字段）
                    // tcpdumpVo.setOperator(operator);
                    // tcpdumpVo.setProvince(province);
                }
            }
        }
    }

    private void handleIopsResultMessage(DeviceMessage message) {
        List<DiskIopsResultVo> diskIopsResultVoList = JsonDataParser.parseJsonData(message.getData(), DiskIopsResultVo.class);
        if(diskIopsResultVoList != null && !diskIopsResultVoList.isEmpty()){
            DiskIopsResultVo diskIopsResultVo = diskIopsResultVoList.get(0);
            // 时间戳转换
            long timestamp = diskIopsResultVo.getTimestamp();
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            AllDiskName allDiskName = new AllDiskName();
            allDiskName.setClientId(message.getClientId());
            allDiskName.setName(diskIopsResultVo.getName());
            allDiskName.setReadIops(diskIopsResultVo.getReadIops());
            allDiskName.setWriteIops(diskIopsResultVo.getWriteIops());
            allDiskNameService.updateAllDiskName(allDiskName);
        }
    }

    private void handleMacvlanMessage(DeviceMessage message) {
        List<MacVlanRspVo> macVlanRspVoList = JsonDataParser.parseJsonData(message.getData(), MacVlanRspVo.class);
        if(!macVlanRspVoList.isEmpty()){
            MacVlanRspVo macVlanRspVo = macVlanRspVoList.get(0);
            // 时间戳转换
            long timestamp = macVlanRspVo.getTimestamp();
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            List<MacVlanRspVo.MacVlanVo> list = macVlanRspVo.getMacVlans();
            for (MacVlanRspVo.MacVlanVo macVlanVo : list) {
                RmPppoeConfigSub rmPppoeConfigSub = new RmPppoeConfigSub();
                rmPppoeConfigSub.setVlanId(Long.valueOf(macVlanVo.getVlanId()));
                rmPppoeConfigSub.setSerialNumber(Long.valueOf(macVlanVo.getMid()));
                rmPppoeConfigSub.setClientId(message.getClientId());
                rmPppoeConfigSub.setStatus(macVlanVo.getStatus());
                rmPppoeConfigSub.setUpdateTime(createTime);
                rmPppoeConfigSubService.updateRmPppoeConfigSubByVlan(rmPppoeConfigSub);
            }
        }
    }

    private void handleUpdateFrpMessage(DeviceMessage message) {
        List<FrpRspVo> rspVoList = JsonDataParser.parseJsonData(message.getData(), FrpRspVo.class);
        if (!rspVoList.isEmpty()) {
            FrpRspVo rsp = rspVoList.get(0);
            // 时间戳转换
            long timestamp = rsp.getTimestamp();
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            if(rsp.getResCode() == 1) {
                // 运行成功 修改相关信息
                RmFrpcConfigManage frpcData = new RmFrpcConfigManage();
                frpcData.setClientId(message.getClientId());
                frpcData.setFrpcName(rsp.getName());
                frpcData.setFrpcStatus(rsp.getIsRunning());
                frpcData.setFrpcLocalPort(rsp.getLocalPort());
                frpcData.setFrpcLocalIp(rsp.getLocalIP());
                frpcData.setFrpcCreateTime(createTime);
                frpcData.setCreateTime(DateUtils.getNowDate());
                frpcData.setFrpcConnectionStatus(2L);
                frpcData.setFrpcRemotePort(rsp.getRemotePort());
                frpcData.setFrpcType(rsp.getType());
                frpcData.setFrpcServerAddr(rsp.getServerAddr());
                frpcData.setFrpcServerPort(rsp.getServerPort());
                rmFrpcConfigManageService.insertRmFrpcConfigManage(frpcData);
            } else if(rsp.getResCode() == 2){
                RmFrpcConfigManage configManage = new RmFrpcConfigManage();
                configManage.setFrpcRemotePort(rsp.getRemotePort());
                rmFrpcConfigManageService.addFrpConnect(configManage);
            } else if(rsp.getResCode() == 3){
                // frpc状态上报
                RmFrpcConfigManage frpcData = new RmFrpcConfigManage();
                frpcData.setClientId(message.getClientId());
                frpcData.setFrpcName(rsp.getName());
                frpcData.setFrpcType(rsp.getType());
                frpcData.setFrpcStatus(rsp.getIsRunning());
                frpcData.setFrpcLocalPort(rsp.getLocalPort());
                frpcData.setFrpcLocalIp(rsp.getLocalIP());
                frpcData.setFrpcRemotePort(rsp.getRemotePort());
                frpcData.setFrpcServerAddr(rsp.getServerAddr());
                frpcData.setFrpcServerPort(rsp.getServerPort());
                frpcData.setUpdateTime(DateUtils.getNowDate());
                rmFrpcConfigManageService.insertRmFrpcConfigManage(frpcData);
            } else{
                if(rsp.getRemotePort() != null){
                    // 生成失败 ，释放端口资源
                    removeUsedPort(Integer.parseInt(rsp.getRemotePort()));
                }
                log.error("生成连接失败：{}",rsp.getResMsg());
            }
        }
    }

    /**
     * 添加端口到已用集合
     */
    public boolean addUsedPort(int port) {
        Long result = redisTemplate.opsForSet().add(USED_PORTS_KEY, String.valueOf(port));
        return result != null && result > 0;
    }

    /**
     * 从已用集合中移除端口
     */
    public boolean removeUsedPort(int port) {
        Long result = redisTemplate.opsForSet().remove(USED_PORTS_KEY, String.valueOf(port));
        return result != null && result > 0;
    }
    /**
     * 保存网卡信息
     * @param message
     */
    private void handleNetWorkDelectMessage(DeviceMessage message) {
        List<RegisterMsgVo> interfaces = JsonDataParser.parseJsonData(message.getData(), RegisterMsgVo.class);
        if(!interfaces.isEmpty()) {
            String clientId = message.getClientId();
            RegisterMsgVo registerMsg = interfaces.get(0);
            // 处理网卡信息
            processNetworkInterfaces(registerMsg, clientId, false);
        }
    }

    /**
     * 获取最新策略
     * @param deviceMessage
     */
    private void handleNewPolicyMessage(DeviceMessage deviceMessage) {
        List<RegisterMsgVo> interfaces = JsonDataParser.parseJsonData(deviceMessage.getData(), RegisterMsgVo.class);
        if(!interfaces.isEmpty()) {
            RegisterMsgVo registerMsgVo = interfaces.get(0);
            String clientId = registerMsgVo.getClientId();
            // 如果未下发监控策略，下发
            rmMonitorPolicyService.issuePolicyMsgByClientId(clientId);
            // 如果未下发服务器脚本策略，下发
            rmDeploymentPolicyService.issueDeployPolicyMsgByClientId(clientId);
            // 如果路由有变化，更新路由信息
            rmNetworkInterfaceService.updateRouteMsg(clientId);
            // 如果业务网卡名称有变化，更新防火墙策略
            rmNetworkInterfaceService.issueNetName(clientId);
        }
    }

    /**
     * 服务器注册
     * @param message
     */
    private void handleRegisterMessage(DeviceMessage message) {
        List<RegisterMsgVo> interfaces = JsonDataParser.parseJsonData(message.getData(), RegisterMsgVo.class);
        if(!interfaces.isEmpty()) {
            String clientId = message.getClientId();
            RegisterMsgVo registerMsg = interfaces.get(0);
            // 自动注册服务器信息
            RmRegisterMsgRemote rmRegisterMsgRemote = new RmRegisterMsgRemote();
            BeanUtils.copyProperties(registerMsg, rmRegisterMsgRemote);
            int rows = remoteRevenueConfigService.innerAddRegist(rmRegisterMsgRemote, SecurityConstants.INNER).getData();
            if(rows == 2){
                // 注册成功，下发优先级为0的策略
                rmMonitorPolicyService.issueDefaultPolicyByClientId(message.getClientId());
                // agent更新表插入数据
                // 存储更新结果
                // agent更新结果存储
                RmAgentManagement query = new RmAgentManagement();
                query.setClientId(clientId);
                List<RmAgentManagement> agentManagements = rmAgentManagementService.selectRmAgentManagementList(query);
                if(agentManagements == null || agentManagements.isEmpty()) {
                    RmAgentManagement insertAgentUpdateData = new RmAgentManagement();
                    insertAgentUpdateData.setClientId(clientId);
                    rmAgentManagementService.insertRmAgentManagement(insertAgentUpdateData);
                }
            }
            // 处理网卡信息
            processNetworkInterfaces(registerMsg, clientId, true);
        }
    }

    /**
     * agent更新响应
     * @param message
     */
    private void handleAgentUpdateRspMessage(DeviceMessage message) {
        List<RspVo> rspVoList = JsonDataParser.parseJsonData(message.getData(), RspVo.class);
        if (!rspVoList.isEmpty()) {
            RspVo rsp = rspVoList.get(0);
            if(rsp.getResCode() == 1){
                RmAgentManagement rmAgentManagement = new RmAgentManagement();
                rmAgentManagement.setClientId(message.getClientId());
                rmAgentManagement.setLastUpdateResult("1");
                rmAgentManagement.setLastUpdateTime(DateUtils.getNowDate());
                rmAgentManagementService.updateRmAgentManagementByHardwareSn(rmAgentManagement);
            }else{
                RmAgentManagement rmAgentManagement = new RmAgentManagement();
                rmAgentManagement.setClientId(message.getClientId());
                rmAgentManagement.setLastUpdateResult("0");
                rmAgentManagement.setLastUpdateTime(DateUtils.getNowDate());
                rmAgentManagementService.updateRmAgentManagementByHardwareSn(rmAgentManagement);
            }
        }
    }

    /**
     * 注册消息处理器
     */
    private void registerHandler(String dataType, Consumer<DeviceMessage> handler) {
        messageHandlers.put(dataType, handler);
    }

    /**
     * 处理设备消息（对外暴露的主方法）
     */
    public void handleMessage(DeviceMessage message) {
        String dataType = message.getDataType();
        Consumer<DeviceMessage> handler = messageHandlers.get(dataType);

        if (handler != null) {
            handler.accept(message);
        } else {
            log.warn("未知数据类型：{}", dataType);
        }
    }

    // ========== 具体的消息处理方法 ==========

    /**
     * 网络流量数据入库
     * @param message
     */
    private void handleNetMessage(DeviceMessage message) {
        List<InitialBandwidthTraffic> interfaces = JsonDataParser.parseJsonData(message.getData(), InitialBandwidthTraffic.class);
        if(!interfaces.isEmpty()){
            String clientId = message.getClientId();
            // 时间戳转换
            long timestamp = interfaces.get(0).getTimestamp();
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime);
            // 创建比timestamp少5分钟的时间
            long fiveMinutesEarlier = millis - (5 * 60 * 1000); // 减去5分钟的毫秒数
            Date fiveMinutesEarlierDate = new Date(fiveMinutesEarlier / 1000 * 1000); // 同样去除毫秒
            // 查询临时表信息，计算实际流量值
            InitialBandwidthTrafficTemp temp = new InitialBandwidthTrafficTemp();
            temp.setCreateTime(fiveMinutesEarlierDate);
            temp.setClientId(clientId);
            List<InitialBandwidthTrafficTemp> tempList = initialBandwidthTrafficTempService.selectInitialBandwidthTrafficTempList(temp);
            if(!tempList.isEmpty()){
                // 1. 构建快速查找的Map，使用MAC地址+网卡名称作为唯一键
                Map<String, InitialBandwidthTrafficTemp> tempMap = tempList.stream()
                        .collect(Collectors.toMap(
                                tempItem -> generateKey(tempItem.getMac(), tempItem.getName()),
                                Function.identity(),
                                (existing, replacement) -> existing
                        ));

                // 2. 预计算除数（避免重复创建对象）
                BigDecimal divisor = new BigDecimal(300);

                interfaces.forEach(iface -> {
                    iface.setClientId(clientId);
                    iface.setCreateTime(createTime);

                    // 设置总流量（转换为比特）
                    iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed()));
                    iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed()));
                    iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed()));
                    iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed()));
                    iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed()));
                    iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed()));
                    // 首次采集，速率设为null
                    iface.setInSpeed(null);
                    iface.setOutSpeed(null);
                    iface.setIpv4InSpeed(null);
                    iface.setIpv4OutSpeed(null);
                    iface.setIpv6InSpeed(null);
                    iface.setIpv6OutSpeed(null);

                    // 使用MAC地址+网卡名称作为查找键
                    String key = generateKey(iface.getMac(), iface.getName());
                    InitialBandwidthTrafficTemp tempInfo = tempMap.get(key);
                    if (tempInfo != null) {
                        // 计算总流入速率
                        if (iface.getTotalInSpeed() != null && tempInfo.getTotalInSpeed() != null) {
                            BigDecimal nowInSpeed = new BigDecimal(iface.getTotalInSpeed());
                            BigDecimal tempInSpeed = new BigDecimal(tempInfo.getTotalInSpeed());
                            BigDecimal inDiff = nowInSpeed.subtract(tempInSpeed);
                            if (inDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setInSpeed(inDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算总流出速率
                        if (iface.getTotalOutSpeed() != null && tempInfo.getTotalOutSpeed() != null) {
                            BigDecimal nowOutSpeed = new BigDecimal(iface.getTotalOutSpeed());
                            BigDecimal tempOutSpeed = new BigDecimal(tempInfo.getTotalOutSpeed());
                            BigDecimal outDiff = nowOutSpeed.subtract(tempOutSpeed);
                            if (outDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setOutSpeed(outDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算IPv4流入速率
                        if (iface.getTotalIpv4InSpeed() != null && tempInfo.getTotalIpv4InSpeed() != null) {
                            BigDecimal nowIpv4In = new BigDecimal(iface.getTotalIpv4InSpeed());
                            BigDecimal tempIpv4In = new BigDecimal(tempInfo.getTotalIpv4InSpeed());
                            BigDecimal ipv4InDiff = nowIpv4In.subtract(tempIpv4In);
                            if (ipv4InDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setIpv4InSpeed(ipv4InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算IPv4流出速率
                        if (iface.getTotalIpv4OutSpeed() != null && tempInfo.getTotalIpv4OutSpeed() != null) {
                            BigDecimal nowIpv4Out = new BigDecimal(iface.getTotalIpv4OutSpeed());
                            BigDecimal tempIpv4Out = new BigDecimal(tempInfo.getTotalIpv4OutSpeed());
                            BigDecimal ipv4OutDiff = nowIpv4Out.subtract(tempIpv4Out);
                            if (ipv4OutDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setIpv4OutSpeed(ipv4OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算IPv6流入速率
                        if (iface.getTotalIpv6InSpeed() != null && tempInfo.getTotalIpv6InSpeed() != null) {
                            BigDecimal nowIpv6In = new BigDecimal(iface.getTotalIpv6InSpeed());
                            BigDecimal tempIpv6In = new BigDecimal(tempInfo.getTotalIpv6InSpeed());
                            BigDecimal ipv6InDiff = nowIpv6In.subtract(tempIpv6In);
                            if (ipv6InDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setIpv6InSpeed(ipv6InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算IPv6流出速率
                        if (iface.getTotalIpv6OutSpeed() != null && tempInfo.getTotalIpv6OutSpeed() != null) {
                            BigDecimal nowIpv6Out = new BigDecimal(iface.getTotalIpv6OutSpeed());
                            BigDecimal tempIpv6Out = new BigDecimal(tempInfo.getTotalIpv6OutSpeed());
                            BigDecimal ipv6OutDiff = nowIpv6Out.subtract(tempIpv6Out);
                            if (ipv6OutDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setIpv6OutSpeed(ipv6OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                    }
                });
                // 清空临时表对应server信息
                initialBandwidthTrafficTempService.deleteTempMsgByClientId(clientId);
            }else{
                interfaces.forEach(iface -> {
                    iface.setClientId(clientId);
                    iface.setCreateTime(createTime);
                    // 设置总流量（转换为比特）
                    iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed()));
                    iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed()));
                    iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed()));
                    iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed()));
                    iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed()));
                    iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed()));

                    // 首次采集，速率设为null
                    iface.setInSpeed(null);
                    iface.setOutSpeed(null);
                    iface.setIpv4InSpeed(null);
                    iface.setIpv4OutSpeed(null);
                    iface.setIpv6InSpeed(null);
                    iface.setIpv6OutSpeed(null);
                });
            }
            InitialBandwidthTraffic data = new InitialBandwidthTraffic();
            // 批量入库集合
            data.setList(interfaces);
            // 临时表 用来计算流量速率
            initialBandwidthTrafficTempService.batchInsertServerTemp(interfaces);
            // 初始流量数据入库
            initialBandwidthTrafficService.batchInsert(data);
            EpsInitialTrafficDataRemote epsInitialTrafficDataRemote = new EpsInitialTrafficDataRemote();
            epsInitialTrafficDataRemote.setStartTime(timeStr);
            epsInitialTrafficDataRemote.setEndTime(timeStr);
            // 复制到业务初始库
            remoteRevenueConfigService.autoSaveServiceTrafficData(epsInitialTrafficDataRemote, SecurityConstants.INNER);
        }else{
            throw new RuntimeException("NET流量data数据为空");
        }
    }

    /**
     * 存储时间戳到Redis
     * @param clientId 客户端ID
     * @param timestamp 时间戳
     */
    private void storeTimestampToRedis(String clientId, long timestamp) {
        try {
            String redisKey = "net_traffic_timestamps:" + clientId;
            String existingTimestamps = redisTemplate.opsForValue().get(redisKey);

            if (existingTimestamps != null && !existingTimestamps.isEmpty()) {
                // 已存在时间戳，追加新的时间戳
                String newTimestamps = existingTimestamps + "," + timestamp;
                redisTemplate.opsForValue().set(redisKey, newTimestamps);
            } else {
                // 首次存储
                redisTemplate.opsForValue().set(redisKey, String.valueOf(timestamp));
            }
        } catch (Exception e) {
            // Redis操作异常处理，记录日志但不中断主流程
            log.error("存储时间戳到Redis失败，clientId: {}, timestamp: {}", clientId, timestamp, e);
        }
    }
    private void handleNetRecoverMessage(DeviceMessage message) {
        String clientId = message.getClientId();
        String lockKey = "traffic:recover:" + clientId;
        RLock lock = redissonClient.getLock(lockKey);
        boolean locked = false;

        try {
            // 尝试获取锁
            locked = lock.tryLock(0, 20, TimeUnit.SECONDS);
            if (locked) {
                log.info("设备{}获取锁成功，开始处理消息", clientId);
                processNetRecoverMessageInternal(message);
                log.info("设备{}消息处理完成", clientId);
            } else {
                log.warn("设备{}获取锁失败，消息处理繁忙", clientId);
                throw new RuntimeException("设备处理繁忙，请重试");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("处理消息时线程被中断", e);
        } finally {
            // 只有成功获取锁的线程才需要释放
            if (locked && lock.isHeldByCurrentThread()) {
                try {
                    lock.unlock();
                    log.debug("设备{}锁已释放", clientId);
                } catch (IllegalMonitorStateException e) {
                    log.warn("设备{}锁释放异常，可能已自动超时", clientId);
                }
            }
        }
    }
    /**
     * 网络流量数据入库
     * @param message
     */
    private void processNetRecoverMessageInternal(DeviceMessage message) {
        List<InitialBandwidthTraffic> interfaces = JsonDataParser.parseJsonData(message.getData(), InitialBandwidthTraffic.class);
        if(!interfaces.isEmpty()){
            String clientId = message.getClientId();
            // 时间戳转换
            long timestamp = interfaces.get(0).getTimestamp();
            // 时间戳存储到redis
            storeTimestampToRedis(clientId, timestamp);
            if(message.isLastTrafficFlag()){
                // 把redis中存储的时间戳提取出来，删除redis中的时间戳
                processExitsTraffic(clientId);
            }
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime);
            // 判断数据库中是否已存在数据
            InitialBandwidthTraffic countQuery = new InitialBandwidthTraffic();
            countQuery.setClientId(clientId);
            countQuery.setCreateTime(createTime);
            countQuery.setStartTime(timeStr);
            int exitsCount = initialBandwidthTrafficService.countByClientIdAndTime(countQuery);
            if(exitsCount > 0){
                return;
            }
            // 创建比timestamp少5分钟的时间
            long fiveMinutesEarlier = millis - (5 * 60 * 1000); // 减去5分钟的毫秒数
            Date fiveMinutesEarlierDate = new Date(fiveMinutesEarlier / 1000 * 1000); // 同样去除毫秒
            // 查询临时表信息，计算实际流量值
            InitialBandwidthTrafficTemp temp = new InitialBandwidthTrafficTemp();
            temp.setCreateTime(fiveMinutesEarlierDate);
            temp.setClientId(clientId);
            List<InitialBandwidthTrafficTemp> tempList = initialBandwidthTrafficTempService.selectInitialBandwidthTrafficRecoverTempList(temp);
            if(!tempList.isEmpty()){
                // 1. 构建快速查找的Map，使用MAC地址+网卡名称作为唯一键
                Map<String, InitialBandwidthTrafficTemp> tempMap = tempList.stream()
                        .collect(Collectors.toMap(
                                tempItem -> generateKey(tempItem.getMac(), tempItem.getName()),
                                Function.identity(),
                                (existing, replacement) -> existing
                        ));

                // 2. 预计算除数（避免重复创建对象）
                BigDecimal divisor = new BigDecimal(300);

                interfaces.forEach(iface -> {
                    iface.setClientId(clientId);
                    iface.setCreateTime(createTime);

                    // 设置总流量（转换为比特）
                    iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed()));
                    iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed()));
                    iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed()));
                    iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed()));
                    iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed()));
                    iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed()));
                    // 首次采集，速率设为null
                    iface.setInSpeed(null);
                    iface.setOutSpeed(null);
                    iface.setIpv4InSpeed(null);
                    iface.setIpv4OutSpeed(null);
                    iface.setIpv6InSpeed(null);
                    iface.setIpv6OutSpeed(null);

                    // 使用MAC地址+网卡名称作为查找键
                    String key = generateKey(iface.getMac(), iface.getName());
                    InitialBandwidthTrafficTemp tempInfo = tempMap.get(key);
                    if (tempInfo != null) {
                        // 计算总流入速率
                        if (iface.getTotalInSpeed() != null && tempInfo.getTotalInSpeed() != null) {
                            BigDecimal nowInSpeed = new BigDecimal(iface.getTotalInSpeed());
                            BigDecimal tempInSpeed = new BigDecimal(tempInfo.getTotalInSpeed());
                            BigDecimal inDiff = nowInSpeed.subtract(tempInSpeed);
                            if (inDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setInSpeed(inDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算总流出速率
                        if (iface.getTotalOutSpeed() != null && tempInfo.getTotalOutSpeed() != null) {
                            BigDecimal nowOutSpeed = new BigDecimal(iface.getTotalOutSpeed());
                            BigDecimal tempOutSpeed = new BigDecimal(tempInfo.getTotalOutSpeed());
                            BigDecimal outDiff = nowOutSpeed.subtract(tempOutSpeed);
                            if (outDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setOutSpeed(outDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算IPv4流入速率
                        if (iface.getTotalIpv4InSpeed() != null && tempInfo.getTotalIpv4InSpeed() != null) {
                            BigDecimal nowIpv4In = new BigDecimal(iface.getTotalIpv4InSpeed());
                            BigDecimal tempIpv4In = new BigDecimal(tempInfo.getTotalIpv4InSpeed());
                            BigDecimal ipv4InDiff = nowIpv4In.subtract(tempIpv4In);
                            if (ipv4InDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setIpv4InSpeed(ipv4InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算IPv4流出速率
                        if (iface.getTotalIpv4OutSpeed() != null && tempInfo.getTotalIpv4OutSpeed() != null) {
                            BigDecimal nowIpv4Out = new BigDecimal(iface.getTotalIpv4OutSpeed());
                            BigDecimal tempIpv4Out = new BigDecimal(tempInfo.getTotalIpv4OutSpeed());
                            BigDecimal ipv4OutDiff = nowIpv4Out.subtract(tempIpv4Out);
                            if (ipv4OutDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setIpv4OutSpeed(ipv4OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算IPv6流入速率
                        if (iface.getTotalIpv6InSpeed() != null && tempInfo.getTotalIpv6InSpeed() != null) {
                            BigDecimal nowIpv6In = new BigDecimal(iface.getTotalIpv6InSpeed());
                            BigDecimal tempIpv6In = new BigDecimal(tempInfo.getTotalIpv6InSpeed());
                            BigDecimal ipv6InDiff = nowIpv6In.subtract(tempIpv6In);
                            if (ipv6InDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setIpv6InSpeed(ipv6InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                        // 计算IPv6流出速率
                        if (iface.getTotalIpv6OutSpeed() != null && tempInfo.getTotalIpv6OutSpeed() != null) {
                            BigDecimal nowIpv6Out = new BigDecimal(iface.getTotalIpv6OutSpeed());
                            BigDecimal tempIpv6Out = new BigDecimal(tempInfo.getTotalIpv6OutSpeed());
                            BigDecimal ipv6OutDiff = nowIpv6Out.subtract(tempIpv6Out);
                            if (ipv6OutDiff.compareTo(BigDecimal.ZERO) >= 0) {
                                iface.setIpv6OutSpeed(ipv6OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString());
                            }
                        }
                    }
                });
                // 清空临时表对应server信息
                InitialBandwidthTrafficTemp delQuery = new InitialBandwidthTrafficTemp();
                delQuery.setClientId(clientId);
                delQuery.setCreateTime(fiveMinutesEarlierDate);
                initialBandwidthTrafficTempService.deleteTempMsgByClientIdAndTime(delQuery);
            }else{
                interfaces.forEach(iface -> {
                    iface.setClientId(clientId);
                    iface.setCreateTime(createTime);
                    // 设置总流量（转换为比特）
                    iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed()));
                    iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed()));
                    iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed()));
                    iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed()));
                    iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed()));
                    iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed()));

                    // 首次采集，速率设为null
                    iface.setInSpeed(null);
                    iface.setOutSpeed(null);
                    iface.setIpv4InSpeed(null);
                    iface.setIpv4OutSpeed(null);
                    iface.setIpv6InSpeed(null);
                    iface.setIpv6OutSpeed(null);
                });
            }
            InitialBandwidthTraffic data = new InitialBandwidthTraffic();
            // 批量入库集合
            data.setList(interfaces);
            // 临时表 用来计算流量速率
            initialBandwidthTrafficTempService.batchInsertServerRecoverTemp(interfaces);
            // 初始流量数据入库
            initialBandwidthTrafficService.batchInsert(data);
            EpsInitialTrafficDataRemote epsInitialTrafficDataRemote = new EpsInitialTrafficDataRemote();
            epsInitialTrafficDataRemote.setStartTime(timeStr);
            epsInitialTrafficDataRemote.setEndTime(timeStr);
            // 复制到业务初始库
            remoteRevenueConfigService.autoSaveServiceTrafficData(epsInitialTrafficDataRemote, SecurityConstants.INNER);
        }else{
            throw new RuntimeException("NET流量data数据为空");
        }
    }

    private void processExitsTraffic(String clientId) {
        try {
            String redisKey = "net_traffic_timestamps:" + clientId;
            String storedTimestamps = redisTemplate.opsForValue().get(redisKey);

            if (storedTimestamps != null && !storedTimestamps.isEmpty()) {
                // 删除Redis中的时间戳数据
                redisTemplate.delete(redisKey);
                // 构建磁盘下发信息
                PolicyTypeVo policyTypeVo = new PolicyTypeVo();
                policyTypeVo.setTrafficUsedTimestamp(storedTimestamps);
                MessageProducer messageProducer = new MessageProducer();
                String configJson = JSONObject.toJSONString(policyTypeVo);
                DeviceMessage message = new DeviceMessage();
                message.setClientId(clientId);
                message.setData(configJson);
                message.setDataType(MsgEnum.获取最新策略应答.getValue());

                messageProducer.sendAsyncProducerMessage(
                        producerMode.getAgentTopic(),
                        "",
                        "",
                        JSONObject.toJSONString(message)
                );
            }
        } catch (Exception e) {
            log.error("提取并删除Redis时间戳失败，clientId: {}", clientId, e);
        }
    }

    private String generateKey(String mac, String name) {
        if (mac == null) {
            mac = "";
        }
        if (name == null) {
            name = "";
        }
        return mac + "|" + name;
    }
    /**
     * docker数据入库
     * @param message
     */
    private void handleDockerMessage(DeviceMessage message) {
        List<InitialDockerInfo> dockers = JsonDataParser.parseJsonData(message.getData(), InitialDockerInfo.class);
        if(!dockers.isEmpty()){
            // 时间戳转换
            long timestamp = dockers.get(0).getTimestamp();
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            dockers.forEach(iface -> {
                iface.setClientId(message.getClientId());
                iface.setCreateTime(createTime);
            });
            // 初始容器数据入库
            initialDockerInfoService.batchInsertInitialDockerInfo(dockers);
        }else{
            throw new RuntimeException("DOCKER容器data数据为空");
        }
    }
    /**
     * cpu数据入库
     * @param message
     */
    private void handleCpuMessage(DeviceMessage message) {
        List<InitialCpuInfo> cpus = JsonDataParser.parseJsonData(message.getData(),InitialCpuInfo.class);
        // 时间戳转换
        long timestamp = cpus.get(0).getTimestamp();
        long millis = timestamp * 1000;
        Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
        if(!cpus.isEmpty()){
            cpus.forEach(iface -> {
                iface.setClientId(message.getClientId());
                iface.setCreateTime(createTime);
                if(iface.getTemperature() == 0L){
                    iface.setTemperature(null);
                }
            });
            // 初始CPU数据入库
            initialCpuInfoService.batchInsertInitialCpuInfo(cpus);
        }else{
            throw new RuntimeException("CPUdata数据为空");
        }
    }
    /**
     * 磁盘数据入库
     * @param message
     */
    private void handleDiskMessage(DeviceMessage message) {
        List<InitialDiskInfo> disks = JsonDataParser.parseJsonData(message.getData(), InitialDiskInfo.class);
        String clientId = message.getClientId();

        if (disks == null || disks.isEmpty()) {
            throw new RuntimeException("磁盘data数据为空");
        }

        long timestamp = disks.get(0).getTimestamp();
        long millis = timestamp * 1000;
        Date createTime = new Date(millis / 1000 * 1000);

        // 关键：每个clientId有自己独立的key
        String diskCountKey = DISK_COUNT_PREFIX + clientId;

        // 1. 给这个客户端的所有磁盘次数+1
        Map<Object, Object> diskCountMap = redisTemplate.opsForHash().entries(diskCountKey);
        for (Map.Entry<Object, Object> entry : diskCountMap.entrySet()) {
            String diskName = (String) entry.getKey();
            String countStr = (String) entry.getValue();

            try {
                int count = Integer.parseInt(countStr) + 1;
                redisTemplate.opsForHash().put(diskCountKey, diskName, String.valueOf(count));
            } catch (NumberFormatException e) {
                redisTemplate.opsForHash().put(diskCountKey, diskName, "1");
            }
        }

        // 2. 处理本次上报的磁盘
        Set<String> reportedDisks = new HashSet<>();
        disks.forEach(disk -> {
            String diskName = disk.getName();
            reportedDisks.add(diskName);

            // 本次上报的磁盘，次数重置为0
            redisTemplate.opsForHash().put(diskCountKey, diskName, "0");

            disk.setClientId(clientId);
            disk.setCreateTime(createTime);
        });

        // 3. 检查次数≥3的磁盘
        diskCountMap = redisTemplate.opsForHash().entries(diskCountKey);
        List<String> disksToRemove = new ArrayList<>();

        for (Map.Entry<Object, Object> entry : diskCountMap.entrySet()) {
            String diskName = (String) entry.getKey();
            String countStr = (String) entry.getValue();

            try {
                int count = Integer.parseInt(countStr);
                // 如果次数≥3且本次没上报
                if (count >= 3 && !reportedDisks.contains(diskName)) {
                    AllDiskName allDiskName = new AllDiskName();
                    allDiskName.setStatus(0);
                    allDiskName.setClientId(clientId);
                    allDiskName.setName(diskName);
                    allDiskNameService.updateAllDiskName(allDiskName);
                    disksToRemove.add(diskName);
                }
            } catch (NumberFormatException e) {
                disksToRemove.add(diskName);
            }
        }

        // 4. 删除已处理的磁盘记录
        if (!disksToRemove.isEmpty()) {
            redisTemplate.opsForHash().delete(diskCountKey, disksToRemove.toArray());
        }

        // 5. 数据入库
        initialDiskInfoService.batchInsertInitialDiskInfo(disks, createTime);
    }
    /**
     * 内存数据入库
     * @param message
     */
    private void handleMemoryMessage(DeviceMessage message) {
        List<InitialMemoryInfo> memorys = JsonDataParser.parseJsonData(message.getData(), InitialMemoryInfo.class);
        if(!memorys.isEmpty()){
            // 时间戳转换
            long timestamp = memorys.get(0).getTimestamp();
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            memorys.forEach(iface -> {
                iface.setClientId(message.getClientId());
                iface.setCreateTime(createTime);
            });
            // 初始内存数据入库
            initialMemoryInfoService.batchInsertInitialMemoryInfo(memorys);
        }else{
            throw new RuntimeException("内存data数据为空");
        }
    }
    /**
     * 挂载点数据入库
     * @param message
     */
    private void handleMountPointMessage(DeviceMessage message) {
        List<InitialMountPointInfo> mountPointInfos = JsonDataParser.parseJsonData(message.getData(), InitialMountPointInfo.class);
        if(!mountPointInfos.isEmpty()){
            // 时间戳转换
            long timestamp = mountPointInfos.get(0).getTimestamp();
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            mountPointInfos.forEach(iface -> {
                iface.setClientId(message.getClientId());
                iface.setCreateTime(createTime);
            });
            // 初始挂载点数据入库
            initialMountPointInfoService.batchInsertInitialMountPointInfo(mountPointInfos);
        }else{
            throw new RuntimeException("挂载点data数据为空");
        }
    }




    /**
     * 系统其他信息
     * @param message
     */
    private void handleOtherSystemMessage(DeviceMessage message) {
        List<CollectDataVo> otherData = JsonDataParser.parseJsonData(message.getData(), CollectDataVo.class);
        if(!otherData.isEmpty()){
            CollectDataVo systemDataVo = otherData.get(0);
            if (systemDataVo != null){
                try {
                    InitialSystemOtherCollectData insertData = new InitialSystemOtherCollectData();

                    // 时间戳转换
                    long timestamp = systemDataVo.getTimestamp();
                    long millis = timestamp * 1000;
                    Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒

                    insertData.setClientId(message.getClientId());
                    insertData.setCreateTime(createTime);
                    String collectType = systemDataVo.getType();
                    String collectValue = systemDataVo.getValue();
                    insertData.setCollectType(collectType);
                    insertData.setCollectValue(collectValue);

                    // 定义需要先删后插的collectType列表
                    List<String> specialCollectTypes = Arrays.asList(
                            "kernelMaxprocCollect",
                            "memorySizeTotalCollect",
                            "systemBoottimeCollect",
                            "systemCpuNum",
                            "systemDiskSizeTotalCollect",
                            "systemLocaltimeCollect",
                            "systemSwArchCollect",
                            "systemSwOsCollect",
                            "systemUnameCollect",
                            "systemUptimeCollect",
                            "procNumCollect"
                    );

                    // 如果collectType在特殊列表中，先删除上个时间段的记录
                    if (specialCollectTypes.contains(collectType)) {

                        // 删除相同clientId、collectType和时间范围内的记录
                        InitialSystemOtherCollectData deleteData = new InitialSystemOtherCollectData();
                        deleteData.setClientId(message.getClientId());
                        deleteData.setCollectType(collectType);
                        deleteData.setCreateTime(createTime);
                        iInitialSystemOtherCollectDataService.deleteInitialSystemOtherCollectData(deleteData);
                    }

                    iInitialSystemOtherCollectDataService.insertInitialSystemOtherCollectData(insertData);

                } catch (Exception e) {
                    log.error("解析JSON数据失败: {}, value: {}", e.getMessage(), systemDataVo.getValue());
                    // 可以选择保存原始数据或进行其他错误处理
                }
            }
        }
    }
    /**
     * 监听心跳
     * @param message
     */
    private void handleHeartbeatMessage(DeviceMessage message) {
        List<InitialHeartbeatListen> heartbeats = JsonDataParser.parseJsonData(message.getData(), InitialHeartbeatListen.class);
        if(!heartbeats.isEmpty()){
            InitialHeartbeatListen heartbeat = heartbeats.get(0);
            String clientId = message.getClientId();
            String version = heartbeat.getVersion();
            String name = heartbeat.getName();
            log.debug("处理心跳消息，客户端ID: {}, 时间: {}", clientId, heartbeat.getTimestamp());

            // 使用Redis存储状态
            String statusKey = HEARTBEAT_STATUS_PREFIX + clientId;
            String timeKey = HEARTBEAT_TIME_PREFIX + clientId;
            String recoveryCountKey = HEARTBEAT_RECOVERY_COUNT_PREFIX + clientId;
            String heartbeatCountKey = HEARTBEAT_COUNT_PREFIX + clientId;
            String alertKey = HEARTBEAT_ALERT_PREFIX + clientId;

            try {
                // 记录处理前状态（调试用）
                String prevStatus = redisTemplate.opsForValue().get(statusKey);
                String prevTime = redisTemplate.opsForValue().get(timeKey);
                String prevHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey);
                Boolean prevAlertStatus = redisTemplate.hasKey(alertKey);
                log.debug("客户端ID: {} 处理前状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}",
                        clientId, prevStatus, prevTime, prevHeartbeatCount, prevAlertStatus);

                // 原子递增心跳计数（线程安全）
                Long newHeartbeatCount = redisTemplate.opsForValue().increment(heartbeatCountKey);

                // 使用事务更新状态和时间
                redisTemplate.execute(new SessionCallback<Object>() {
                    @Override
                    public Object execute(RedisOperations operations) throws DataAccessException {
                        operations.multi();
                        // 重置丢失计数为0，设置最后心跳时间
                        operations.opsForValue().set(statusKey, "0");
                        operations.opsForValue().set(timeKey, String.valueOf(System.currentTimeMillis()));
                        return operations.exec();
                    }
                });

                log.debug("客户端ID: {} 心跳处理完成，当前心跳次数: {}", clientId, newHeartbeatCount);

                // 检查是否之前有告警状态（心跳恢复检测）
                if (Boolean.TRUE.equals(redisTemplate.hasKey(alertKey))) {
                    log.info("客户端ID: {} 检测到心跳恢复", clientId);

                    // 原子递增恢复计数
                    Long recoveryCount = redisTemplate.opsForValue().increment(recoveryCountKey);

                    log.debug("客户端ID: {} 恢复计数: {}", clientId, recoveryCount);

                    if (recoveryCount >= 2) {
                        // 达到2次恢复，清除告警状态
                        log.warn("客户端ID: {} 心跳恢复达到{}次，清除告警状态", clientId, recoveryCount);
                        insertHeartbeatLog(clientId, "2", "心跳恢复,设备在线状态改为在线");

                        // 清理告警相关key
                        redisTemplate.delete(alertKey);
                        redisTemplate.delete(recoveryCountKey);

                        // 修改资源状态为在线
                        updateResourceStatus(clientId, "1");

                        log.info("客户端ID: {} 告警状态已清除", clientId);
                    } else {
                        // 未达到2次，只记录恢复次数
                        log.info("客户端ID: {} 心跳恢复第{}次", clientId, recoveryCount);
                    }
                }

                // 只有达到3次心跳才执行数据库操作
                if (newHeartbeatCount >= 3) {
                    log.debug("客户端ID: {} 达到{}次心跳，开始执行数据库操作", clientId, newHeartbeatCount);

                    // 添加逻辑节点标识
                    RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote();
                    updateData.setClientId(message.getClientId());
                    updateData.setLogicalNodeId(heartbeat.getLogicalNode());
                    updateData.setHardwareSn(heartbeat.getSn());
                    updateData.setOnlineStatus("1");
                    updateData.setAgentVersion(version);
                    updateData.setOnboardTime(DateUtils.getNowDate());
                    remoteRevenueConfigService.innerUpdateRegist(updateData, SecurityConstants.INNER);

                    // agent更新结果存储
                    RmAgentManagement query = new RmAgentManagement();
                    query.setClientId(clientId);
                    List<RmAgentManagement> agentManagements = rmAgentManagementService.selectRmAgentManagementList(query);
                    if(!agentManagements.isEmpty()){
                        RmAgentManagement rmAgentManagement = agentManagements.get(0);
                        if(!StringUtils.equals(rmAgentManagement.getAgentVersion(), version)){
                            // 存储更新结果
                            RmAgentManagement updateResultData = new RmAgentManagement();
                            updateResultData.setId(rmAgentManagement.getId());
                            updateResultData.setAgentVersion(version);
                            updateResultData.setLastUpdateTime(DateUtils.getNowDate());
                            updateResultData.setLastUpdateResult("1");
                            rmAgentManagementService.updateRmAgentManagementByHardwareSn(updateResultData);
                        }
                    }

                    // 交换机在线状态改为在线（只在第3次心跳时执行）
                    if(newHeartbeatCount == 3){
                        RmSwitchManagementRemote rmSwitchManagementRemote = new RmSwitchManagementRemote();
                        rmSwitchManagementRemote.setClientId(clientId);
                        R<List<RmSwitchManagementRemote>> rmSwitchManagementRemoteListR = remoteRevenueConfigService.getSwitchNameByClientId(rmSwitchManagementRemote, SecurityConstants.INNER);
                        if(rmSwitchManagementRemoteListR != null &&
                                rmSwitchManagementRemoteListR.getData()!=null &&
                                !rmSwitchManagementRemoteListR.getData().isEmpty()){
                            RmSwitchManagementRemote switchUpdate = new RmSwitchManagementRemote();
                            switchUpdate.setClientId(clientId);
                            switchUpdate.setOnlineStatus("1");
                            remoteRevenueConfigService.updateSwitchMsgByClientId(switchUpdate, SecurityConstants.INNER);
                            log.info("客户端ID: {} 交换机状态更新完成", clientId);
                        } else {
                            log.debug("客户端ID: {} 未找到对应的交换机信息", clientId);
                        }
                    }
                } else {
                    log.debug("客户端ID: {} 当前心跳次数: {}，未达到3次，跳过数据库操作", clientId, newHeartbeatCount);
                }

                // 记录处理后状态（调试用）
                String currentStatus = redisTemplate.opsForValue().get(statusKey);
                String currentTime = redisTemplate.opsForValue().get(timeKey);
                String currentHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey);
                Boolean currentAlertStatus = redisTemplate.hasKey(alertKey);
                log.debug("客户端ID: {} 处理后状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}",
                        clientId, currentStatus, currentTime, currentHeartbeatCount, currentAlertStatus);

            } catch (Exception e) {
                log.error("处理心跳消息异常, clientId: {}", clientId, e);
            }
        }
    }

    // 添加一个定时任务方法，定期检查心跳状态
    @Scheduled(fixedRate = 60000) // 每60s检查一次
    public void checkHeartbeatStatus() {
        long currentTime = System.currentTimeMillis();
        log.debug("开始心跳状态检查，当前时间: {}", currentTime);

        // 获取所有客户端时间键
        Set<String> timeKeys = redisTemplate.keys(HEARTBEAT_TIME_PREFIX + "*");
        if (timeKeys == null) {
            log.debug("未找到任何心跳时间键");
            return;
        }

        log.debug("找到 {} 个客户端需要检查", timeKeys.size());

        for (String timeKey : timeKeys) {
            String clientId = timeKey.substring(HEARTBEAT_TIME_PREFIX.length());
            String statusKey = HEARTBEAT_STATUS_PREFIX + clientId;
            String alertKey = HEARTBEAT_ALERT_PREFIX + clientId;

            try {
                // 检查是否已经存在告警
                String existingAlert = redisTemplate.opsForValue().get(alertKey);
                if ("1".equals(existingAlert)) {
                    log.debug("客户端ID: {} 已有告警，跳过检查", clientId);
                    continue; // 如果已有告警，跳过处理
                }

                String lastTimeStr = redisTemplate.opsForValue().get(timeKey);
                if (lastTimeStr == null) {
                    log.debug("客户端ID: {} 时间键为空，跳过", clientId);
                    continue;
                }

                long lastHeartbeatTime = Long.parseLong(lastTimeStr);
                long timeDiff = currentTime - lastHeartbeatTime;

                log.debug("客户端ID: {} 最后心跳: {}, 时间差: {}ms, 超时阈值: {}ms",
                        clientId, lastHeartbeatTime, timeDiff, HEARTBEAT_TIMEOUT);

                if (timeDiff > HEARTBEAT_TIMEOUT) {
                    // 心跳超时处理 - 使用原子操作增加计数
                    Long lostCount = redisTemplate.opsForValue().increment(statusKey);
                    if (lostCount == 1) {
                        // 确保第一次增加时值为1
                        redisTemplate.opsForValue().set(statusKey, "1");
                        lostCount = 1L;
                    }

                    log.warn("客户端ID: {} 心跳超时，连续次数: {}, 时间差: {}ms",
                            clientId, lostCount, timeDiff);

                    if (lostCount >= 3) {
                        log.warn("客户端ID: {} 连续三次心跳丢失，触发告警", clientId);
                        insertHeartbeatLog(clientId, "3", "连续三次心跳丢失");
                        // 告警
                        insertAlarmRecords(clientId);
                        redisTemplate.opsForValue().set(HEARTBEAT_ALERT_PREFIX + clientId, "1");
                        // 设置告警后删除timeKey和statusKey
                        redisTemplate.delete(timeKey);
                        redisTemplate.delete(statusKey);

                        log.info("客户端ID: {} 已设置告警并清理心跳记录", clientId);
                        // frpc状态改为未运行
                        RmFrpcConfigManage updateFrp = new RmFrpcConfigManage();
                        updateFrp.setClientId(clientId);
                        updateFrp.setFrpcStatus("0");
                        rmFrpcConfigManageService.updateRmFrpcConfigManage(updateFrp);
                        // 修改资源状态
                        updateResourceStatus(clientId, "0");
                    }
                } else {
                    // 如果心跳正常，重置丢失次数
                    String currentStatus = redisTemplate.opsForValue().get(statusKey);
                    if (!"0".equals(currentStatus)) {
                        redisTemplate.opsForValue().set(statusKey, "0");
                        log.debug("客户端ID: {} 心跳正常，重置丢失次数从 {} 到 0", clientId, currentStatus);
                    } else {
                        log.debug("客户端ID: {} 心跳正常，状态已是0", clientId);
                    }
                }
            } catch (Exception e) {
                log.error("检查心跳状态异常, clientId: {}", clientId, e);
            }
        }

        log.debug("心跳状态检查完成");
    }

    // 更新资源状态的公共方法
    private void updateResourceStatus(String clientId, String status) {
        log.info("开启更新资源状态========");
        RmResourceRegistrationRemote query = new RmResourceRegistrationRemote();
        query.setClientId(clientId);
        R<RmResourceRegistrationRemote> registerMsgR = remoteRevenueConfigService.getListByHardwareSn(query, SecurityConstants.INNER);
        if(registerMsgR != null && registerMsgR.getData() != null){
            RmResourceRegistrationRemote rmResourceRegistrationRemote = new RmResourceRegistrationRemote();
            rmResourceRegistrationRemote.setOnlineStatus(status);
            rmResourceRegistrationRemote.setClientId(clientId);
            remoteRevenueConfigService.updateStatusByResource(rmResourceRegistrationRemote, SecurityConstants.INNER);
        }
        RmSwitchManagementRemote rmSwitchManagementRemote = new RmSwitchManagementRemote();
        rmSwitchManagementRemote.setClientId(clientId);
        R<List<RmSwitchManagementRemote>> rmSwitchManagementRemoteListR = remoteRevenueConfigService.getSwitchNameByClientId(rmSwitchManagementRemote, SecurityConstants.INNER);
        if(rmSwitchManagementRemoteListR != null &&
                rmSwitchManagementRemoteListR.getData()!=null &&
                !rmSwitchManagementRemoteListR.getData().isEmpty()){
            RmSwitchManagementRemote switchUpdate = new RmSwitchManagementRemote();
            switchUpdate.setClientId(clientId);
            switchUpdate.setOnlineStatus(status);
            remoteRevenueConfigService.updateSwitchMsgByClientId(switchUpdate, SecurityConstants.INNER);
        }
    }
    // 插入心跳日志到数据库
    private void insertHeartbeatLog(String machineId, String status, String remark) {
        try {
            InitialHeartbeatListenLog listenLog = new InitialHeartbeatListenLog();
            listenLog.setClientId(machineId);
            listenLog.setStatus(status); // 0-离线 1-在线 2-恢复 3-三次丢失
            listenLog.setRemark(remark);
            listenLog.setCreateTime(new Date());

            // 调用DAO或Service插入日志
            initialHeartbeatListenLog.insertInitialHeartbeatListenLog(listenLog);
            log.info("已记录心跳日志，客户端ID: {}, 状态: {}", machineId, status);
        } catch (Exception e) {
            log.error("插入心跳日志失败", e);
        }
    }

    /**
     * 添加告警日志并推送消息到企业微信
     * @param clientId
     */
    private void insertAlarmRecords(String clientId) {
        // 查询clientId是否为交换机唯一标识
        boolean isSwitch = false;
        String switchName = "";
        RmSwitchManagementRemote rmSwitchManagementRemote = new RmSwitchManagementRemote();
        rmSwitchManagementRemote.setClientId(clientId);
        R<List<RmSwitchManagementRemote>> rmSwitchManagementRemoteListR = remoteRevenueConfigService.getSwitchNameByClientId(rmSwitchManagementRemote, SecurityConstants.INNER);
        if (rmSwitchManagementRemoteListR != null &&
                rmSwitchManagementRemoteListR.getData() != null &&
                !rmSwitchManagementRemoteListR.getData().isEmpty()) {
            isSwitch = true;
            switchName = rmSwitchManagementRemoteListR.getData().get(0).getSwitchName();
        }

        // 创建告警日志记录
        RmAlarmLog rmAlarmLog = createAlarmLog(clientId, isSwitch, switchName);

        // 插入告警日志
        rmAlarmLogService.insertRmAlarmLog(rmAlarmLog);

        // 发送告警推送
        sendAlarmPush(rmAlarmLog, isSwitch);
    }

    /**
     * 创建告警日志记录
     */
    private RmAlarmLog createAlarmLog(String clientId, boolean isSwitch, String switchName) {
        RmAlarmLog rmAlarmLog = new RmAlarmLog();
        String alarmContent = clientId + "下线";

        if (isSwitch) {
            rmAlarmLog.setClientId(switchName);
            rmAlarmLog.setAlarmType("2");
            alarmContent = switchName + "下线";
        } else {
            // 查询管理网公网ip
            RmNetworkInterface rmNetworkInterface = new RmNetworkInterface();
            rmNetworkInterface.setClientId(clientId);
            rmNetworkInterface.setNewFlag(1);
            List<RmNetworkInterface> interfaceList = rmNetworkInterfaceService.selectRmNetworkInterfaceList(rmNetworkInterface);
            if (interfaceList != null && !interfaceList.isEmpty()) {
                interfaceList.stream()
                        .filter(info -> "2".equals(info.getBindIp()) || "3".equals(info.getBindIp()))
                        .findFirst()
                        .ifPresent(networkInterface -> {
                            rmAlarmLog.setMgmPublicIp(networkInterface.getPublicIp());
                        });
            }
            // 查询业务名称
            RmResourceRegistrationRemote query = new RmResourceRegistrationRemote();
            query.setClientId(clientId);
            R<RmResourceRegistrationRemote> registrationR = remoteRevenueConfigService.getListByHardwareSn(query, SecurityConstants.INNER);
            if(registrationR != null){
                RmResourceRegistrationRemote registrationMsg = registrationR.getData();
                if(registrationMsg != null){
                    rmAlarmLog.setBusinessName(registrationMsg.getBusinessName());
                }
            }
            rmAlarmLog.setClientId(clientId);
            rmAlarmLog.setAlarmType("1");
        }

        rmAlarmLog.setAlarmContent(alarmContent);
        rmAlarmLog.setAlarmTime(DateUtils.getNowDate());
        return rmAlarmLog;
    }

    /**
     * 发送告警推送
     */
    private void sendAlarmPush(RmAlarmLog rmAlarmLog, boolean isSwitch) {
        String alarmTypeCode = isSwitch ? AlarmTypeEnum.交换机下线.getCode() : AlarmTypeEnum.服务器下线.getCode();
        String alarmTypeMsg = isSwitch ? AlarmTypeEnum.交换机下线.getMsg() : AlarmTypeEnum.服务器下线.getMsg();

        RmAlarmPushConfig rmAlarmPushConfig = new RmAlarmPushConfig();
        rmAlarmPushConfig.setPushMethod(PushMethodEnum.企业微信.getCode());
        rmAlarmPushConfig.setPushAlarmTypes(alarmTypeCode);
        List<RmAlarmPushConfig> alarmConfigList = rmAlarmPushConfigService.selectRmAlarmPushConfigList(rmAlarmPushConfig);
        RmAlarmLog updateData = new RmAlarmLog();
        updateData.setId(rmAlarmLog.getId());
        if (alarmConfigList != null && !alarmConfigList.isEmpty()) {
            try {
                for (RmAlarmPushConfig alarmPushConfig : alarmConfigList) {
                    String contentTemplate = alarmPushConfig.getMessageContent();
                    String webhookUrl = alarmPushConfig.getPushAddress();
                    Map<String, Object> alarmMap = new HashMap<>();
                    alarmMap.put("告警时间", rmAlarmLog.getAlarmTime());
                    alarmMap.put("IP", rmAlarmLog.getMgmPublicIp());
                    alarmMap.put("告警类型", alarmTypeMsg);
                    alarmMap.put("告警设备", rmAlarmLog.getClientId());
                    alarmMap.put("告警内容", rmAlarmLog.getAlarmContent());
                    alarmMap.put("业务名称", rmAlarmLog.getBusinessName());

                    if (alarmPushConfig.getContactPhones() != null) {
                        String[] phones = alarmPushConfig.getContactPhones().split(",");
                        WeChatWorkBot.sendTemplateMessage(webhookUrl, contentTemplate, alarmMap, rmAlarmLog.getAlarmContent(), phones, false);
                    } else {
                        WeChatWorkBot.sendTemplateMessage(webhookUrl, contentTemplate, alarmMap);
                    }
                }
                updateData.setPushFlag(1L);
            } catch (Exception e){
                updateData.setPushFlag(0L);
                log.error("消息推送失败：{}", e.getMessage());
            }
        }else{
            updateData.setPushFlag(2L);
        }
        rmAlarmLogService.updateRmAlarmLog(updateData);
    }

    /**
     * 应答信息
     * @param message
     */
    private RspVo handleResponseMessage(DeviceMessage message) {
        List<RspVo> rspVoList = JsonDataParser.parseJsonData(message.getData(), RspVo.class);
        if (!rspVoList.isEmpty()) {
            RspVo rsp = rspVoList.get(0);
            log.info("应答信息：{}",rsp);
            return rsp;
        }
        return null;
    }
    /**
     * 脚本策略应答信息
     * @param message
     */
    private void handleScriptRspMessage(DeviceMessage message) {
        List<RspVo> rspVoList = JsonDataParser.parseJsonData(message.getData(), RspVo.class);
        if (!rspVoList.isEmpty()) {
            RspVo rsp = rspVoList.get(0);
            // 时间戳转换
            long timestamp = rsp.getTimestamp();
            long millis = timestamp * 1000;
            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
            if(rsp.getResCode() == 1){
                if(rsp.getResult() != null){
                    List<RspResultVo> resultVos = JsonDataParser.parseJsonData(rsp.getResult(), RspResultVo.class);
                    RspResultVo rspResultVo = resultVos.get(0);
                    JSONObject jsonObject = JSONObject.parseObject(rsp.getResult());
                    String resOut = jsonObject.getString("resOut");
                    // 查询执行的脚本id
                    RmDeploymentPolicy deploymentPolicy = rmDeploymentPolicyService.selectRmDeploymentPolicyById(Long.valueOf(rspResultVo.getScriptId()));
                    Long businessScriptId = deploymentPolicy.getScriptId();
                    // 判断失败返回结果是否包含脚本执行失败关键字
                    R<EpsBusinessScriptRemote> businessScriptMsgR = remoteRevenueConfigService.getBusinessScriptMsgByScriptId(businessScriptId, SecurityConstants.INNER);
                    // 构建脚本执行结果实体类
                    RmResourceRemote insertData = new RmResourceRemote();
                    insertData.setClientId(message.getClientId());
                    insertData.setScriptId(Long.valueOf(rspResultVo.getScriptId()));
                    if(businessScriptMsgR != null && businessScriptMsgR.getData() != null){
                        EpsBusinessScriptRemote businessScriptMsg = businessScriptMsgR.getData();
                        String failedKeywords = businessScriptMsg.getFailedKeywords();
                        if(failedKeywords != null && resOut.contains(failedKeywords)){
                            insertData.setResultFlag(0);
                        }else {
                            insertData.setResultFlag(1);
                        }
                    }else{
                        insertData.setResultFlag(1);
                    }
                    insertData.setDescription(rsp.getResult());
                    insertData.setCreateTime(createTime);
                    // 执行插入sql
                    rmResourceRemoteService.insertRmResourceRemote(insertData);
                    log.info("脚本执行结果入库成功：{}",rsp);
                }
            }else{
                // 构建脚本执行结果实体类
                RmResourceRemote insertData = new RmResourceRemote();
                insertData.setClientId(message.getClientId());
                insertData.setResultFlag(0);
                if(rsp.getResult() != null){
                    List<RspResultVo> resultVos = JsonDataParser.parseJsonData(rsp.getResult(), RspResultVo.class);
                    RspResultVo rspResultVo = resultVos.get(0);
                    insertData.setDescription(rsp.getResult());
                    insertData.setScriptId(Long.valueOf(rspResultVo.getScriptId()));
                }
                insertData.setCreateTime(createTime);
                // 执行插入sql
                rmResourceRemoteService.insertRmResourceRemote(insertData);
                log.error("脚本执行失败:{}", rsp);
            }
        }
    }
    /**
     * 公共方法：处理网卡信息
     * @param registerMsg 注册消息
     * @param clientId 客户端ID
     * @param isRegister 是否为注册消息
     */
    private void processNetworkInterfaces(RegisterMsgVo registerMsg, String clientId, boolean isRegister) {
        // 时间戳转换
        long timestamp = registerMsg.getTimestamp();
        long millis = timestamp * 1000;
        Date createTime = new Date(millis / 1000 * 1000);

        List<NetworkInfo> networkInfoList = registerMsg.getNetworkInfo();
        if (networkInfoList.isEmpty()) {
            return;
        }

        boolean isSingleInterface = networkInfoList.size() == 1;

        // 查询数据库中当前的网卡数量
        RmNetworkInterface countQuery = new RmNetworkInterface();
        countQuery.setClientId(clientId);
        countQuery.setNewFlag(1);
        List<RmNetworkInterface> currentInterfaces = rmNetworkInterfaceService.selectRmNetworkInterfaceList(countQuery);
        // 构建数据库中当前网卡MAC地址的集合
        Set<String> currentMacSet = currentInterfaces.stream()
                .map(RmNetworkInterface::getMacAddress)
                .filter(Objects::nonNull)
                .collect(Collectors.toSet());

        // 构建新数据中网卡MAC地址的集合
        Set<String> newMacSet = networkInfoList.stream()
                .map(NetworkInfo::getMac)
                .filter(Objects::nonNull)
                .collect(Collectors.toSet());

        // 找出需要删除的网卡（数据库中有但新数据中没有的）
        Set<String> macsToDelete = new HashSet<>(currentMacSet);
        macsToDelete.removeAll(newMacSet);
        int currentInterfaceCount = currentInterfaces.size();
        // 判断网卡数量是否发生变化
        boolean interfaceCountChanged = false;
        if (!isRegister && currentInterfaceCount != networkInfoList.size()) {
            interfaceCountChanged = true;
        }
        // 执行删除操作
        if (!macsToDelete.isEmpty() && !isRegister) {
            deleteNetworkInterfaces(clientId, macsToDelete);
        }

        // 获取当前客户端的所有子网卡
        RmNetworkInterfaceChild childQuery = new RmNetworkInterfaceChild();
        childQuery.setClientId(clientId);
        List<RmNetworkInterfaceChild> existingChildren = rmNetworkInterfaceChildService.selectRmNetworkInterfaceChildList(childQuery);
        if(existingChildren == null){
            existingChildren = new ArrayList<>();
        }
        // 构建数据库中所有子网卡的key集合 (clientId_mac_name)
        Map<String, RmNetworkInterfaceChild> existingChildMap = new HashMap<>();
        for (RmNetworkInterfaceChild child : existingChildren) {
            String key = child.getClientId() + "_" + child.getMacAddress() + "_" + child.getInterfaceName();
            existingChildMap.put(key, child);
        }

        // 构建本次接收到的所有子网卡key集合
        Set<String> newChildKeys = new HashSet<>();

        for (NetworkInfo networkInfo : networkInfoList) {
            List<NetworkInfo> childList = networkInfo.getSubInterfaces();

            // 查询该网卡信息是否存在
            RmNetworkInterface queryParam = new RmNetworkInterface();
            queryParam.setClientId(clientId);
            queryParam.setMacAddress(networkInfo.getMac());
            queryParam.setNewFlag(1);
            List<RmNetworkInterface> exits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(queryParam);

            if (exits.isEmpty()) {
                // 新增网卡信息
                RmNetworkInterface insertData = new RmNetworkInterface();
                setNetworkInterfaceData(insertData, networkInfo, clientId);
                insertData.setCreateTime(createTime);
                // 设置bindIp
                if (isSingleInterface) {
                    insertData.setBindIp("3");
                }
                rmNetworkInterfaceService.insertRmNetworkInterface(insertData);

                // 处理子网卡
                if (childList != null && !childList.isEmpty()) {
                    for (NetworkInfo info : childList) {
                        RmNetworkInterfaceChild insertChild = new RmNetworkInterfaceChild();
                        setNetworkInterfaceChildData(insertChild, info, clientId, networkInfo.getName());
                        insertChild.setUpdateTime(DateUtils.getNowDate());
                        // 设置bindIp
                        if (isSingleInterface) {
                            insertChild.setBindIp("3");
                        }

                        // 构建子网卡唯一key
                        String childKey = clientId + "_" + info.getMac() + "_" + info.getName();
                        newChildKeys.add(childKey);

                        // 插入或更新子网卡
                        rmNetworkInterfaceChildService.insertRmNetworkInterfaceChild(insertChild);
                    }
                }

                // 如果网卡数量有变动，需要更新网卡绑定状态
                if (!isRegister && interfaceCountChanged && !isSingleInterface) {
                    RmResourceRegistrationRemote updateParam = new RmResourceRegistrationRemote();
                    updateParam.setClientId(clientId);
                    updateParam.setMultiPublicIpStatus("0");
                    remoteRevenueConfigService.updateStatusByResource(updateParam, SecurityConstants.INNER);
                }
            } else {
                // 更新网卡信息
                RmNetworkInterface oldInterfaceMsg = exits.get(0);

                // 处理子网卡
                if (childList != null && !childList.isEmpty()) {
                    for (NetworkInfo info : childList) {
                        RmNetworkInterfaceChild insertChild = new RmNetworkInterfaceChild();
                        setNetworkInterfaceChildData(insertChild, info, clientId, networkInfo.getName());
                        insertChild.setUpdateTime(DateUtils.getNowDate());
                        // 构建子网卡唯一key
                        String childKey = clientId + "_" + info.getMac() + "_" + info.getName();
                        newChildKeys.add(childKey);
                        // 插入或更新子网卡
                        rmNetworkInterfaceChildService.insertRmNetworkInterfaceChild(insertChild);
                    }
                }
                // 判断是否需要创建新记录
                boolean needCreateNew = (!StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName())
                        || !StringUtils.equals(networkInfo.getGateway(), oldInterfaceMsg.getGateway())) && StringUtils.equals(networkInfo.getMac(), oldInterfaceMsg.getMacAddress());
                // 更新现有记录
                updateNetworkInterface(networkInfo, clientId, oldInterfaceMsg, isSingleInterface, isRegister);
            }
        }

        // 删除数据库中不存在于本次接收数据中的子网卡
        // 计算需要删除的子网卡：数据库中有的，但本次没有接收到的
        Set<String> childKeysToDelete = new HashSet<>(existingChildMap.keySet());
        childKeysToDelete.removeAll(newChildKeys);

        if (!childKeysToDelete.isEmpty() && !isRegister) {
            for (String keyToDelete : childKeysToDelete) {
                String[] parts = keyToDelete.split("_", 3);
                if (parts.length == 3) {
                    String clientIdPart = parts[0];
                    String mac = parts[1];
                    String name = parts[2];

                    // 只删除当前客户端的子网卡
                    if (clientId.equals(clientIdPart)) {
                        deleteChildNetworkInterfaces(clientId, mac, name);
                    }
                }
            }
        }
    }
    /**
     * 删除过时的网卡信息
     * @param clientId 客户端ID
     * @param macsToDelete 需要删除的MAC地址集合
     */
    private void deleteNetworkInterfaces(String clientId, Set<String> macsToDelete) {
        for (String macAddress : macsToDelete) {
            RmNetworkInterface query = new RmNetworkInterface();
            query.setClientId(clientId);
            query.setMacAddress(macAddress);
            query.setNewFlag(1);
            List<RmNetworkInterface> oldExits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(query);
            if(!oldExits.isEmpty()) {
                oldExits.forEach(oldMsg -> {
                    rmNetworkInterfaceService.deleteRmNetworkInterfaceById(oldMsg.getId());
                });
            }
        }

        // 记录删除操作日志
        log.info("删除客户端 {} 的过时网卡信息，MAC地址: {}", clientId, macsToDelete);
    }
    /**
     * 删除过时的子网卡信息
     * @param clientId
     * @param name
     * @param mac
     */
    private void deleteChildNetworkInterfaces(String clientId, String mac, String name) {
        RmNetworkInterfaceChild query = new RmNetworkInterfaceChild();
        query.setClientId(clientId);
        query.setMacAddress(mac);
        query.setInterfaceName(name);
        query.setNewFlag(1);
        List<RmNetworkInterfaceChild> oldExits = rmNetworkInterfaceChildService.selectRmNetworkInterfaceChildList(query);
        if(!oldExits.isEmpty()) {
            oldExits.forEach(oldMsg -> {
                rmNetworkInterfaceChildService.deleteRmNetworkInterfaceChildById(oldMsg.getId());
            });
        }

        // 记录删除操作日志
        log.info("删除客户端 {} 的过时子网卡信息，MAC地址: {},网卡名称：{}", clientId, mac, name);
    }
    /**
     * 设置网卡信息公共字段
     */
    private void setNetworkInterfaceData(RmNetworkInterface networkInterface, NetworkInfo networkInfo, String clientId) {
        networkInterface.setClientId(clientId);
        networkInterface.setIsp(networkInfo.getCarrier());
        networkInterface.setCity(networkInfo.getCity());
        networkInterface.setGateway(networkInfo.getGateway());
        networkInterface.setInterfaceName(networkInfo.getName());
        networkInterface.setIpv4Address(networkInfo.getIpv4());
        networkInterface.setIpv6Address(networkInfo.getIpv6());
        networkInterface.setMacAddress(networkInfo.getMac());
        networkInterface.setProvince(networkInfo.getProvince());
        networkInterface.setPublicIp(networkInfo.getPublicIp());
        networkInterface.setInterfaceType(networkInfo.getType());
    }
    private void setNetworkInterfaceChildData(RmNetworkInterfaceChild networkInterface, NetworkInfo networkInfo, String clientId, String parentName) {
        networkInterface.setClientId(clientId);
        networkInterface.setParentInterface(parentName);
        networkInterface.setIsp(networkInfo.getCarrier());
        networkInterface.setCity(networkInfo.getCity());
        networkInterface.setGateway(networkInfo.getGateway());
        networkInterface.setInterfaceName(networkInfo.getName());
        networkInterface.setIpv4Address(networkInfo.getIpv4());
        networkInterface.setIpv6Address(networkInfo.getIpv6());
        networkInterface.setMacAddress(networkInfo.getMac());
        networkInterface.setProvince(networkInfo.getProvince());
        networkInterface.setPublicIp(networkInfo.getPublicIp());
        networkInterface.setInterfaceType(networkInfo.getType());
        networkInterface.setStatus(networkInfo.getStatus());
        networkInterface.setNetCreateTime(networkInfo.getNetCreateTime());
    }

    /**
     * 清理旧记录
     */
    private void cleanOldRecords(String clientId, String macAddress) {
        RmNetworkInterface query = new RmNetworkInterface();
        query.setClientId(clientId);
        query.setMacAddress(macAddress);
        query.setNewFlag(999);
        List<RmNetworkInterface> oldExits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(query);
        if(!oldExits.isEmpty()) {
            oldExits.forEach(oldMsg -> {
                rmNetworkInterfaceService.deleteRmNetworkInterfaceById(oldMsg.getId());
            });
        }
    }
    /**
     * 清理旧记录
     */
    private void cleanChildOldRecords(String clientId, String macAddress) {
        RmNetworkInterfaceChild childQuery = new RmNetworkInterfaceChild();
        childQuery.setClientId(clientId);
        childQuery.setMacAddress(macAddress);
        List<RmNetworkInterfaceChild> oldChildExits = rmNetworkInterfaceChildService.selectRmNetworkInterfaceChildList(childQuery);
        if(!oldChildExits.isEmpty()) {
            oldChildExits.forEach(oldMsg -> {
                rmNetworkInterfaceChildService.deleteRmNetworkInterfaceChildById(oldMsg.getId());
            });
        }
    }
    /**
     * 更新网卡信息
     */
    private void updateNetworkInterface(NetworkInfo networkInfo, String clientId,
                                        RmNetworkInterface oldInterfaceMsg, boolean isSingleInterface, boolean isRegister) {
        RmNetworkInterface updateData = new RmNetworkInterface();
        BeanUtils.copyProperties(oldInterfaceMsg, updateData);

        boolean needUpdate = false;

        // 逐个字段比较是否需要更新
        if (!StringUtils.equals(networkInfo.getCity(), oldInterfaceMsg.getCity())) {
            updateData.setCity(networkInfo.getCity());
            needUpdate = true;
        }
        if (!StringUtils.equals(networkInfo.getIpv4(), oldInterfaceMsg.getIpv4Address())) {
            updateData.setIpv4Address(networkInfo.getIpv4());
            needUpdate = true;
        }
        if (!StringUtils.equals(networkInfo.getIpv6(), oldInterfaceMsg.getIpv6Address())) {
            updateData.setIpv6Address(networkInfo.getIpv6());
            needUpdate = true;
        }
        if (!StringUtils.equals(networkInfo.getProvince(), oldInterfaceMsg.getProvince())) {
            updateData.setProvince(networkInfo.getProvince());
            needUpdate = true;
        }
        if (!StringUtils.equals(networkInfo.getPublicIp(), oldInterfaceMsg.getPublicIp())) {
            updateData.setPublicIp(networkInfo.getPublicIp());
            needUpdate = true;
        }
        if (!StringUtils.equals(networkInfo.getCarrier(), oldInterfaceMsg.getIsp())) {
            updateData.setIsp(networkInfo.getCarrier());
            needUpdate = true;
        }
        if (!StringUtils.equals(networkInfo.getType(), oldInterfaceMsg.getInterfaceType())) {
            updateData.setInterfaceType(networkInfo.getType());
            needUpdate = true;
        }
        if (!StringUtils.equals(networkInfo.getGateway(), oldInterfaceMsg.getGateway())) {
            updateData.setGateway(networkInfo.getGateway());
            needUpdate = true;
        }
        if (!StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName())) {
            if(networkInfo.getName() != null){
                // 添加业务变更记录
                EpsMethodChangeRecordVo recordAddData = new EpsMethodChangeRecordVo();
                recordAddData.setClientId(clientId);
                recordAddData.setTrafficPort(networkInfo.getName());
                recordAddData.setUpdateTime(DateUtils.getNowDate());
                recordAddData.setCreateTime(DateUtils.getNowDate());
                recordAddData.setUpdateBy(SecurityUtils.getUsername());
                recordAddData.setCreatBy(SecurityUtils.getUsername());
                StringBuilder content = new StringBuilder();
                content.append("流量网口设置为").append(networkInfo.getName());
                recordAddData.setChangeContent(content.toString());
                rmNetworkInterfaceService.addTrafficPortChangeRecord(recordAddData);
                updateData.setInterfaceName(networkInfo.getName());
                needUpdate = true;
            }
        }

        // 只有有字段变化时才执行更新
        if (needUpdate) {
            updateData.setClientId(clientId);
            updateData.setMacAddress(oldInterfaceMsg.getMacAddress());
            rmNetworkInterfaceService.updateNetMsgByMac(updateData);
        }
    }
    public String checkIPVersion(String ipAddress) {
        // 检查是否为 IPv4 地址
        if (ipAddress.contains(".")) {
            // 简单的格式验证：IPv4 应该有 4 个部分，每个部分用点分隔
            String[] parts = ipAddress.split("\\.");
            if (parts.length == 4) {
                try {
                    // 验证每个部分是否在 0-255 范围内
                    for (String part : parts) {
                        int num = Integer.parseInt(part);
                        if (num < 0 || num > 255) {
                            return "Invalid IP Address";
                        }
                    }
                    return "IPv4";
                } catch (NumberFormatException e) {
                    return "Invalid IP Address";
                }
            }
        }

        // 检查是否为 IPv6 地址
        if (ipAddress.contains(":")) {
            // 简单的格式验证：IPv6 应该包含冒号分隔的十六进制数
            String[] parts = ipAddress.split(":");
            if (parts.length >= 3 && parts.length <= 8) {
                try {
                    // 验证每个部分是否为有效的十六进制数
                    for (String part : parts) {
                        if (!part.isEmpty()) {
                            // 允许空的部分（表示连续的零）
                            Integer.parseInt(part, 16);
                        }
                    }
                    return "IPv6";
                } catch (NumberFormatException e) {
                    return "Invalid IP Address";
                }
            }
        }

        return "Invalid IP Address";
    }
    /**
     * 查询IP地址归属地信息，返回运营商和省份
     */
    private Map<String, String> queryIpLocation(String ip) {
        String apiUrl = "http://172.16.15.51:10000/?ip=" + ip;
        CloseableHttpClient httpClient = HttpClients.createDefault();
        HttpGet httpGet = new HttpGet(apiUrl);

        try {
            CloseableHttpResponse response = httpClient.execute(httpGet);

            if (response.getStatusLine().getStatusCode() == 200) {
                HttpEntity entity = response.getEntity();
                if (entity != null) {
                    String jsonResponse = EntityUtils.toString(entity, "UTF-8");

                    // 使用Map直接解析JSON
                    Map<String, Object> resultMap = JsonDataParser.parseJsonData(jsonResponse, Map.class);
                    if (resultMap != null) {
                        return extractLocationInfo(resultMap);
                    }
                }
            } else {
                System.err.println("获取IP归属地信息失败，HTTP状态码: " + response.getStatusLine().getStatusCode());
            }
        } catch (Exception e) {
            System.err.println("查询IP归属地时发生异常: " + e.getMessage());
            e.printStackTrace();
        } finally {
            try {
                httpClient.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    /**
     * 从JSON Map中提取运营商和省份
     */
    private Map<String, String> extractLocationInfo(Map<String, Object> resultMap) {
        Map<String, String> info = new HashMap<>();

        // 提取运营商（从as.info获取中文名称）
        if (resultMap.containsKey("as")) {
            Object asObj = resultMap.get("as");
            if (asObj instanceof Map) {
                Map<?, ?> asMap = (Map<?, ?>) asObj;
                if (asMap.containsKey("info")) {
                    info.put("operator", asMap.get("info").toString());
                }
            }
        }

        // 提取省份
        if (resultMap.containsKey("regions_short")) {
            Object regionsObj = resultMap.get("regions_short");
            if (regionsObj instanceof List && ((List<?>) regionsObj).size() > 0) {
                info.put("province", ((List<?>) regionsObj).get(0).toString());
            }
        } else if (resultMap.containsKey("regions")) {
            Object regionsObj = resultMap.get("regions");
            if (regionsObj instanceof List && ((List<?>) regionsObj).size() > 0) {
                String province = ((List<?>) regionsObj).get(0).toString();
                // 处理"浙江省" -> "浙江"
                if (province.endsWith("省")) {
                    province = province.substring(0, province.length() - 1);
                }
                info.put("province", province);
            }
        }

        return info;
    }
}