pipe-ner-server/radar-service/src/main/java/com/zksy/radar/utils/MessageHandler.java

package com.zksy.radar.utils;

import static com.zksy.common.utils.CollectionSideEffects.*;

import com.zksy.api.domain.AlarmData;
import com.zksy.api.domain.WarningThreshold;
import com.zksy.api.service.AlarmDataService;
import com.zksy.api.service.EarlyWarningAutoService;
import com.zksy.api.service.WarningThresholdService;
import com.zksy.api.utils.AlarmLevelUtil;
import com.zksy.api.utils.SmsUtil;
import com.zksy.common.exception.InvalidMessageException;
import com.zksy.radar.domain.RadarData;
import com.zksy.radar.service.RadarDataService;
import com.zksy.api.service.DeviceContactService;
import io.netty.buffer.ByteBuf;
import io.netty.util.ReferenceCountUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.ReadTimeoutException;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@ChannelHandler.Sharable
@Slf4j
@Component
public class MessageHandler extends ChannelInboundHandlerAdapter {
	private static Logger logger = LoggerFactory.getLogger(MessageHandler.class);
	private final RadarDataService service;
	private final WarningThresholdService warningThresholdService;
	private final AlarmDataService alarmDataService;
	private final EarlyWarningAutoService earlyWarningAutoService;
	private final SmsUtil smsUtil;
	private final DeviceContactService devicePhoneFetchUtil;
    private final RestTemplate restTemplate;

    @Autowired
    private DeviceOfflineCheckTask deviceOfflineCheckTask;

    @Value("${device.status.api.url:http://zk-api-service/equipmentStatus/updateAlarmStatus}")
    private String deviceStatusApiUrl;

    @Autowired
    public MessageHandler(RadarDataService RadarDataService, WarningThresholdService warningThresholdService,
                           AlarmDataService alarmDataService, EarlyWarningAutoService earlyWarningAutoService,
                           SmsUtil smsUtil, DeviceContactService devicePhoneFetchUtil,
                           RestTemplate restTemplate) {
        this.service = RadarDataService;
        this.warningThresholdService = warningThresholdService;
        this.alarmDataService = alarmDataService;
        this.earlyWarningAutoService = earlyWarningAutoService;
        this.smsUtil = smsUtil;
        this.devicePhoneFetchUtil = devicePhoneFetchUtil;
        this.restTemplate = restTemplate;
    }

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		super.channelActive(ctx);
		//sendDataToDevice(ctx);
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		try {
            ByteBuf in = (ByteBuf) msg;
            byte[] msgBytes = new byte[in.readableBytes()];
            in.readBytes(msgBytes);
			logger.debug("接收到的原始数据帧: {}", printHexBinary(msgBytes));

			// 提取帧类型
			byte frameType = msgBytes[6];
			logger.debug("帧类型: 0x{}", String.format("%02X", frameType));

			// 1. 完整协议校验
			DataParser.validateMessage(msgBytes);
			logger.info("数据帧校验通过");

			// 2. 数据解析
			RadarData resultData = DataParser.parseMessage(msgBytes);
			if (resultData.getSystemIdentifier() == null) {
				resultData.setSystemIdentifier("123456");
			}

			// 3. 根据帧类型处理
			if (frameType == 0x31) {
				if (msgBytes.length > 34) {
					byte flagByte = msgBytes[34];
					String binary = String.format("%8s", Integer.toBinaryString(flagByte & 0xFF)).replace(' ', '0');
					boolean isEnd = (flagByte & 0x08) != 0;

					logger.debug("结束位检测: 字节=0x{}, 二进制={}, 第5位(结束标识)={}",
							String.format("%02X", flagByte), binary, isEnd ? "1(结束)" : "0(未结束)");
				} else {
					logger.warn("数据长度不足，无法检测结束位");
				}
				// 入库
				requireSaved(service.save(resultData));
                String deviceCode = resultData.getDeviceCode();
                run(deviceCode, "接收时间更新", () ->
                        DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(deviceCode, new java.util.Date()));
                run(deviceCode, "采集应答", () -> {
                    byte[] response = ProtocolUtils.buildCustomReplyFrame(msgBytes);
                    ctx.writeAndFlush(Unpooled.copiedBuffer(response));
                    if (Boolean.TRUE.equals(resultData.getIsLastPacket())) {
                        byte[] endResponse = ProtocolUtils.buildEndReplyFrame(msgBytes);
                        ctx.executor().schedule(() -> run(deviceCode, "结束应答", () -> {
                            if (ctx.channel().isActive()) {
                                ctx.writeAndFlush(Unpooled.copiedBuffer(endResponse));
                            }
                        }), 1, java.util.concurrent.TimeUnit.SECONDS);
                    }
                });
                run(deviceCode, "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(deviceCode));
                run(deviceCode, "报警预警评估", () -> handleRadarAlarm(resultData));

			} else if (frameType == 0x34) {
				/*logger.info("收到结束通讯帧，准备发送回应包");
				byte[] response = ProtocolUtils.buildShutdownAckPacket(msgBytes);
				ctx.writeAndFlush(Unpooled.copiedBuffer(response));
				logger.debug("已发送结束通讯回应包");*/

			} else {
				logger.warn("收到未知帧类型: 0x{}", String.format("%02X", frameType));
			}

		} catch (InvalidMessageException e) {
			logger.error("数据校验失败，不入库: {}", e.getMessage());
			sendErrorResponse(ctx, "数据校验失败");

		} catch (Exception e) {
			logger.error("数据解析或入库异常", e);
			sendErrorResponse(ctx, "数据处理异常");
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }


	// 工具方法：发送错误响应
	private void sendErrorResponse(ChannelHandlerContext ctx, String msg) {
		ctx.writeAndFlush(Unpooled.copiedBuffer(("数据处理失败: " + msg).getBytes()));
	}
	private String printHexBinary(byte[] bytes) {
		StringBuilder sb = new StringBuilder();
		for (byte b : bytes) {
			sb.append(String.format("%02X ", b));
		}
		System.out.println("Received raw data: " + sb.toString());
		logger.info("Received raw data: " + sb.toString());
		return sb.toString();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		if (cause instanceof ReadTimeoutException) {
			logger.info("来自" + ctx.channel().remoteAddress() + "的连接超时断开");
		} else {
			logger.error("设备连接异常", cause);
			logger.info("来自" + ctx.channel().remoteAddress() + "的连接异常断开");
			ctx.close();
		}
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		ctx.flush();
	}

	@Override
	public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
		logger.info("来自" + ctx.channel().remoteAddress() + "的连接主动断开");
		ctx.fireChannelUnregistered();
	}

