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

package com.zksy.audio.utils;

import static com.zksy.common.utils.CollectionSideEffects.*;
import com.zksy.audio.domain.ChannelInfo;
import com.zksy.audio.domain.NoiseInfo;
import com.zksy.audio.service.NoiseInfoService;
import com.zksy.audio.service.NoiseLeakAlarmService;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;

/** 只处理已组好的文本行或完整WAV，一份音频只计算和保存一次。 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class MessageHandler extends ChannelInboundHandlerAdapter {
    public final ConcurrentMap<ChannelId, ChannelInfo> channelGroup = new ConcurrentHashMap<>();
    public final ConcurrentMap<String, ChannelInfo> deviceGroup = new ConcurrentHashMap<>();
    @Value("${wlds.fileUploadBaseDir}") private String fileUploadBaseDir;
    // 正常/异常录音每24小时上报；连续重传按滑动静默窗口去重，下一份录音独立处理。
    @Value("${wlds.retransmission-window-seconds:120}") private long retransmissionWindowSeconds = 120;
    private final AudioUploadDeduplicator uploads = new AudioUploadDeduplicator();
    private final NoiseInfoService service;
    @Autowired private DeviceOfflineCheckTask deviceOfflineCheckTask;
    @Autowired private NoiseLeakAlarmService noiseLeakAlarmService;
    @Autowired public MessageHandler(NoiseInfoService service) { this.service = service; }

    @javax.annotation.PostConstruct
    public void validateAudioTiming() {
        if (retransmissionWindowSeconds <= 0 || retransmissionWindowSeconds > 86400) {
            throw new IllegalArgumentException("音频重传静默窗口必须为1~86400秒");
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            ByteBuf input = (ByteBuf) msg;
            if (!input.isReadable()) return;
            ChannelInfo info = channelGroup.computeIfAbsent(ctx.channel().id(), k -> new ChannelInfo("puqi", ctx.channel()));
            if (input.readableBytes() < 4 || input.getInt(input.readerIndex()) != 0x52494646) {
                String line = input.toString(StandardCharsets.US_ASCII).trim();
                if (line.startsWith("$$$")) {
                    String device = line.substring(3).trim();
                    if (!device.matches("[A-Za-z0-9_-]{1,80}")) throw new IllegalArgumentException("音频设备编号无效");
                    if (info.getDeviceNo() != null) deviceGroup.remove(info.getDeviceNo(), info);
                    info.setDeviceNo(device);
                    deviceGroup.put(device, info);
                } else if (line.startsWith("VBAT:")) {
                    info.setPower(parseNumber(line.substring(5).replace("V", "")));
                } else if (line.startsWith("Gain:")) {
                    info.setGain(parseNumber(line.substring(5)));
                }
                return;
            }
            if (info.getDeviceNo() == null) throw new IllegalStateException("音频上传前必须申报设备身份");
            byte[] bytes = new byte[input.readableBytes()];
            input.readBytes(bytes);
            boolean fresh = uploads.receive(info.getDeviceNo(), bytes, retransmissionWindowSeconds,
                    () -> saveAudio(ctx, info, bytes));
            if (!fresh) {
                // 已落库的同一份音频仅补发确认和更新在线状态，不重复漏失计数。
                acknowledge(ctx, info.getDeviceNo());
            }
        } catch (Exception e) {
            log.error("音频采集处理失败，继续接收后续采样", e);
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    private void saveAudio(ChannelHandlerContext ctx, ChannelInfo info, byte[] bytes) throws Exception {
        Path file = null;
        boolean saved = false;
        try {
            LocalDateTime now = LocalDateTime.now();
            String relative = "/devices/" + info.getDeviceNo() + "/" + now.toLocalDate().toString().replace("-", "")
                    + "/" + UUID.randomUUID() + ".wav";
            Path base = Paths.get(fileUploadBaseDir).toAbsolutePath().normalize();
            file = base.resolve(relative.substring(1)).normalize();
            if (!file.startsWith(base)) throw new IllegalArgumentException("音频路径无效");
            Files.createDirectories(file.getParent());
            // 完整写入后再计算，避免异步追加造成不存在/不完整的文件。
            Files.write(file, bytes, StandardOpenOption.CREATE_NEW);
            NoiseInfo noise = new NoiseInfo();
            noise.setEncode(info.getDeviceNo());
            noise.setPower(info.getPower());
            noise.setGain(info.getGain());
            noise.setRms(DataParser.calculateRMS(file.toString()));
            noise.setCenterFrequency(DataParser.calcCenterFrequency(file.toString()));
            noise.setFilepath(relative);
            noise.setTs(now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            requireSaved(service.save(noise));
            saved = true;
            info.setPath(relative);
            info.setFilePath(relative);
            acknowledge(ctx, noise.getEncode());
            run(noise.getEncode(), "漏失报警预警", () -> noiseLeakAlarmService.evaluate(noise));
        } finally {
            if (!saved && file != null) {
                try { Files.deleteIfExists(file); } catch (Exception e) { log.warn("清理未入库音频失败", e); }
            }
        }
    }

    private void acknowledge(ChannelHandlerContext ctx, String device) {
        run(device, "采集应答", () -> ctx.writeAndFlush(Unpooled.copiedBuffer("OK\r\n", StandardCharsets.US_ASCII)));
        run(device, "接收时间更新", () -> DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(device, new Date()));
        run(device, "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(device));
    }

    private double parseNumber(String value) {
        double result = Double.parseDouble(value.trim());
        if (!Double.isFinite(result)) throw new IllegalArgumentException("音频参数不是有效数值");
        return result;
    }

    @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ChannelInfo info = channelGroup.remove(ctx.channel().id());
        if (info != null && info.getDeviceNo() != null) deviceGroup.remove(info.getDeviceNo(), info);
        super.channelInactive(ctx);
    }
    @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("音频连接异常", cause);
        ctx.close();
    }
}