fix(water): scope supply monitoring to water device tree (#8)

pipe-network-service/sql/water_supply_monitoring_test_data.sql

@@ -6,7 +6,7 @@
 --      app_user.equipment_type、app_user.equipment_base、app_user.radar_data、
 --      app_user.telemetry_data、app_user.firefighting_pressure、app_user.noise_info
 --   3. 覆盖场景：雷达流量、遥测流量、压力、漏失、无数据设备、
---      数值范围过滤、时间范围过滤与近24小时之外的旧趋势点。
+--      非供水同名设备排除、数值范围过滤、时间范围过滤与近24小时之外的旧趋势点。
 -- ============================================================
 
 -- ------------------------------------------------------------
@@ -20,14 +20,34 @@ DELETE FROM app_user.equipment_base WHERE equipment_id LIKE 'WS-TEST-EQ-%';
 DELETE FROM app_user.equipment_type WHERE id LIKE 'WS-TEST-TYPE-%';
 
 -- ------------------------------------------------------------
--- 1. 设备类型与监测设备（equipment_type_id 显式关联类型主键）
+-- 1. 设备类型与监测设备（供水测试类型挂在顶级类型“供水”之下）
 -- ------------------------------------------------------------
-INSERT INTO app_user.equipment_type
-    (id, type_id, type_name, parent_type_id, type_remark)
-VALUES
-    ('WS-TEST-TYPE-FLOW', 'ws_test_flow_monitoring', '流量监测(测试)', '0', '供水T4流量监测测试类型'),
-    ('WS-TEST-TYPE-PRESSURE', 'ws_test_pressure_monitoring', '压力监测(测试)', '0', '供水T4压力监测测试类型'),
-    ('WS-TEST-TYPE-LEAKAGE', 'ws_test_leakage_monitoring', '漏失监测(测试)', '0', '供水T4漏失监测测试类型');
+DO $$
+DECLARE
+    v_water_top_id varchar(64);
+BEGIN
+    SELECT id INTO v_water_top_id
+      FROM app_user.equipment_type
+     WHERE type_name = '供水' AND parent_type_id = '0'
+     LIMIT 1;
+
+    IF v_water_top_id IS NULL THEN
+        INSERT INTO app_user.equipment_type
+            (id, type_id, type_name, parent_type_id, type_remark)
+        VALUES
+            ('WS-TEST-TYPE-WATER', 'ws_test_water_top', '供水', '0', '供水T4运行监测测试顶级类型');
+        v_water_top_id := 'WS-TEST-TYPE-WATER';
+    END IF;
+
+    INSERT INTO app_user.equipment_type
+        (id, type_id, type_name, parent_type_id, type_remark)
+    VALUES
+        ('WS-TEST-TYPE-FLOW', 'ws_test_flow_monitoring', '流量监测(测试)', v_water_top_id, '供水T4流量监测测试类型'),
+        ('WS-TEST-TYPE-PRESSURE', 'ws_test_pressure_monitoring', '压力监测(测试)', v_water_top_id, '供水T4压力监测测试类型'),
+        ('WS-TEST-TYPE-LEAKAGE', 'ws_test_leakage_monitoring', '漏失监测(测试)', v_water_top_id, '供水T4漏失监测测试类型'),
+        ('WS-TEST-TYPE-OTHER', 'ws_test_other_top', '其他系统(测试)', '0', '供水T4边界测试顶级类型'),
+        ('WS-TEST-TYPE-OTHER-FLOW', 'ws_test_other_flow', '流量监测(测试)', 'WS-TEST-TYPE-OTHER', '供水T4非供水边界测试类型');
+END $$;
 
 INSERT INTO app_user.equipment_base
     (equipment_id, equipment_code, equipment_name, equipment_model, equipment_type_id,
@@ -53,6 +73,10 @@ VALUES
     ('WS-TEST-EQ-FLOW-EMPTY', 'WS-TEST-FLOW-EMPTY', '测试无测量流量监测点', 'FLOW-EMPTY-TEST',
      'WS-TEST-TYPE-FLOW', 113.32000000, 23.16000000, '供水测试大道5号',
      '供水测试维护员', '13800000005', '供水测试公司', '供水T4运行监测无数据测试设备',
+     now() - interval '7 days', now(), '0', '供水测试区'),
+    ('WS-TEST-EQ-OTHER-FLOW', 'WS-TEST-OTHER-FLOW', '测试非供水流量监测点', 'OTHER-FLOW-TEST',
+     'WS-TEST-TYPE-OTHER-FLOW', 113.33000000, 23.17000000, '其他系统测试大道6号',
+     '供水测试维护员', '13800000006', '供水测试公司', '供水T4非供水设备边界测试数据',
      now() - interval '7 days', now(), '0', '供水测试区');
 
 -- ------------------------------------------------------------
@@ -68,7 +92,9 @@ VALUES
     ('WS-TEST-RADAR-FLOW-002', 'WS-TEST-SYSTEM', 'WS-TEST-FLOW-RADAR', '120', '3.50',
      now() - interval '1 hour'),
     ('WS-TEST-RADAR-FLOW-003', 'WS-TEST-SYSTEM', 'WS-TEST-FLOW-RADAR', '18', '0.80',
-     now() - interval '5 days');
+     now() - interval '5 days'),
+    ('WS-TEST-RADAR-OTHER-FLOW-001', 'WS-TEST-SYSTEM', 'WS-TEST-OTHER-FLOW', '66', '2.00',
+     now() - interval '1 hour');
 
 -- ------------------------------------------------------------
 -- 3. 遥测终端流量数据（device_code 与 equipment_base.equipment_code 关联）
@@ -121,4 +147,5 @@ VALUES
 --   压力分页 pressureMin=0.30&pressureMax=0.60：0.42，共1条；
 --   漏失分页 powerMin=8&powerMax=15：12，共1条；
 --   WS-TEST-FLOW-EMPTY 无测量数据，不应出现在任何监测页。
+--   WS-TEST-OTHER-FLOW 有同名流量类型和测量数据，但不在供水类型树内，不应出现在任何监测页。
 -- ============================================================

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

@@ -20,6 +20,7 @@ import com.zksy.base.radar.mapper.RadarDataMapper;
 import com.zksy.base.service.MonitoringService;
 import com.zksy.base.telemetry.domain.TelemetryData;
 import com.zksy.base.telemetry.mapper.TelemetryDataMapper;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -34,6 +35,9 @@ public class MonitoringServiceImpl implements MonitoringService {
     private static final String RADAR_SOURCE = "雷达流量计";
     private static final String TELEMETRY_SOURCE = "遥测终端";
 
+    @Value("${waterSupply.monitoring.top-level-type:供水}")
+    private String waterSupplyTopLevelType = "供水";
+
     @Autowired
     private EquipmentBaseMapper equipmentBaseMapper;
 
@@ -181,7 +185,7 @@ public class MonitoringServiceImpl implements MonitoringService {
 
     @Override
     public List<Map<String, Object>> getFlowTrend(String equipmentId, LocalDateTime startTime, LocalDateTime endTime) {
-        EquipmentBase device = equipmentBaseMapper.selectById(equipmentId);
+        EquipmentBase device = selectMonitoringDevice(equipmentId, "flow");
         if (device == null || StrUtil.isBlank(device.getEquipmentCode())) {
             return new ArrayList<>();
         }
@@ -225,7 +229,7 @@ public class MonitoringServiceImpl implements MonitoringService {
 
     @Override
     public List<Map<String, Object>> getPressureTrend(String equipmentId, LocalDateTime startTime, LocalDateTime endTime) {
-        EquipmentBase device = equipmentBaseMapper.selectById(equipmentId);
+        EquipmentBase device = selectMonitoringDevice(equipmentId, "pressure");
         if (device == null || StrUtil.isBlank(device.getEquipmentCode())) {
             return new ArrayList<>();
         }
@@ -252,7 +256,7 @@ public class MonitoringServiceImpl implements MonitoringService {
 
     @Override
     public List<Map<String, Object>> getLeakageTrend(String equipmentId, LocalDateTime startTime, LocalDateTime endTime) {
-        EquipmentBase device = equipmentBaseMapper.selectById(equipmentId);
+        EquipmentBase device = selectMonitoringDevice(equipmentId, "leakage");
         if (device == null || StrUtil.isBlank(device.getEquipmentCode())) {
             return new ArrayList<>();
         }
@@ -287,10 +291,21 @@ public class MonitoringServiceImpl implements MonitoringService {
                 .like(StrUtil.isNotBlank(equipmentCode), EquipmentBase::getEquipmentCode, equipmentCode)
                 .like(StrUtil.isNotBlank(equipmentLocation), EquipmentBase::getEquipmentLocation, equipmentLocation)
                 .orderByDesc(EquipmentBase::getCreateTime);
-        constrainToMonitoringTypes(wrapper, monitorKind);
+        Set<String> monitoringTypeIds = selectMonitoringTypeIds(monitorKind);
+        wrapper.in(EquipmentBase::getEquipmentTypeId,
+                monitoringTypeIds.isEmpty() ? Collections.singleton("__NO_WATER_MONITORING_TYPE__") : monitoringTypeIds);
         return equipmentBaseMapper.selectList(wrapper);
     }
 
+    private EquipmentBase selectMonitoringDevice(String equipmentId, String monitorKind) {
+        EquipmentBase device = equipmentBaseMapper.selectById(equipmentId);
+        if (device == null || StrUtil.isBlank(device.getEquipmentCode())) {
+            return null;
+        }
+        Set<String> monitoringTypeIds = selectMonitoringTypeIds(monitorKind);
+        return monitoringTypeIds.contains(device.getEquipmentTypeId()) ? device : null;
+    }
+
     private Map<String, EquipmentType> selectTypeMap(List<EquipmentBase> devices) {
         boolean anyTypePresent = devices.stream()
                 .map(EquipmentBase::getEquipmentTypeId)
@@ -485,8 +500,8 @@ public class MonitoringServiceImpl implements MonitoringService {
         return endTime != null ? null : LocalDateTime.now().minusHours(24);
     }
 
-    /** Restrict the equipment query before pagination so each monitoring page only contains its sub-devices. */
-    private void constrainToMonitoringTypes(LambdaQueryWrapper<EquipmentBase> wrapper, String monitorKind) {
+    /** Restrict monitoring pages to sub-devices inside the configured water-supply type tree. */
+    private Set<String> selectMonitoringTypeIds(String monitorKind) {
         String[] keywords;
         switch (monitorKind) {
             case "pressure":
@@ -500,17 +515,45 @@ public class MonitoringServiceImpl implements MonitoringService {
                 break;
         }
         List<EquipmentType> types = equipmentTypeMapper.selectList(new LambdaQueryWrapper<>());
-        Set<String> ids = types.stream()
+        Set<String> waterTypeIds = waterSupplyTypeTreeIds(types);
+        if (waterTypeIds.isEmpty()) {
+            return Collections.emptySet();
+        }
+        return types.stream()
+                .filter(type -> waterTypeIds.contains(type.getId()))
                 .filter(type -> {
-                    String value = ((type.getId() == null ? "" : type.getId()) + " "
-                            + (type.getTypeId() == null ? "" : type.getTypeId()) + " "
+                    String value = ((type.getTypeId() == null ? "" : type.getTypeId()) + " "
                             + (type.getTypeName() == null ? "" : type.getTypeName())).toLowerCase(Locale.ROOT);
                     return Arrays.stream(keywords).anyMatch(keyword -> value.contains(keyword.toLowerCase(Locale.ROOT)));
                 })
                 .map(EquipmentType::getId)
                 .filter(StrUtil::isNotBlank)
                 .collect(Collectors.toSet());
-        wrapper.in(EquipmentBase::getEquipmentTypeId,
-                ids.isEmpty() ? Collections.singleton("__NO_MONITORING_TYPE__") : ids);
+    }
+
+    private Set<String> waterSupplyTypeTreeIds(List<EquipmentType> types) {
+        Map<String, List<EquipmentType>> childrenByParent = types.stream()
+                .filter(type -> StrUtil.isNotBlank(type.getParentTypeId()) && StrUtil.isNotBlank(type.getId()))
+                .collect(Collectors.groupingBy(EquipmentType::getParentTypeId));
+
+        Set<String> typeIds = new LinkedHashSet<>();
+        Deque<String> pending = types.stream()
+                .filter(type -> "0".equals(type.getParentTypeId())
+                        && waterSupplyTopLevelType.equals(type.getTypeName())
+                        && StrUtil.isNotBlank(type.getId()))
+                .map(EquipmentType::getId)
+                .collect(Collectors.toCollection(ArrayDeque::new));
+        while (!pending.isEmpty()) {
+            String typeId = pending.poll();
+            if (!typeIds.add(typeId)) {
+                continue;
+            }
+            childrenByParent.getOrDefault(typeId, Collections.emptyList())
+                    .stream()
+                    .map(EquipmentType::getId)
+                    .filter(StrUtil::isNotBlank)
+                    .forEach(pending::offer);
+        }
+        return typeIds;
     }
 }

pipe-network-service/zksy-system/src/test/java/com/zksy/base/service/impl/MonitoringServiceImplTest.java

@@ -1,5 +1,8 @@
 package com.zksy.base.service.impl;
 
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.zksy.base.audio.domain.NoiseInfo;
 import com.zksy.base.audio.mapper.NoiseInfoMapper;
@@ -16,16 +19,20 @@ import com.zksy.base.radar.domain.RadarData;
 import com.zksy.base.radar.mapper.RadarDataMapper;
 import com.zksy.base.telemetry.domain.TelemetryData;
 import com.zksy.base.telemetry.mapper.TelemetryDataMapper;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import java.time.LocalDateTime;
+import java.util.Collection;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -38,6 +45,12 @@ class MonitoringServiceImplTest {
 
     private static final LocalDateTime BASE_TIME = LocalDateTime.of(2026, 9, 1, 12, 0);
 
+    @BeforeAll
+    static void initializeMybatisPlusLambdaCache() {
+        TableInfoHelper.initTableInfo(
+                new MapperBuilderAssistant(new MybatisConfiguration(), ""), EquipmentBase.class);
+    }
+
     private MonitoringServiceImpl service;
     private EquipmentBaseMapper equipmentBaseMapper;
     private EquipmentTypeMapper equipmentTypeMapper;
@@ -64,9 +77,10 @@ class MonitoringServiceImplTest {
         ReflectionTestUtils.setField(service, "noiseInfoMapper", noiseMapper);
 
         when(equipmentTypeMapper.selectList(any())).thenReturn(Arrays.asList(
-                type("flow-type", "flow-type", "流量监测"),
-                type("pressure-type", "pressure-type", "压力监测"),
-                type("leakage-type", "leakage-type", "漏失监测")
+                type("water-root", "water_root", "供水", "0"),
+                type("flow-type", "flow-type", "流量监测", "water-root"),
+                type("pressure-type", "pressure-type", "压力监测", "water-root"),
+                type("leakage-type", "leakage-type", "漏失监测", "water-root")
         ));
         when(radarDataMapper.selectList(any())).thenReturn(Collections.emptyList());
         when(telemetryDataMapper.selectList(any())).thenReturn(Collections.emptyList());
@@ -167,6 +181,57 @@ class MonitoringServiceImplTest {
         assertTrue(leakage.getRecords().isEmpty());
     }
 
+    @Test
+    void flowMonitoringOnlyIncludesDevicesInsideTheWaterTypeTree() {
+        when(equipmentTypeMapper.selectList(any())).thenReturn(Arrays.asList(
+                type("water-root", "water_root", "供水", "0"),
+                type("water-flow", "water_flow", "流量监测", "water-root"),
+                type("other-root", "other_root", "其他系统", "0"),
+                type("other-flow", "other_flow", "流量监测", "other-root")
+        ));
+        when(equipmentBaseMapper.selectList(any())).thenAnswer(invocation -> {
+            LambdaQueryWrapper<EquipmentBase> wrapper = invocation.getArgument(0);
+            Set<Object> allowedTypeIds = flattenWrapperValues(wrapper);
+            return Arrays.asList(
+                    device("water-flow-device", "WS-FLOW", "供水流量监测点", "water-flow", BASE_TIME),
+                    device("other-flow-device", "OTHER-FLOW", "其他系统流量监测点", "other-flow", BASE_TIME),
+                    device("unknown-type-device", "UNKNOWN-FLOW", "未归属流量监测点", "unknown-type", BASE_TIME)
+            ).stream()
+                    .filter(device -> allowedTypeIds.contains(device.getEquipmentTypeId()))
+                    .collect(Collectors.toList());
+        });
+        when(radarDataMapper.selectList(any())).thenReturn(Arrays.asList(
+                radar("WS-FLOW", "10", BASE_TIME),
+                radar("OTHER-FLOW", "20", BASE_TIME),
+                radar("UNKNOWN-FLOW", "30", BASE_TIME)
+        ));
+
+        Page<FlowMonitoringVO> page = service.getFlowMonitorPage(
+                1, 10, null, null, null, null, null, null, null);
+
+        assertEquals(1, page.getTotal());
+        assertEquals("water-flow-device", page.getRecords().get(0).getEquipmentId());
+    }
+
+    @Test
+    void trendEndpointsRejectDevicesOutsideTheWaterTypeTree() {
+        when(equipmentTypeMapper.selectList(any())).thenReturn(Arrays.asList(
+                type("water-root", "water_root", "供水", "0"),
+                type("water-flow", "water_flow", "流量监测", "water-root"),
+                type("other-root", "other_root", "其他系统", "0"),
+                type("other-flow", "other_flow", "流量监测", "other-root")
+        ));
+        EquipmentBase otherDevice = device("other-flow-device", "OTHER-FLOW", "其他系统流量监测点", "other-flow", BASE_TIME);
+        when(equipmentBaseMapper.selectById("other-flow-device")).thenReturn(otherDevice);
+        when(radarDataMapper.selectList(any())).thenReturn(Collections.singletonList(
+                radar("OTHER-FLOW", "20", BASE_TIME)
+        ));
+
+        List<Map<String, Object>> trend = service.getFlowTrend("other-flow-device", null, null);
+
+        assertTrue(trend.isEmpty());
+    }
+
     @Test
     void trendEndpointsUseTheRequestedTimeRangeForAllMonitoringTypes() {
         LocalDateTime start = BASE_TIME.minusHours(2);
@@ -242,10 +307,31 @@ class MonitoringServiceImplTest {
     }
 
     private static EquipmentType type(String id, String typeId, String name) {
+        return type(id, typeId, name, "water-root");
+    }
+
+    private static Set<Object> flattenWrapperValues(
+            LambdaQueryWrapper<EquipmentBase> wrapper) {
+        wrapper.getSqlSegment();
+        Set<Object> values = new HashSet<>();
+        for (Object value : wrapper.getParamNameValuePairs().values()) {
+            if (value instanceof Collection<?>) {
+                values.addAll((Collection<?>) value);
+            } else if (value instanceof Object[]) {
+                values.addAll(Arrays.asList((Object[]) value));
+            } else {
+                values.add(value);
+            }
+        }
+        return values;
+    }
+
+    private static EquipmentType type(String id, String typeId, String name, String parentTypeId) {
         EquipmentType type = new EquipmentType();
         type.setId(id);
         type.setTypeId(typeId);
         type.setTypeName(name);
+        type.setParentTypeId(parentTypeId);
         return type;
     }