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 channelGroup = new ConcurrentHashMap<>(); public final ConcurrentMap 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(); } }