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

package com.zksy.gas.utils;

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

import cn.hutool.core.lang.UUID;
import com.alibaba.fastjson.JSONObject;
import com.zksy.api.domain.WarningThreshold;
import com.zksy.api.service.EarlyWarningAutoService;
import com.zksy.api.service.WarningThresholdService;
import com.zksy.api.domain.AlarmData;
import com.zksy.api.service.AlarmDataService;
import com.zksy.api.utils.AlarmLevelUtil;
import com.zksy.api.utils.SmsUtil;
import com.zksy.common.exception.InvalidMessageException;
import com.zksy.gas.domain.GasMonitorData;
import com.zksy.gas.service.GasMonitorDataService;
import com.zksy.api.service.DeviceContactService;
import io.netty.buffer.ByteBuf;
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 io.netty.util.ReferenceCountUtil;
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 GasMonitorDataService service;
	@Autowired
	private WarningThresholdService warningThresholdService;
	@Autowired
	private AlarmDataService alarmDataService;
	@Autowired
	private EarlyWarningAutoService earlyWarningAutoService;
	@Autowired
	private SmsUtil smsUtil;
	@Autowired
    private DeviceContactService devicePhoneFetchUtil;
    @Autowired
    private AlarmUtil alarmUtil;
    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private DeviceOfflineCheckTask deviceOfflineCheckTask;

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

	@Autowired
	public MessageHandler(GasMonitorDataService firefightingPressureService) {
		this.service = firefightingPressureService;
	}

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

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		try {
			ByteBuf msgByteBuf = (ByteBuf) msg;
			if (msgByteBuf == null || !msgByteBuf.isReadable()) {
				logger.warn("接收到无效的消息");
				return;
			}
			int readable = msgByteBuf.readableBytes();
			byte[] msgBytes = new byte[readable];
			msgByteBuf.readBytes(msgBytes);
			logger.info("接收到 {} 字节的数据: {}, 来自: {}",
					readable, ProtocolUtils.bytesToHex(msgBytes), ctx.channel().remoteAddress());

			int checksumTotalLength = 2; // 末尾2字节校验码（如ED34）
			int bodyStartOffset = 16;    // 校验和07的主体起始索引
			int fullDataStartOffset = 3; // 校验和16的完整数据起始索引

			// 校验报文长度合法性
			if (readable < fullDataStartOffset + checksumTotalLength) {
				throw new InvalidMessageException("报文长度不足，无法完成校验");
			}

			int bodyEndOffset = msgBytes.length - checksumTotalLength; // 避开末尾2字节校验码
			int bodyLength = bodyEndOffset - bodyStartOffset;
			if (bodyLength <= 0) {
				throw new InvalidMessageException("主体数据长度无效，无法计算校验和07");
			}
			int checksum07 = DataCheckUtil.calcSumOnly(msgBytes, bodyStartOffset, bodyLength);

			int fullDataEndOffset = msgBytes.length - 1; // 包含倒数第二个字节
			int fullDataLength = fullDataEndOffset - fullDataStartOffset;
			if (fullDataLength <= 0) {
				throw new InvalidMessageException("完整数据长度无效，无法计算校验和16");
			}
			int checksum16 = DataCheckUtil.calcChecksum(msgBytes, fullDataStartOffset, fullDataLength);

			int calculatedChecksum = (checksum07 << 8) | checksum16;
			int receivedChecksum = ((msgBytes[msgBytes.length - 2] & 0xFF) << 8)
					| (msgBytes[msgBytes.length - 1] & 0xFF);

			// 打印调试信息，确认拼接结果
			logger.debug("校验和07（高8位，仅求和）: {}", Integer.toHexString(checksum07).toUpperCase());
			logger.debug("校验和16（低8位，FF-求和）: {}", Integer.toHexString(checksum16).toUpperCase());
			logger.debug("修改后拼接校验码: {}", Integer.toHexString(calculatedChecksum).toUpperCase());
			logger.debug("报文携带校验码: {}", Integer.toHexString(receivedChecksum).toUpperCase());

			if (calculatedChecksum != receivedChecksum) {
				throw new InvalidMessageException(
						String.format("数据校验失败：本地计算（07+16）=%04X（07=%02X, 16=%02X），报文携带=%04X",
								calculatedChecksum, checksum07, checksum16, receivedChecksum));
			}

			com.zksy.gas.domain.GasMonitorData resultData = DataParser.parseMessage(msgBytes);
			resultData.setId(java.util.UUID.randomUUID().toString());
			requireSaved(service.save(resultData));
			// 更新设备最后接收时间，并标记设备在线
			run(resultData.getMacAddress(), "接收时间更新", () -> DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(resultData.getMacAddress(), new java.util.Date()));
			run(resultData.getMacAddress(), "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(resultData.getMacAddress()));
			run(null, "报警预警评估", () -> checkIfSmsAlertNeeded(resultData));
			logger.info("数据解析入库成功: {}", resultData);

		} catch (InvalidMessageException e) {
			logger.error("数据校验失败: {}", e.getMessage());
			ctx.writeAndFlush(Unpooled.copiedBuffer("数据校验失败".getBytes()));
		} catch (Exception e) {
			logger.error("数据解析/入库异常", e);
			ctx.writeAndFlush(Unpooled.copiedBuffer("数据处理异常".getBytes()));
		} finally {
			ReferenceCountUtil.release(msg);
		}
	}

	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());
		}
	}
	public void checkIfSmsAlertNeeded(GasMonitorData resultData){
		String deviceId = resultData == null ? null : resultData.getMacAddress();
		try {
			if (resultData == null || deviceId == null || deviceId.trim().isEmpty()) {
				log.warn("可燃气体报警数据缺少设备编码，跳过状态覆盖");
				return;
			}
			boolean hasAlarm = evaluateAlarmStatus(resultData);
			updateDeviceAlarmStatus(deviceId, hasAlarm ? 1 : 0);
		} catch (Exception e) {
			// 评估不完整时不覆盖上一状态，避免异常数据误清除报警
			log.error("可燃气体设备报警处理失败，保留原报警状态：deviceCode={}", deviceId, e);
		}
	}

	/** 评估单条采集数据是否存在报警。只有完整评估成功才允许同步设备状态。 */
	private boolean evaluateAlarmStatus(GasMonitorData resultData){
	        //用MAC地址作为设备的设备编号
	        String deviceId = resultData.getMacAddress();
	        boolean hasAlarm = false;

		//获取气体浓度阈值（按气体类型只监测浓度，当前仅CH4；且仅按上限判定）
		String gasWarningCode = "WARN-GAS-CONCENTRATION";
		WarningThreshold gasThreshold = checkThreshold(deviceId, gasWarningCode);
		Double gasMaxValue = gasThreshold != null ? gasThreshold.getMaxValue() : null;
        String gasWarningType = "气体浓度预警";
        String gasRemark = null;

		//判断气体浓度：仅当浓度达到或超过上限时视为越限；浓度低（接近0）属正常，不告警
		BigDecimal gasConcentration = resultData.getGasConcentration();
        if (gasConcentration == null) { throw new IllegalArgumentException("缺少气体浓度值"); }
		if (gasConcentration != null && gasMaxValue != null) {
			BigDecimal gasMaxBd = BigDecimal.valueOf(gasMaxValue);
			if (gasConcentration.floatValue() >= gasMaxValue) {
				// 越限报警：等级计算以0为下限基线（气体浓度不会低于0），保证仅上限时也能正确分级
				int alarmLevel = saveAlarmData(deviceId, gasWarningType, gasWarningCode,
						BigDecimal.ZERO, gasMaxBd, gasConcentration,
						gasRemark != null ? gasRemark : "可燃气体浓度报警");
				run(deviceId, gasWarningCode + ":关闭预警", () -> earlyWarningAutoService.closeAutoWarning(deviceId, gasWarningCode, "已达报警阈值，转报警处理"));
				hasAlarm = true;
				if (alarmLevel == 1) {
					run(deviceId, gasWarningCode + ":短信通知", () -> sendMessage(deviceId, gasWarningType, resultData.getLongitude(), resultData.getLatitude()));
				}
			} else {
				// 未越限：接近上限时生成自动预警（min传null，仅按上限评估预警区）
                run(deviceId, gasWarningCode + ":告警恢复", () -> alarmDataService.recoverThresholdAlarm(
                        deviceId, gasWarningCode, null, gasMaxBd, gasConcentration));
				run(deviceId, gasWarningCode + ":生成预警", () -> earlyWarningAutoService.checkAndGenerate(deviceId, gasWarningType, gasWarningCode,
                        null, gasMaxBd, gasConcentration, gasRemark));
			}
		}

		return hasAlarm;
	}

    private WarningThreshold checkThreshold(String deviceCode, String warningCode) {
        return warningThresholdService.getWarningThresholdByDeviceAndCode(deviceCode, warningCode);
    }

	private int saveAlarmData(String deviceCode, String warningType, String warningCode,
							  BigDecimal minValue, BigDecimal maxValue,
							  BigDecimal actualValue, String remark) {
		int alarmLevel = 4;
		try {
			AlarmData alarmData = new AlarmData();
			alarmData.setDeviceCode(deviceCode);
			
			alarmData.setWarningCode(warningCode);
			alarmData.setMinValue(minValue);
			alarmData.setMaxValue(maxValue);
			alarmData.setActualValue(actualValue);

			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 (alarmLevel == 1) {
				log.warn("【1级紧急告警】deviceCode={}, warningCode={}, deviationRatio={}%",
						deviceCode, warningCode, deviationRatio);
			}
			if (saved) { log.info("告警数据入库成功: deviceCode={}, warningCode={}, level={}级", deviceCode, warningCode, alarmLevel); }
		} catch (Exception e) {
			log.error("保存告警数据失败", e);
		}
		return alarmLevel;
	}
	//构造发送短信参数&&发送短信
	public void sendMessage(String deviceNo, String alarmType, BigDecimal lng, BigDecimal lat){
		// 构造短信参数
		JSONObject params = new JSONObject();
		params.put("deviceNo", deviceNo);
		params.put("alarmType", alarmType);
		// 经纬度处理（位置）
		params.put("location", String.format("经纬度：%.6f,%.8f", lng, lat));
		List<String> phoneListByDeviceId = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceNo);
		Map<String, Boolean> sendResults = smsUtil.sendBatchSms(phoneListByDeviceId, params.toJSONString());
		log.info("发送信息完成");
    }

    /**
     * 更新设备报警状态
     * @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"))) {
                log.warn("设备报警状态更新未成功: deviceCode={}, alarmStatus={}, response={}", deviceCode, alarmStatus, response);
                return;
            }
            log.info("设备报警状态更新成功: deviceCode={}, alarmStatus={}", deviceCode, alarmStatus);
        } catch (Exception e) {
            log.error("更新设备报警状态失败: deviceCode={}, alarmStatus={}", deviceCode, alarmStatus, e);
        }
    }
}