Merge remote-tracking branch 'origin/master'

audio-service/src/main/java/com/zksy/audio/config/NettyServerThread.java

@@ -27,8 +27,9 @@ public class NettyServerThread {
 	private int port;
     @Value("${collection.worker-threads:16}")
     private int businessThreads = 16;
-    @Value("${wlds.raw-pcm-bytes:0}")
-    private int rawPcmBytes;
+    // 厂家handler.txt支持80KiB无头PCM，收齐后补WAV头；带头WAV仍按RIFF长度接收。
+    @Value("${wlds.raw-pcm-bytes:81920}")
+    private int rawPcmBytes = 80 * 1024;
     @Value("${device.offline.timeout-minutes:1560}")
     private int offlineTimeoutMinutes = 1560;
 	private static Logger logger = LoggerFactory.getLogger(NettyServerThread.class);

audio-service/src/main/java/com/zksy/audio/utils/AudioFrameDecoder.java

@@ -11,6 +11,8 @@ import java.util.List;
 public final class AudioFrameDecoder extends ByteToMessageDecoder {
     private static final int MAX_WAV_BYTES = 8 * 1024 * 1024;
     private final int rawPcmBytes;
+    // 损坏WAV头之后寻找下一条元数据或有效WAV，不能将坏头残留误当80K PCM等待。
+    private boolean resynchronizing;
     public AudioFrameDecoder() { this(0); }
     public AudioFrameDecoder(int rawPcmBytes) {
         if (rawPcmBytes < 0 || rawPcmBytes > MAX_WAV_BYTES - 44 || rawPcmBytes % 2 != 0) {
@@ -33,16 +35,19 @@ public final class AudioFrameDecoder extends ByteToMessageDecoder {
                     return;
                 }
                 out.add(in.readRetainedSlice(end - start));
+                resynchronizing = false;
             } else if (prefix.equals("RIFF")) {
                 if (in.readableBytes() < 12) return;
                 long length = in.getUnsignedIntLE(start + 4) + 8;
                 if (length < 44 || length > MAX_WAV_BYTES || in.getInt(start + 8) != 0x57415645) {
                     in.skipBytes(4);
+                    resynchronizing = true;
                     continue;
                 }
                 if (in.readableBytes() < length) return;
                 out.add(in.readRetainedSlice((int) length));
-            } else if (rawPcmBytes > 0) {
+                resynchronizing = false;
+            } else if (rawPcmBytes > 0 && !resynchronizing) {
                 if (in.readableBytes() < rawPcmBytes) return;
                 ByteBuf wav = Unpooled.buffer(44 + rawPcmBytes);
                 wav.writeInt(0x52494646).writeIntLE(36 + rawPcmBytes).writeInt(0x57415645);

audio-service/src/main/resources/application-dev.yaml

@@ -20,6 +20,7 @@ device:
     timeout-minutes: 1560 # 每24小时上报，超过26小时未收到有效数据才判离线
     check-interval-ms: 300000
 wlds:
+  raw-pcm-bytes: 81920 # 厂家80K无头PCM接收长度，另兼容带头WAV；0仅接收带头WAV
   retransmission-window-seconds: 120 # 每24小时上传正常/异常录音；同内容连续重传刷新2分钟静默窗口
   fileUploadBaseDir: D:\home\wlds\uploads
   leak:

audio-service/src/main/resources/application-druid.yaml

@@ -20,6 +20,7 @@ device:
     timeout-minutes: 1560 # 每24小时上报，超过26小时未收到有效数据才判离线
     check-interval-ms: 300000
 wlds:
+  raw-pcm-bytes: 81920 # 厂家80K无头PCM接收长度，另兼容带头WAV；0仅接收带头WAV
   retransmission-window-seconds: 120 # 每24小时上传正常/异常录音；同内容连续重传刷新2分钟静默窗口
   fileUploadBaseDir: \home\wlds\uploads
   leak:

audio-service/src/main/resources/application-prod.yaml

@@ -14,6 +14,7 @@ device:
     timeout-minutes: 1560 # 每24小时上报，超过26小时未收到有效数据才判离线
     check-interval-ms: 300000
 wlds:
+  raw-pcm-bytes: 81920 # 厂家80K无头PCM接收长度，另兼容带头WAV；0仅接收带头WAV
   retransmission-window-seconds: 120 # 每24小时上传正常/异常录音；同内容连续重传刷新2分钟静默窗口
   fileUploadBaseDir: \home\wlds\uploads
   leak:

pipe-network-service/zksy-system/src/main/java/com/zksy/base/alarm/mapper/AlarmDataMapper.java

@@ -1,10 +1,23 @@
 package com.zksy.base.alarm.mapper;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.zksy.base.alarm.domain.AlarmData;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
 
 @Mapper
 public interface AlarmDataMapper extends BaseMapper<AlarmData> {
 
+    IPage<AlarmData> selectLatestPage(Page<AlarmData> page,
+                                      @Param("deviceCodes") List<String> deviceCodes,
+                                      @Param("deviceCode") String deviceCode,
+                                      @Param("warningType") String warningType,
+                                      @Param("warningCode") String warningCode,
+                                      @Param("alarmStatus") Integer alarmStatus,
+                                      @Param("startTime") java.time.LocalDateTime startTime,
+                                      @Param("endTime") java.time.LocalDateTime endTime);
 }

pipe-network-service/zksy-system/src/main/java/com/zksy/base/alarm/service/impl/AlarmDataServiceImpl.java

@@ -7,13 +7,18 @@ import com.zksy.base.alarm.domain.AlarmData;
 import com.zksy.base.alarm.mapper.AlarmDataMapper;
 import com.zksy.base.alarm.service.AlarmDataService;
 import com.zksy.base.domain.EquipmentBase;
+import com.zksy.base.domain.EquipmentType;
 import com.zksy.base.mapper.EquipmentBaseMapper;
+import com.zksy.base.mapper.EquipmentTypeMapper;
 import com.zksy.ops.mapper.WorkOrderMapper;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 @Service
@@ -23,6 +28,9 @@ public class AlarmDataServiceImpl extends ServiceImpl<AlarmDataMapper, AlarmData
     @Autowired
     private EquipmentBaseMapper equipmentBaseMapper;
 
+    @Autowired
+    private EquipmentTypeMapper equipmentTypeMapper;
+
     @Autowired
     private WorkOrderMapper workOrderMapper;
 
@@ -32,38 +40,18 @@ public class AlarmDataServiceImpl extends ServiceImpl<AlarmDataMapper, AlarmData
                                         Integer alarmStatus, String equipmentType,
                                         String startTime, String endTime) {
         Page<AlarmData> page = new Page<>(pageNum, pageSize);
-        LambdaQueryWrapper<AlarmData> queryWrapper = new LambdaQueryWrapper<>();
 
-        // 当传入 equipmentType='drainage' 时，通过 equipment_base 关联查询排水类设备编码列表
-        // alarm_data 表的 warning_type 字段含义不同于 early_warning 表，不能直接用 '2' 过滤
-        if ("drainage".equals(equipmentType)) {
-            LambdaQueryWrapper<EquipmentBase> eqQuery = new LambdaQueryWrapper<>();
-            eqQuery.in(EquipmentBase::getEquipmentTypeId, "6", "7", "8", "9")
-                   .select(EquipmentBase::getEquipmentCode);
-            List<EquipmentBase> devices = equipmentBaseMapper.selectList(eqQuery);
-            List<String> deviceCodes = devices.stream()
-                    .map(EquipmentBase::getEquipmentCode)
-                    .filter(code -> code != null && !code.isEmpty())
-                    .collect(Collectors.toList());
-            if (deviceCodes.isEmpty()) {
-                // 没有排水类设备，直接返回空结果
+        // 当传入 equipmentType='drainage' 时，通过设备类型树关联查询排水类设备编码列表。
+        // 不能再写死历史类型 ID：当前排水根类型为 2，子类型由 equipment_type.parent_id 维护。
+        List<String> deviceCodes = deviceCodesForQuery(equipmentType);
+        if ("drainage".equals(equipmentType) && (deviceCodes == null || deviceCodes.isEmpty())) {
                 return new Page<>(pageNum, pageSize);
-            }
-            queryWrapper.in(AlarmData::getDeviceCode, deviceCodes);
         }
 
-        queryWrapper.like(deviceCode != null, AlarmData::getDeviceCode, deviceCode)
-                .like(warningType != null, AlarmData::getWarningType, warningType)
-                .like(warningCode != null, AlarmData::getWarningCode, warningCode)
-                .eq(alarmStatus != null, AlarmData::getAlarmStatus, alarmStatus);
-        if (startTime != null && !startTime.isEmpty()) {
-            queryWrapper.ge(AlarmData::getAlarmTime, parseDateTime(startTime));
-        }
-        if (endTime != null && !endTime.isEmpty()) {
-            queryWrapper.le(AlarmData::getAlarmTime, parseDateTime(endTime));
-        }
-        queryWrapper.orderByDesc(AlarmData::getAlarmTime);
-        Page<AlarmData> result = this.page(page, queryWrapper);
+        // 数据库先按设备编码保留最新一条，再执行分页，避免内存去重造成分页漏数据。
+        Page<AlarmData> result = (Page<AlarmData>) baseMapper.selectLatestPage(
+                page, deviceCodes, deviceCode, warningType,
+                warningCode, alarmStatus, parseStartTime(startTime), parseEndTime(endTime));
 
         // 填充 deviceName/longitude/latitude：通过 device_code 从 equipment_base 表关联查询
         fillDeviceInfo(result.getRecords());
@@ -73,6 +61,66 @@ public class AlarmDataServiceImpl extends ServiceImpl<AlarmDataMapper, AlarmData
         return result;
     }
 
+    private List<String> deviceCodesForQuery(String equipmentType) {
+        if (!"drainage".equals(equipmentType)) {
+            return null;
+        }
+        List<String> drainageTypeIds = findDrainageTypeIds();
+        if (drainageTypeIds.isEmpty()) {
+            return List.of();
+        }
+        LambdaQueryWrapper<EquipmentBase> eqQuery = new LambdaQueryWrapper<>();
+        eqQuery.in(EquipmentBase::getEquipmentTypeId, drainageTypeIds)
+                .select(EquipmentBase::getEquipmentCode);
+        return equipmentBaseMapper.selectList(eqQuery).stream()
+                .map(EquipmentBase::getEquipmentCode)
+                .filter(code -> code != null && !code.isEmpty())
+                .distinct()
+                .collect(Collectors.toList());
+    }
+
+    /**
+     * 读取排水设备根类型及其全部后代类型，避免设备类型调整后报警列表被错误过滤为空。
+     */
+    private List<String> findDrainageTypeIds() {
+        List<EquipmentType> allTypes = equipmentTypeMapper.selectList(null);
+        if (allTypes == null || allTypes.isEmpty()) {
+            return new ArrayList<>();
+        }
+
+        Set<Integer> drainageIds = new LinkedHashSet<>();
+        for (EquipmentType type : allTypes) {
+            if (type == null || type.getId() == null) {
+                continue;
+            }
+            boolean namedDrainageRoot = "排水设备".equals(type.getTypeName())
+                    && (type.getParentId() == null || type.getParentId() == 0);
+            boolean currentDrainageRoot = type.getId() == 2
+                    && (type.getParentId() == null || type.getParentId() == 0);
+            if (namedDrainageRoot || currentDrainageRoot) {
+                drainageIds.add(type.getId());
+            }
+        }
+
+        // 兼容数据中根类型名称被修改但仍使用稳定业务 ID=2 的情况。
+        if (drainageIds.isEmpty()) {
+            drainageIds.add(2);
+        }
+
+        boolean added;
+        do {
+            added = false;
+            for (EquipmentType type : allTypes) {
+                if (type != null && type.getId() != null && type.getParentId() != null
+                        && drainageIds.contains(type.getParentId())) {
+                    added |= drainageIds.add(type.getId());
+                }
+            }
+        } while (added);
+
+        return drainageIds.stream().map(String::valueOf).collect(Collectors.toList());
+    }
+
     private void fillWorkOrderStatus(List<AlarmData> records) {
         if (records == null || records.isEmpty()) {
             return;
@@ -106,6 +154,25 @@ public class AlarmDataServiceImpl extends ServiceImpl<AlarmDataMapper, AlarmData
         return LocalDateTime.parse(value, java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
     }
 
+    private LocalDateTime parseStartTime(String value) {
+        return parseDateTime(value, " 00:00:00");
+    }
+
+    private LocalDateTime parseEndTime(String value) {
+        return parseDateTime(value, " 23:59:59");
+    }
+
+    private LocalDateTime parseDateTime(String value, String dateSuffix) {
+        if (value == null || value.trim().isEmpty()) {
+            return null;
+        }
+        String normalized = value.trim();
+        if (normalized.length() == 10) {
+            normalized += dateSuffix;
+        }
+        return parseDateTime(normalized);
+    }
+
     /**
      * 根据 deviceCode 从 equipment_base 表批量填充 deviceName、longitude、latitude
      */

pipe-network-service/zksy-system/src/main/resources/mapper/base/AlarmDataMapper.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.zksy.base.alarm.mapper.AlarmDataMapper">
+
+    <select id="selectLatestPage" resultType="com.zksy.base.alarm.domain.AlarmData">
+        SELECT a.*
+        FROM alarm_data a
+        <where>
+            <if test="deviceCodes != null and deviceCodes.size() > 0">
+                AND a.device_code IN
+                <foreach collection="deviceCodes" item="code" open="(" separator="," close=")">
+                    #{code}
+                </foreach>
+            </if>
+            <if test="deviceCode != null and deviceCode != ''">
+                AND a.device_code LIKE CONCAT('%', #{deviceCode}, '%')
+            </if>
+            <if test="warningType != null and warningType != ''">
+                AND a.warning_type LIKE CONCAT('%', #{warningType}, '%')
+            </if>
+            <if test="warningCode != null and warningCode != ''">
+                AND a.warning_code LIKE CONCAT('%', #{warningCode}, '%')
+            </if>
+            <if test="alarmStatus != null">
+                AND a.alarm_status = #{alarmStatus}
+            </if>
+            <if test="startTime != null">
+                AND a.alarm_time &gt;= #{startTime}
+            </if>
+            <if test="endTime != null">
+                AND a.alarm_time &lt;= #{endTime}
+            </if>
+            AND NOT EXISTS (
+                SELECT 1
+                FROM alarm_data newer
+                WHERE newer.device_code = a.device_code
+                  AND (
+                      COALESCE(newer.alarm_time, newer.create_time) &gt; COALESCE(a.alarm_time, a.create_time)
+                      OR (
+                          COALESCE(newer.alarm_time, newer.create_time) = COALESCE(a.alarm_time, a.create_time)
+                          AND newer.id &gt; a.id
+                      )
+                  )
+            )
+        </where>
+        ORDER BY a.alarm_time DESC, a.create_time DESC, a.id DESC
+    </select>
+
+</mapper>