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

package com.zksy.pressure.utils;

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

import cn.hutool.core.lang.UUID;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.apache.hc.client5.http.utils.Hex;
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.EarlyWarningLevelUtil;
import com.zksy.api.utils.SmsUtil;
import com.zksy.common.exception.InvalidMessageException;
import com.zksy.pressure.domain.FirefightingPressure;
import com.zksy.pressure.service.FirefightingPressureService;
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.Date;
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);

	/** 压力值预警编码，阈值取自 app_user.warning_threshold(device_code + warning_code) */
	private static final String PRESSURE_WARNING_CODE = "WARN-FIRE-PRESSURE";
	private static final String PRESSURE_WARNING_TYPE = "消防压力监测";

	private final FirefightingPressureService firefightingPressureService;
	@Autowired
	private AlarmDataService alarmDataService;
	@Autowired
	private WarningThresholdService warningThresholdService;
	@Autowired
	private EarlyWarningAutoService earlyWarningAutoService;
	@Autowired
	private SmsUtil smsUtil;
	@Autowired
    private DeviceContactService devicePhoneFetchUtil;
    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private DeviceOfflineCheckTask deviceOfflineCheckTask;

    @Value("${device.status.api.url}")
    private String deviceStatusApiUrl;
	@Autowired
	public MessageHandler(FirefightingPressureService firefightingPressureService) {
		this.firefightingPressureService = 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;
			}
			logger.info("接收到 {} 字节的数据，来自: {}", msgByteBuf.readableBytes(), ctx.channel().remoteAddress());
			byte[] msgBytes = new byte[msgByteBuf.readableBytes()];
			msgByteBuf.readBytes(msgBytes);
			String result = printHexBinary(msgBytes);
			String msgString = result.replaceAll("\\s", "");
			String CRCString = msgString.substring(msgString.length() - 4);
			String bodyResult = msgString.substring(0, msgString.length() - 4);
			String codeString = DataCheckUtil.crc16Tall(bodyResult);
			System.out.println("校验码：" + CRCString + "-----------生成的校验码：" + codeString);
			if (!CRCString.equals(codeString)) {
				throw new InvalidMessageException("数据校验不成功");
			} else {
				String hexString = Hex.encodeHexString(msgBytes).toUpperCase();
				String uploadEndFlag = hexString.substring(hexString.length() - 6, hexString.length() - 4);
				logger.info("成功！！！接收到 {} 字节的数据，来自: {}", msgString, ctx.channel().remoteAddress());
				FirefightingPressure resultData = DataParser.parseMessage(msgString);

				String functionCode = msgString.substring(20, 22);
				logger.info("生成的回复报文功能码: {}", functionCode);
				if("EF".equals(functionCode)){
					String responseMsg = buildResponseMessage(resultData.getTelemeteringStation(), resultData.getCentralStation(), resultData.getPassword(),resultData.getFunctionCode(), "0000", uploadEndFlag);
					logger.info("生成的回复报文: {}", responseMsg);
					byte[] messageBytes = hexStringToByteArray(responseMsg);
					ctx.channel().writeAndFlush(Unpooled.copiedBuffer(messageBytes));
					return;
				}

				resultData.setId(UUID.randomUUID().toString());
				// 更新设备最后一次接收数据的时间，并标记设备在线
				run(resultData.getTelemeteringStation(), "接收时间更新", () -> DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(resultData.getTelemeteringStation(), new Date()));
				requireSaved(firefightingPressureService.save(resultData));

                run(resultData.getTelemeteringStation(), "采集应答", () -> {
				// 4. 构建回复报文
				String responseMsg = buildResponseMessage(resultData.getTelemeteringStation(), resultData.getCentralStation(), resultData.getPassword(), resultData.getFunctionCode(), resultData.getSerialNumber(), uploadEndFlag);
				logger.info("生成的回复报文: {}", responseMsg);
				byte[] messageBytes = hexStringToByteArray(responseMsg);
				ctx.channel().writeAndFlush(Unpooled.copiedBuffer(messageBytes));
                });
                run(resultData.getTelemeteringStation(), "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(resultData.getTelemeteringStation()));
				// 阈值监测：报警 + 接近阈值自动预警
				run(null, "报警预警评估", () -> checkIfSmsAlertNeeded(resultData));
			}
		} catch (InvalidMessageException e) {
			logger.error("数据入库失败: {}", e.getMessage());
			ctx.writeAndFlush(Unpooled.copiedBuffer("数据入库失败".getBytes()));
        } catch (Exception e) {
            logger.error("当前采集报文处理失败，连接继续接收后续数据", e);
        } finally {
            ReferenceCountUtil.release(msg);
        }
	}
	private byte[] hexStringToByteArray(String hexString) {
		// 校验输入合法性
		if (hexString == null || hexString.length() % 2 != 0) {
			throw new IllegalArgumentException("十六进制字符串长度必须为偶数，输入：" + hexString);
		}

		int len = hexString.length();
		byte[] data = new byte[len / 2];
		// 每2个十六进制字符转换为1个字节
		for (int i = 0; i < len; i += 2) {
			data[i / 2] = (byte) ((Character.digit(hexString.charAt(i), 16) << 4)
					+ Character.digit(hexString.charAt(i + 1), 16));
		}
		return data;
	}

	/**
	 * 构建回复报文
	 * @param telemetryAddr 遥测站地址（来自上传报文）
	 * @param centerAddr 中心站地址（来自上传报文）
	 * @param password 密码（来自上传报文）
	 * @param funcCode 功能码（来自上传报文）
	 * @param serialNo 流水号（来自上传报文）
	 * @param uploadEndFlag 上传报文结束符（03/17）
	 * @return 完整的回复报文十六进制字符串（包含CRC）
	 */
	private String buildResponseMessage(String telemetryAddr, String centerAddr, String password,
										String funcCode, String serialNo, String uploadEndFlag) {
		// 1. 固定字段
		String frameHeader = "7E7E"; // 帧头
		String downLen = "8008";    // 下行/载荷长度
		String msgStartFlag = "02"; // 报文起始符

		// 2. 动态字段 - 发报时间（当前时间，格式：年月日时分秒，十六进制）
		String sendTime = DataCheckUtil.getCurrentTimeHex();

		// 3. 回复报文结束符（上传是03则回复04，上传是17则回复06）
		String responseEndFlag = "04";
		if ("17".equals(uploadEndFlag)) {
			responseEndFlag = "06";
		}

		// 4. 拼接回复报文主体（不含CRC）
		StringBuilder bodyBuilder = new StringBuilder();
		bodyBuilder.append(frameHeader)          // 帧头 7E7E
				.append(telemetryAddr)        // 遥测站地址
				.append(centerAddr)           // 中心站地址
				.append(password)             // 密码
				.append(funcCode)             // 功能码
				.append(downLen)              // 下行/载荷长度
				.append(msgStartFlag)         // 报文起始符
				.append(serialNo)             // 流水号
				.append(sendTime)             // 发报时间
				.append(responseEndFlag);     // 报文结束符
		String body = bodyBuilder.toString();
		System.out.println("回复报文主体：" + body);
		// 5. 计算CRC16校验码
		String crc = DataCheckUtil.crc16Tall(body);
		// 6. 拼接最终报文（主体 + CRC）
		return body + crc;
	}
	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());
		}
	}
	/**
	 * 压力值阈值评估：不再使用设备上报的 D14~D10 五种压力报警标志，
	 * 统一按 pressure_value 与 WARN-FIRE-PRESSURE 配置阈值判定。
	 */
	public void checkIfSmsAlertNeeded(FirefightingPressure data) {
		if (data == null) {
			log.warn("消防压力数据为空，跳过阈值预警评估");
			return;
		}
		String deviceCode = data.getTelemeteringStation() == null ? null : data.getTelemeteringStation().trim();
		if (deviceCode == null || deviceCode.isEmpty()) {
			log.warn("消防压力数据缺少遥测站地址，跳过阈值预警评估");
			return;
		}
		// 评估失败（阈值未配置、压力值缺失）时返回 null，保留设备上一报警状态
		Boolean status = evaluate(deviceCode, PRESSURE_WARNING_CODE + ":压力阈值评估",
				() -> evaluatePressure(deviceCode, data));
		if (status != null) {
			updateDeviceAlarmStatus(deviceCode, status ? 1 : 0);
		}
	}

	/**
	 * 按阈值判定本次压力值：越限写报警并关闭预警；未越限则处理告警恢复并生成接近阈值预警。
	 *
	 * @return true 报警，false 正常/预警区
	 */
	private boolean evaluatePressure(String deviceCode, FirefightingPressure data) {
		if (warningThresholdService == null || alarmDataService == null || earlyWarningAutoService == null) {
			throw new IllegalStateException("阈值预警依赖未注入");
		}
		if (data.getPressureValue() == null) {
			throw new IllegalArgumentException("采集报文缺少压力值，无法判定报警状态");
		}
		BigDecimal actual = BigDecimal.valueOf(data.getPressureValue());
		WarningThreshold threshold = warningThresholdService
				.getWarningThresholdByDeviceAndCode(deviceCode, PRESSURE_WARNING_CODE);
		if (threshold == null || (threshold.getMinValue() == null && threshold.getMaxValue() == null)) {
			throw new IllegalStateException("消防压力阈值未配置，无法判定报警状态: deviceCode=" + deviceCode);
		}

		BigDecimal min = threshold.getMinValue() == null ? null : BigDecimal.valueOf(threshold.getMinValue());
		BigDecimal max = threshold.getMaxValue() == null ? null : BigDecimal.valueOf(threshold.getMaxValue());
		// 单侧阈值时补齐等级/预警区计算基线：压力不会为负，仅上限以 0 作下限；仅下限以两倍下限作上限，
		// 保证只配一侧时等级计算仍有可比区间（与可燃气体仅按上限判定的处理保持一致）。
		BigDecimal levelMin = min != null ? min : BigDecimal.ZERO;
		BigDecimal levelMax = max != null ? max : min.multiply(BigDecimal.valueOf(2));

		if (EarlyWarningLevelUtil.isOverThreshold(actual, min, max)) {
			int level = saveAlarmData(deviceCode, PRESSURE_WARNING_TYPE, PRESSURE_WARNING_CODE,
					levelMin, levelMax, actual, "消防压力值越限报警");
			run(deviceCode, PRESSURE_WARNING_CODE + ":关闭预警",
					() -> earlyWarningAutoService.closeAutoWarning(deviceCode, PRESSURE_WARNING_CODE, "已达报警阈值，转报警处理"));
			if (level == 1) {
				run(deviceCode, PRESSURE_WARNING_CODE + ":一级告警短信",
						() -> sendMessage(deviceCode, PRESSURE_WARNING_TYPE, data.getLongitude(), data.getLatitude()));
			}
			return true;
		}

		run(deviceCode, PRESSURE_WARNING_CODE + ":告警恢复",
				() -> alarmDataService.recoverThresholdAlarm(deviceCode, PRESSURE_WARNING_CODE, levelMin, levelMax, actual));
		run(deviceCode, PRESSURE_WARNING_CODE + ":生成预警",
				() -> earlyWarningAutoService.checkAndGenerate(deviceCode, PRESSURE_WARNING_TYPE, PRESSURE_WARNING_CODE,
						levelMin, levelMax, actual, null));
		return false;
	}

	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.setAlarmContent(buildAlarmContent(warningType, actualValue, minValue, maxValue, remark));

			alarmData.setAlarmStatus(0);
			alarmData.setCreateTime(LocalDateTime.now());
			alarmData.setUpdateTime(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;
	}

	/** 告警内容与其他采集服务保持一致：指标名称 + 越限方向，便于前端直接展示。 */
	private String buildAlarmContent(String warningType, BigDecimal actualValue,
									 BigDecimal minValue, BigDecimal maxValue, String remark) {
		String content = warningType == null || warningType.trim().isEmpty() ? remark : warningType.trim();
		if (actualValue == null) {
			return content;
		}
		if (minValue != null && actualValue.compareTo(minValue) <= 0) {
			return content + "达到或低于下限";
		}
		if (maxValue != null && actualValue.compareTo(maxValue) >= 0) {
			return content + "达到或超过上限";
		}
		return content;
	}

	//构造发送短信参数&&发送短信
	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", (lng != null && lat != null)
				? 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);
			}
		}
	}