	// 主动向设备发送数据
	public void sendDataToDevice(ChannelHandlerContext ctx) {
		try {
			// 将字符串形式的十六进制数据转换为字节数组
			String hexData = "01 03 00 00 00 01 84 0A ";
			String[] hexArray = hexData.split(" ");
			byte[] dataBytes = new byte[hexArray.length];
			for (int i = 0; i < hexArray.length; i++) {
				dataBytes[i] = (byte) Integer.parseInt(hexArray[i], 16);
			}
			ByteBuf byteBuf = Unpooled.copiedBuffer(dataBytes);
			ctx.writeAndFlush(byteBuf);
			logger.info("已向设备发送数据: {}", printHexBinary(dataBytes));
		} catch (Exception e) {
			logger.error("发送数据失败: {}", e.getMessage());
		}
	}

	/**
		 * 处理雷达设备告警
		 */
    private void handleRadarAlarm(RadarData resultData) {
        String deviceCode = resultData.getDeviceCode();
        Integer status = alarmStatus(
                handleAlarm(deviceCode, "WARN_INSTANT_FLOW", "瞬时流量", resultData.getMeter1InstantFlow()),
                handleAlarm(deviceCode, "WARN_FLOW_SPEED", "流速", resultData.getFlowSpeed()));
        if (status != null) {
            updateDeviceAlarmStatus(deviceCode, status);
        }
    }

	/**
		 * 处理单个告警
		 * @return 是否触发告警
		 */
    private Boolean handleAlarm(String deviceCode, String warningCode, String defaultType, String valueStr) {
        if (valueStr == null) {
            return null;
        }
        return evaluate(deviceCode, warningCode, () -> {
            BigDecimal actual = new BigDecimal(valueStr);
            WarningThreshold threshold = warningThresholdService.getWarningThresholdByDeviceAndCode(deviceCode, warningCode);
            if (threshold == null || (threshold.getMinValue() == null && threshold.getMaxValue() == null)) {
                throw new IllegalStateException("阈值缺失，不能判断设备已恢复正常");
            }
            String type = defaultType;
            BigDecimal min = threshold.getMinValue() == null ? null : BigDecimal.valueOf(threshold.getMinValue());
            BigDecimal max = threshold.getMaxValue() == null ? null : BigDecimal.valueOf(threshold.getMaxValue());
            boolean over = (min != null && actual.compareTo(min) <= 0) || (max != null && actual.compareTo(max) >= 0);
            if (over) {
                run(deviceCode, warningCode + ":报警入库及通知", () -> saveAlarmData(deviceCode, type, warningCode, min, max, actual, null));
                run(deviceCode, warningCode + ":关闭预警", () -> earlyWarningAutoService.closeAutoWarning(deviceCode, warningCode, "已达报警阈值，转报警处理"));
            } else {
                run(deviceCode, warningCode + ":告警恢复", () -> alarmDataService.recoverThresholdAlarm(deviceCode, warningCode, min, max, actual));
                run(deviceCode, warningCode + ":生成预警", () -> earlyWarningAutoService.checkAndGenerate(deviceCode, type, warningCode, min, max, actual, null));
            }
            return over;
        });
    }

