MiddlePlatform/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java

package com.maibu.service;

import java.io.IOException;
import java.lang.reflect.Field;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.maibu.common.NettyCacheKey;
import com.maibu.core.business.DeviceRunningStatusHistory;
import com.maibu.core.business.DeviceStatusRecordDTO;
import com.maibu.core.business.ErrorIdentificationStandard;
import com.maibu.core.business.alarm_center.AlarmMessage;
import com.maibu.core.business.device.NettyDevice;
import com.maibu.core.business.inter.WebsocketMesDispather;
import com.maibu.core.enums.CompareEnum;
import com.maibu.core.enums.ErrorSource;
import com.maibu.core.enums.RespondCode;
import com.maibu.core.redis.RedisCache;
import com.maibu.dto.DeviceErrorPushDTO;
import com.maibu.manager.DeviceSessionManager;
import com.maibu.mapper.ErrorIdentificationStandardMapper;
import com.maibu.memory.GlobalMemory;
import com.maibu.memory.SiteMemory;
import com.maibu.mqtt.MqttTopic;
import com.maibu.utils.CompareUtils;

import cn.hutool.core.io.IORuntimeException;
import cn.hutool.core.io.IoUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class DeviceThreadService {

    @Autowired
    private ResourceLoader resourceLoader;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private DeviceSessionManager sessionManager;

    @Autowired
    private ErrorIdentificationStandardMapper standardMapper;

    @Autowired
    private WebsocketMesDispather websocketMesDispather;

    @Autowired
    private AlertPushService alertPushService;

    public void deviceErrorMonitor() throws InterruptedException, IOException {
        initStandard();

        // TODO fix 后续是否分多个线程 一个site一个线程 
        // 有还需要更新接受故障方式。

        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleWithFixedDelay(() -> {
            try {
                String key = NettyCacheKey.deviceRunningStatusKey;
                Collection<String> keys = redisCache.getListKeyByPrefix(key);
                if (!CollectionUtils.isEmpty(keys)) {
                    List<DeviceErrorPushDTO> pushList = new ArrayList<>();
                    keys.forEach(k -> {
                        DeviceStatusRecordDTO recordDTO = redisCache.getCacheObject(k);
                        if (recordDTO != null && !CollectionUtils.isEmpty(recordDTO.getHistories())) {
                            recordDTO.getHistories()
                                    .sort(Comparator.comparing(DeviceRunningStatusHistory::getCreateTime).reversed());
                            DeviceRunningStatusHistory history = recordDTO.getHistories().get(0);

                            String[] parts = k.split(":");
                            String deviceId = parts[1];
                            NettyDevice slaveDevice = sessionManager.getDevice(deviceId);
                            Long siteId = slaveDevice != null ? slaveDevice.getDevice().getSiteId() : null;
                            Long orgId = slaveDevice != null ? slaveDevice.getDevice().getOrgId() : null;

                            SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
                            List<AlarmMessage> compared = compareErrorStandard(siteId, orgId, history,
                                    siteMemory.standards.get(ErrorSource.MOWER));

                            DeviceErrorPushDTO pushDTO = new DeviceErrorPushDTO();
                            pushDTO.setDeviceId(deviceId);
                            pushDTO.setTime(System.currentTimeMillis());
                            pushDTO.setDetails(compared);
                            pushDTO.setEvent(RespondCode.device_error_push);
                            pushList.add(pushDTO);
                        }
                    });
                    // 按实际deviceId分组，推送错误信息
                    Map<String, List<DeviceErrorPushDTO>> pushMap = pushList.stream()
                            .collect(java.util.stream.Collectors.groupingBy(DeviceErrorPushDTO::getDeviceId));
                    pushMap.forEach((deviceId, dtos) -> {
                        // pushErrorMessage(deviceId, JSON.toJSONString(dtos));
                        // todo 判断错误级别，是否需要推送邮件 或者钉钉。
                        // 找到这个设备的场站 添加场站是否开启关闭 钉钉或者邮件推送，推送邮件 添加人员是否订阅邮件，钉钉是加入钉钉群
                        alertPushService.messagePushHandler(deviceId, dtos);
                    });
                }
            } catch (Exception e) {
                log.error("执行监测错误线程错误：{}", e.getMessage());
            }
        }, 0, 5, TimeUnit.SECONDS);
    }

    public void initStandard() throws IOException {
        GlobalMemory.getAllSiteMemory().forEach(siteMemory -> {
            if (siteMemory != null && siteMemory.standards == null) {
                LambdaQueryWrapper<ErrorIdentificationStandard> queryWrapper = new LambdaQueryWrapper<>();
                queryWrapper.eq(ErrorIdentificationStandard::getSiteId, siteMemory.sysSite.getId());
                List<ErrorIdentificationStandard> list = standardMapper.selectList(queryWrapper);
                if (CollectionUtils.isEmpty(list)) {
                    try {
                        // 读取resource目录下的device-config.json文件
                        Resource resource = resourceLoader.getResource("classpath:errorStandard.json");
                        // 使用fastjson2转换为DeviceConfig对象
                        String jsonContent;
                        jsonContent = IoUtil.readUtf8(resource.getInputStream());
                        List<ErrorIdentificationStandard> newStandards = JSON.parseArray(jsonContent,
                                ErrorIdentificationStandard.class);
                        if (!CollectionUtils.isEmpty(newStandards)) {
                            newStandards.forEach(x -> {
                                x.setOrgId(siteMemory.sysSite.getOrgId());
                                x.setSiteId(siteMemory.sysSite.getId());
                            });
                            standardMapper.insertBatch(newStandards);
                            siteMemory.standards = newStandards.stream()
                                    .filter(e -> e.getErrorSource() != null)
                                    .collect(Collectors.groupingBy(
                                            ErrorIdentificationStandard::getErrorSource,
                                            ConcurrentHashMap::new,
                                            Collectors.toList()));

                        }
                    } catch (IORuntimeException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                } else {
                    siteMemory.standards = list.stream()
                            .filter(e -> e.getErrorSource() != null)
                            .collect(Collectors.groupingBy(
                                    ErrorIdentificationStandard::getErrorSource,
                                    ConcurrentHashMap::new,
                                    Collectors.toList()));
                }
            }
        });
    }

    public void pushErrorMessage(String deviceId, String dto) {
        // NettyDevice device = sessionManager.getDevice(deviceId);
        // if (device != null) {
        // ByteBuf buf = Unpooled.buffer();
        // CommandUtils.buildCommand(buf, dto, CommandConstant.interaction);
        // if (device.getChannel() != null && device.getChannel().isActive()) {
        // device.getChannel().writeAndFlush(buf);
        // }
        // }
        try {
            GlobalMemory.mqttClientUtil.publish(String.format(MqttTopic.DEVICE_ERROR_PUSH_TOPIC, deviceId), dto);
            List<NettyDevice> controlMasters = sessionManager.getAllSlaveControl(deviceId);
            if (!CollectionUtils.isEmpty(controlMasters)) {
                controlMasters.forEach(x -> {
                    websocketMesDispather.dispather(x.getConnectorId(), dto);
                    log.info("推送错误信息 device:{},content:{}", deviceId, dto);
                });
            }
        } catch (MqttException e) {
            log.error("推送错误信息失败 device:{}error:{}", deviceId, e.getMessage());
        }
    }

    public List<AlarmMessage> compareErrorStandard(Long siteId, Long orgId, DeviceRunningStatusHistory history,
            List<ErrorIdentificationStandard> standards) {
        if (history == null || CollectionUtils.isEmpty(standards))
            return null;
        List<AlarmMessage> list = new ArrayList<>();
        standards.forEach(standard -> {
            String fieldName = standard.getField();
            String compareValues = standard.getCompareValues();
            CompareEnum compareType = standard.getCompareType();
            String targetValue = knownClassFieldFinding(fieldName, history);
            if (!StringUtils.isEmpty(compareValues) && !StringUtils.isEmpty(targetValue)) {
                boolean result = CompareUtils.compare(compareType, compareValues, targetValue);
                if (!result) {
                    AlarmMessage alarm = new AlarmMessage();
                    alarm.setContent(standard.getErrorDescription());
                    alarm.setDeviceSn(history.getDeviceId());
                    alarm.setDeviceName(history.getDeviceId());
                    alarm.setLevel(standard.getErrorLevel());
                    alarm.setLocation(history.getLatitude() + "," + history.getLongitude());
                    alarm.setName(standard.getErrorName());
                    alarm.setNo(standard.getErrorCode());
                    alarm.setSuggestion(standard.getSuggestion());
                    alarm.setStatus("未处理");
                    alarm.setTime(LocalDateTime.now());
                    alarm.setOrgId(orgId);
                    alarm.setSiteId(siteId);
                    list.add(alarm);
                }
            }
        });
        return list;
    }

    private static String knownClassFieldFinding(String fieldName, DeviceRunningStatusHistory history) {
        try {
            // 1. 获取类的Class对象
            Class<?> historyClass = DeviceRunningStatusHistory.class;
            Field field = historyClass.getDeclaredField(fieldName);
            field.setAccessible(true);
            Object value = field.get(history);
            return (String) value;
        } catch (NoSuchFieldException | IllegalAccessException e) {
            System.out.println("获取字段失败: " + e.getMessage());
            log.error("获取属性值失败:{}", e.getMessage());
        }
        return null;
    }

}