	/**
	 * 保存告警数据到 alarm_data 表
	 */
	private void saveAlarmData(String deviceCode, String warningType, String warningCode,
	                           BigDecimal minValue, BigDecimal maxValue, BigDecimal actualValue, String remark) {
		try {
			AlarmData alarmData = new AlarmData();
			alarmData.setDeviceCode(deviceCode);
			alarmData.setWarningCode(warningCode);
			alarmData.setMinValue(minValue);
			alarmData.setMaxValue(maxValue);
			alarmData.setActualValue(actualValue);

			int alarmLevel = AlarmLevelUtil.calculateLevel(actualValue, minValue, maxValue);
			BigDecimal deviationRatio = AlarmLevelUtil.calculateDeviationRatio(actualValue, minValue, maxValue);
			alarmData.setAlarmLevel(alarmLevel);
			alarmData.setDeviationRatio(deviationRatio);

			alarmData.setAlarmStatus(0);
			alarmData.setCreateTime(LocalDateTime.now());
			alarmData.setCreateTime(LocalDateTime.now());
			boolean saved = run(deviceCode, warningCode + ":报警入库", () -> requireSaved(alarmDataService.saveAlarmData(alarmData)));
			if (saved) { logger.info("告警数据入库成功: deviceCode={}, warningCode={}, level={}级", deviceCode, warningCode, alarmLevel); }

			if (alarmLevel == 1) {
				logger.warn("【1级紧急告警】deviceCode={}, warningCode={}, deviationRatio={}%",
						deviceCode, warningCode, deviationRatio);
				sendAlarmSms(deviceCode, warningType, actualValue, minValue, maxValue);
			}
		} catch (Exception e) {
			logger.error("保存告警数据失败", e);
		}
	}

	private void sendAlarmSms(String deviceCode, String warningType, BigDecimal actualValue,
	                          BigDecimal minValue, BigDecimal maxValue) {
		try {
			List<String> phoneList = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceCode);
			if (phoneList == null || phoneList.isEmpty()) {
				logger.warn("报警短信发送失败：未找到设备{}的维护人员手机号", deviceCode);
				return;
			}
			com.alibaba.fastjson.JSONObject params = new com.alibaba.fastjson.JSONObject();
			params.put("deviceNo", deviceCode);
			params.put("alarmType", warningType);
			params.put("location", "当前值：" + actualValue + "，正常范围：" + minValue + "~" + maxValue);

			Map<String, Boolean> result = smsUtil.sendBatchSms(phoneList, params.toJSONString());
			long successCount = result.values().stream().filter(Boolean::booleanValue).count();
			logger.info("1级告警短信发送完成，成功{}条，失败{}条", successCount, phoneList.size() - successCount);
		} catch (Exception e) {
			logger.error("发送报警短信异常: {}", e.getMessage());
			}
		}

		/**
		 * 更新设备报警状态
		 * @param deviceCode 设备编码
		 * @param alarmStatus 报警状态：0-正常，1-报警
		 */
		private void updateDeviceAlarmStatus(String deviceCode, int alarmStatus) {
			try {
				Map<String, Object> params = new HashMap<>();
				params.put("deviceCode", deviceCode);
				params.put("alarmStatus", alarmStatus);

				HttpHeaders headers = new HttpHeaders();
				headers.setContentType(MediaType.APPLICATION_JSON);
				HttpEntity<Map<String, Object>> request = new HttpEntity<>(params, headers);

				Map<?, ?> response = restTemplate.postForObject(deviceStatusApiUrl, request, Map.class);
				if (response == null || !Boolean.TRUE.equals(response.get("success"))) {
					logger.warn("设备报警状态更新未成功: deviceCode={}, alarmStatus={}, response={}", deviceCode, alarmStatus, response);
					return;
				}
				logger.info("设备报警状态更新成功: deviceCode={}, alarmStatus={}", deviceCode, alarmStatus);
			} catch (Exception e) {
				logger.error("更新设备报警状态失败: deviceCode={}, alarmStatus={}", deviceCode, alarmStatus, e);
			}
		}
	}