feat(drainage): 完善排水预警与易积水点业务，新增管网综合风险评估

- 发布预警补齐预警类型、权属单位、发布人等信息，写入异常时返回可读提示
- 预警解除记录处置人与处置时间
- 易积水点补充责任部门、责任人、风险等级、积水深度与积水面积等台账信息，
  责任单位与责任人改为关联系统部门与用户，同时保留名称快照
- 新增排水管网综合风险评估：按管网本体、运行报警、环境叠加、数据质量四个维度评分，
  支持列表筛选、汇总统计、地图分布与单条详情，结果统一缓存避免重复取数
- 派单前按设备归属单位校验操作范围，处置人与核验人支持按名称检索
- 设备查询增加安装位置、所属片区、设备状态三个筛选条件
- 整理管网旧查询入口中已无调用方的维护类能力

附：新增数据初始化与结构补全脚本

pipe-network-service/sql/apply_drainage_jcsj_fix.sql

@@ -0,0 +1,366 @@
+-- ============================================================================
+-- 排水基础数据：一次性补齐脚本（KingbaseES, schema app_user）
+--
+-- 适用场景
+--   排水子系统页面报「字段 xxx 不存在」，例如：
+--     ERROR: 字段 "duty_dept_id" 不存在
+--     ERROR: 关系 "early_warning" 的 "longitude" 字段不存在
+--   说明库里的表还停留在旧版本。本脚本把本轮所有相关改动合并成一次执行，
+--   不改表结构之外的东西，不删任何数据。
+--
+-- 包含四段（均可单独复制执行）
+--   1. drainage_waterlogging_point：补齐实体用到的全部列 + 索引
+--   2. 责任单位/责任人：duty_dept_id / duty_user_id 的存量回填（只做精确匹配）
+--   3. 字典 drainage_district：0城南 1城北 2太常
+--   4. early_warning：device_name 补齐、warning_level 转 integer、
+--      接收人表 psgw_warning_scope_user 建表、旧版遗留列放开非空
+--
+-- 幂等：可重复执行，已到位的部分只会输出 NOTICE。
+-- ============================================================================
+
+-- 结束上一次失败执行遗留的 aborted 事务（在事务外执行只会告警，不报错）
+ROLLBACK;
+
+BEGIN;
+
+SET search_path TO app_user, public;
+
+-- ============================================================================
+-- 1. 易积水点表：补齐实体所需的全部列
+--    （老库可能只缺 duty_* 两列，也可能缺 water_depth / risk_level 等，
+--      这里按实体字段全集逐列判断，缺哪列补哪列）
+-- ============================================================================
+DO $$
+DECLARE
+  col record;
+  added int := 0;
+BEGIN
+  IF to_regclass('app_user.drainage_waterlogging_point') IS NULL THEN
+    RAISE NOTICE '1. 表 app_user.drainage_waterlogging_point 不存在，跳过补列';
+    RETURN;
+  END IF;
+
+  FOR col IN
+    SELECT * FROM (VALUES
+        ('point_code',          'varchar(64)'),
+        ('point_name',          'varchar(200)'),
+        ('district',            'varchar(100)'),
+        ('location',            'varchar(500)'),
+        ('longitude',           'numeric(12,8)'),
+        ('latitude',            'numeric(12,8)'),
+        ('waterlogging_level',  'int2'),
+        ('water_depth',         'numeric(10,2)'),
+        ('water_volume',        'numeric(14,2)'),
+        ('water_area',          'numeric(14,2)'),
+        ('risk_level',          'varchar(8)'),
+        ('duty_unit',           'varchar(200)'),
+        ('duty_person',         'varchar(64)'),
+        ('duty_phone',          'varchar(32)'),
+        ('duty_dept_id',        'bigint'),
+        ('duty_user_id',        'bigint'),
+        ('camera_ids',          'varchar(500)'),
+        ('remark',              'varchar(500)'),
+        ('create_by',           'varchar(64)'),
+        ('create_time',         'timestamp'),
+        ('update_by',           'varchar(64)'),
+        ('update_time',         'timestamp'),
+        ('del_flag',            'char(1)')
+      ) AS x(name, type)
+  LOOP
+    IF NOT EXISTS (
+        SELECT 1 FROM information_schema.columns
+         WHERE table_schema = 'app_user'
+           AND table_name = 'drainage_waterlogging_point'
+           AND column_name = col.name
+    ) THEN
+      EXECUTE format('ALTER TABLE app_user.drainage_waterlogging_point ADD COLUMN %I %s',
+                     col.name, col.type);
+      added := added + 1;
+      RAISE NOTICE '1. 已补列：drainage_waterlogging_point.%', col.name;
+    END IF;
+  END LOOP;
+
+  IF added = 0 THEN
+    RAISE NOTICE '1. drainage_waterlogging_point 列已齐全，无需改动';
+  END IF;
+END $$;
+
+-- del_flag 有历史 NULL 值时逻辑删除过滤会漏行，统一补成 '0'
+DO $$
+BEGIN
+  IF to_regclass('app_user.drainage_waterlogging_point') IS NULL THEN
+    RAISE NOTICE '1. 表不存在，跳过索引与注释';
+    RETURN;
+  END IF;
+
+  UPDATE app_user.drainage_waterlogging_point SET del_flag = '0' WHERE del_flag IS NULL;
+
+  EXECUTE 'CREATE INDEX IF NOT EXISTS idx_dwp_point_code ON app_user.drainage_waterlogging_point (point_code)';
+  EXECUTE 'CREATE INDEX IF NOT EXISTS idx_dwp_point_name ON app_user.drainage_waterlogging_point (point_name)';
+  EXECUTE 'CREATE INDEX IF NOT EXISTS idx_dwp_district   ON app_user.drainage_waterlogging_point (district)';
+  EXECUTE 'CREATE INDEX IF NOT EXISTS idx_dwp_level      ON app_user.drainage_waterlogging_point (waterlogging_level)';
+  EXECUTE 'CREATE INDEX IF NOT EXISTS idx_dwp_risk_level ON app_user.drainage_waterlogging_point (risk_level)';
+  EXECUTE 'CREATE INDEX IF NOT EXISTS idx_dwp_duty_dept  ON app_user.drainage_waterlogging_point (duty_dept_id)';
+  EXECUTE 'CREATE INDEX IF NOT EXISTS idx_dwp_duty_user  ON app_user.drainage_waterlogging_point (duty_user_id)';
+
+  EXECUTE 'COMMENT ON COLUMN app_user.drainage_waterlogging_point.district     IS ''所属片区（字典 drainage_district：0城南 1城北 2太常）''';
+  EXECUTE 'COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_unit    IS ''责任单位（名称快照，对应 duty_dept_id）''';
+  EXECUTE 'COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_person  IS ''责任人（姓名快照，对应 duty_user_id）''';
+  EXECUTE 'COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_phone   IS ''联系电话（快照，选人时取 sys_user.phonenumber）''';
+  EXECUTE 'COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_dept_id IS ''责任单位ID（sys_dept.dept_id）''';
+  EXECUTE 'COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_user_id IS ''责任人ID（sys_user.user_id）''';
+END $$;
+
+-- ============================================================================
+-- 2. 责任单位/责任人回填：按名称精确匹配，匹配不上保持 NULL（不猜）
+--    同一名称对应多条时取 ID 最小的一个，避免一对多歧义放大。
+-- ============================================================================
+DO $$
+DECLARE
+  v_dept integer := 0;
+  v_user integer := 0;
+BEGIN
+  IF to_regclass('app_user.drainage_waterlogging_point') IS NULL THEN
+    RAISE NOTICE '2. 表不存在，跳过回填';
+    RETURN;
+  END IF;
+
+  UPDATE app_user.drainage_waterlogging_point p
+     SET duty_dept_id = d.dept_id
+    FROM (SELECT dept_name, MIN(dept_id) AS dept_id
+            FROM app_user.sys_dept
+           WHERE del_flag = '0'
+           GROUP BY dept_name) d
+   WHERE p.duty_dept_id IS NULL
+     AND p.duty_unit IS NOT NULL
+     AND trim(p.duty_unit) = d.dept_name;
+  GET DIAGNOSTICS v_dept = ROW_COUNT;
+
+  UPDATE app_user.drainage_waterlogging_point p
+     SET duty_user_id = u.user_id
+    FROM (SELECT nick_name, MIN(user_id) AS user_id
+            FROM app_user.sys_user
+           WHERE del_flag = '0'
+           GROUP BY nick_name) u
+   WHERE p.duty_user_id IS NULL
+     AND p.duty_person IS NOT NULL
+     AND trim(p.duty_person) = u.nick_name;
+  GET DIAGNOSTICS v_user = ROW_COUNT;
+
+  RAISE NOTICE '2. 存量回填：责任单位匹配 % 条，责任人匹配 % 条', v_dept, v_user;
+END $$;
+
+-- ============================================================================
+-- 3. 字典：所属片区 drainage_district
+--    本库 sys_dict_data / sys_dict_type 是 MySQL 兼容 auto_increment，
+--    自增计数器未随迁移数据推进，主键一律显式取 max+序号。
+-- ============================================================================
+INSERT INTO app_user.sys_dict_type
+    (dict_id, dict_name, dict_type, status, create_by, create_time, remark)
+SELECT (SELECT COALESCE(MAX(dict_id), 0) FROM app_user.sys_dict_type) + 1,
+       '所属片区', 'drainage_district', '0', 'admin', CURRENT_TIMESTAMP,
+       '排水所属片区：0城南 1城北 2太常（与 DistrictEnum 对齐）'
+ WHERE NOT EXISTS (SELECT 1 FROM app_user.sys_dict_type WHERE dict_type = 'drainage_district');
+
+UPDATE app_user.sys_dict_type
+   SET dict_name = '所属片区',
+       status = '0',
+       update_by = 'admin',
+       update_time = CURRENT_TIMESTAMP,
+       remark = '排水所属片区：0城南 1城北 2太常（与 DistrictEnum 对齐）'
+ WHERE dict_type = 'drainage_district';
+
+DELETE FROM app_user.sys_dict_data WHERE dict_type = 'drainage_district';
+
+INSERT INTO app_user.sys_dict_data
+    (dict_code, dict_sort, dict_label, dict_value, dict_type, css_class, list_class,
+     is_default, status, create_by, create_time, remark)
+SELECT (SELECT COALESCE(MAX(dict_code), 0) FROM app_user.sys_dict_data) + v.rn,
+       v.dict_sort, v.dict_label, v.dict_value, 'drainage_district',
+       '', '', v.is_default, '0', 'admin', CURRENT_TIMESTAMP, v.remark
+  FROM (VALUES
+        (1, 1, '城南', '0', 'N', '所属片区：城南'),
+        (2, 2, '城北', '1', 'N', '所属片区：城北'),
+        (3, 3, '太常', '2', 'N', '所属片区：太常')
+       ) AS v(rn, dict_sort, dict_label, dict_value, is_default, remark)
+ ORDER BY v.rn;
+
+-- 存量片区值归一：中文地名 → 编码（无法识别的一律原样保留，避免误伤脏数据）
+DO $$
+BEGIN
+  IF to_regclass('app_user.drainage_waterlogging_point') IS NULL THEN
+    RAISE NOTICE '3. 表不存在，跳过片区值归一';
+    RETURN;
+  END IF;
+
+  UPDATE app_user.drainage_waterlogging_point
+     SET district = CASE TRIM(district)
+                      WHEN '城南' THEN '0'
+                      WHEN '城北' THEN '1'
+                      WHEN '太常' THEN '2'
+                    END
+   WHERE TRIM(district) IN ('城南', '城北', '太常');
+END $$;
+
+-- 字典自增序列尽力校准（三种方式兜底，失败只提示）
+DO $$
+DECLARE
+  t         record;
+  v_seq     text;
+  v_max     bigint;
+  v_default text;
+BEGIN
+  FOR t IN
+    SELECT * FROM (VALUES ('sys_dict_data', 'dict_code'),
+                          ('sys_dict_type', 'dict_id')) AS x(tbl, col)
+  LOOP
+    v_seq := NULL;
+    v_default := NULL;
+
+    BEGIN
+      v_seq := pg_get_serial_sequence(format('app_user.%I', t.tbl), t.col);
+    EXCEPTION WHEN OTHERS THEN
+      v_seq := NULL;
+    END;
+
+    IF v_seq IS NULL THEN
+      BEGIN
+        SELECT pg_get_expr(ad.adbin, ad.adrelid) INTO v_default
+          FROM pg_attribute a
+          JOIN pg_attrdef ad ON ad.adrelid = a.attrelid AND ad.adnum = a.attnum
+         WHERE a.attrelid = format('app_user.%I', t.tbl)::regclass
+           AND a.attname::text = t.col
+           AND a.attnum > 0
+           AND NOT a.attisdropped;
+      EXCEPTION WHEN OTHERS THEN
+        v_default := NULL;
+      END;
+
+      IF v_default LIKE '%nextval%' THEN
+        v_seq := substring(v_default from 'nextval\(''([^'']+)''');
+      END IF;
+    END IF;
+
+    IF v_seq IS NULL THEN
+      BEGIN
+        SELECT d.objid::regclass::text INTO v_seq
+          FROM pg_depend d
+          JOIN pg_attribute a ON a.attrelid = d.refobjid AND a.attnum = d.refobjsubid
+          JOIN pg_class sc ON sc.oid = d.objid AND sc.relkind = 'S'
+         WHERE d.refobjid = format('app_user.%I', t.tbl)::regclass
+           AND a.attname::text = t.col
+           AND d.deptype IN ('a', 'i')
+         LIMIT 1;
+      EXCEPTION WHEN OTHERS THEN
+        v_seq := NULL;
+      END;
+    END IF;
+
+    IF v_seq IS NULL THEN
+      RAISE NOTICE '3. %.% 没有关联序列（MySQL 兼容 auto_increment），跳过校准', t.tbl, t.col;
+      CONTINUE;
+    END IF;
+
+    EXECUTE format('SELECT COALESCE(max(%I), 0) FROM app_user.%I', t.col, t.tbl) INTO v_max;
+    BEGIN
+      PERFORM setval(v_seq::regclass, GREATEST(v_max, 1), true);
+      RAISE NOTICE '3. %.% 自增序列已校准 (max=%)', t.tbl, t.col, v_max;
+    EXCEPTION WHEN OTHERS THEN
+      RAISE NOTICE '3. %.% 序列校准失败: %，跳过', t.tbl, t.col, SQLERRM;
+    END;
+  END LOOP;
+END $$;
+
+-- ============================================================================
+-- 4. 预警表 early_warning
+-- ============================================================================
+DO $$
+DECLARE
+  has_device_name  boolean;
+  has_warning_name boolean;
+  v_type           text;
+  col_name         text;
+  exists_col       boolean;
+BEGIN
+  IF to_regclass('app_user.early_warning') IS NULL THEN
+    RAISE NOTICE '4. 表 app_user.early_warning 不存在，跳过';
+    RETURN;
+  END IF;
+
+  -- 4.1 设备名称列：实体映射 device_name
+  SELECT EXISTS (SELECT 1 FROM information_schema.columns
+                  WHERE table_schema = 'app_user' AND table_name = 'early_warning'
+                    AND column_name = 'device_name') INTO has_device_name;
+  SELECT EXISTS (SELECT 1 FROM information_schema.columns
+                  WHERE table_schema = 'app_user' AND table_name = 'early_warning'
+                    AND column_name = 'warning_name') INTO has_warning_name;
+
+  IF has_device_name THEN
+    RAISE NOTICE '4.1 early_warning.device_name 已存在';
+  ELSE
+    ALTER TABLE app_user.early_warning ADD COLUMN device_name varchar(256);
+    IF has_warning_name THEN
+      EXECUTE 'UPDATE app_user.early_warning SET device_name = warning_name WHERE device_name IS NULL';
+      RAISE NOTICE '4.1 已新增 device_name，并搬迁 warning_name 的数据';
+    ELSE
+      RAISE NOTICE '4.1 已新增 device_name';
+    END IF;
+  END IF;
+
+  -- 4.2 warning_level 统一为 integer（实体/查询/前端都是整数）
+  SELECT data_type INTO v_type FROM information_schema.columns
+   WHERE table_schema = 'app_user' AND table_name = 'early_warning' AND column_name = 'warning_level';
+
+  IF v_type IS NULL THEN
+    ALTER TABLE app_user.early_warning ADD COLUMN warning_level integer;
+    RAISE NOTICE '4.2 已新增 warning_level integer';
+  ELSIF v_type IN ('integer', 'smallint', 'bigint', 'numeric', 'decimal') THEN
+    RAISE NOTICE '4.2 warning_level 已是数值类型（%），跳过', v_type;
+  ELSE
+    ALTER TABLE app_user.early_warning ALTER COLUMN warning_level DROP DEFAULT;
+    ALTER TABLE app_user.early_warning
+        ALTER COLUMN warning_level TYPE integer
+        USING (CASE WHEN trim(warning_level) ~ '^[0-9]+$' THEN trim(warning_level)::integer ELSE 1 END);
+    RAISE NOTICE '4.2 warning_level 已由 % 转为 integer', v_type;
+  END IF;
+
+  -- 4.3 旧版实体遗留列：程序已不再写入，放开非空避免插入失败
+  FOREACH col_name IN ARRAY ARRAY['current_value', 'threshold', 'unit', 'warning_time'] LOOP
+    SELECT EXISTS (SELECT 1 FROM information_schema.columns
+                    WHERE table_schema = 'app_user' AND table_name = 'early_warning'
+                      AND column_name = col_name) INTO exists_col;
+    IF exists_col THEN
+      EXECUTE format('ALTER TABLE app_user.early_warning ALTER COLUMN %I DROP NOT NULL', col_name);
+      RAISE NOTICE '4.3 early_warning.% 已改为可空', col_name;
+    END IF;
+  END LOOP;
+END $$;
+
+-- 4.4 指定人员发布的接收人表
+CREATE TABLE IF NOT EXISTS app_user.psgw_warning_scope_user (
+    id           varchar(64) NOT NULL,
+    warning_id   varchar(64),
+    user_id      bigint,
+    user_name    varchar(128),
+    approved     integer DEFAULT 0,
+    approve_time timestamp,
+    create_time  timestamp,
+    PRIMARY KEY (id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_psgw_warning_scope_user_warning
+    ON app_user.psgw_warning_scope_user (warning_id);
+
+COMMENT ON TABLE app_user.psgw_warning_scope_user IS '排水预警接收人员';
+
+COMMIT;
+
+-- ============================================================================
+-- 核验（可选，取消注释后单独执行）
+-- ============================================================================
+-- SELECT column_name, data_type FROM information_schema.columns
+--  WHERE table_schema = 'app_user' AND table_name = 'drainage_waterlogging_point'
+--  ORDER BY ordinal_position;
+--
+-- SELECT dict_value, dict_label FROM app_user.sys_dict_data
+--  WHERE dict_type = 'drainage_district' ORDER BY dict_sort;

pipe-network-service/sql/migration_drainage_district_dict.sql

@@ -0,0 +1,195 @@
+-- 排水「所属片区」字典（KingbaseES）
+-- 用途：排水子系统里凡是出现「所属区域 / 所属片区」的下拉框，统一读取本字典，
+--      不再从业务表的存量值里去重生成选项（旧做法会把脏数据回灌到下拉里）。
+-- 口径：与既有 DistrictEnum（com.zksy.base.enums.DistrictEnum）、
+--      equipment_base.district 保持一致：0-城南 / 1-城北 / 2-太常。
+--      库中存编码，中文只在前端展示。
+-- 影响：drainage_waterlogging_point.district（varchar）由「中文地名」改为「编码」。
+-- 说明：脚本幂等，可重复执行；只维护字典 + 归一存量片区值，不改表结构。
+--
+-- =====================================================================
+-- 为什么本脚本要「显式指定主键」
+-- ---------------------------------------------------------------------
+-- 本库由 MySQL 迁移至 Kingbase，sys_dict_data / sys_dict_type 建表沿用了 MySQL 的
+-- auto_increment（KES 的 MySQL 兼容自增，不是 PG 的 serial/identity）。迁数据时是带显式
+-- 主键写入的，自增计数器没有跟着推进，于是「INSERT 不写主键」会发出一个已被占用的号：
+--     ERROR: 重复键违反唯一约束 "sys_dict_data_pkey"
+--     DETAIL: 键值 "dict_code"=(415) 已经存在
+-- 且该列不是 identity 列，pg_get_serial_sequence / pg_depend 都拿不到序列，
+-- ALTER COLUMN ... RESTART WITH 也会报「不是标识列」。
+--
+-- 因此处理方式是：
+--   ① INSERT 一律写成 max(主键) + n，显式指定，彻底绕开自增计数器；
+--   ② 插入完再把能找到的序列 setval 到最新 max（步骤 5），让后续新增不受影响；
+--      若确实没有关联序列（纯 auto_increment），步骤 5 只提示、不报错。
+-- =====================================================================
+--
+-- 前置（必须先执行）：结束上一次失败执行遗留的 aborted 事务
+--   上一次语句报错后，PG/KES 会把该事务标记为已失败，此时同一连接上后续任何语句都只返回
+--   「当前事务被终止，事务块结束之前的查询被忽略」，必须先回滚才能继续。
+--   注意：本句会回滚该连接上所有未提交事务；若你有别的未提交手工操作，请先单独提交再跑本脚本。
+ROLLBACK;
+
+BEGIN;
+
+-- 1. 字典类型（幂等登记；dict_id 显式取 max+1）
+INSERT INTO app_user.sys_dict_type
+    (dict_id, dict_name, dict_type, status, create_by, create_time, remark)
+SELECT (SELECT COALESCE(MAX(dict_id), 0) FROM app_user.sys_dict_type) + 1,
+       '所属片区', 'drainage_district', '0', 'admin', CURRENT_TIMESTAMP,
+       '排水所属片区：0城南 1城北 2太常（与 DistrictEnum 对齐）'
+ WHERE NOT EXISTS (SELECT 1 FROM app_user.sys_dict_type WHERE dict_type = 'drainage_district');
+
+UPDATE app_user.sys_dict_type
+   SET dict_name = '所属片区',
+       status = '0',
+       update_by = 'admin',
+       update_time = CURRENT_TIMESTAMP,
+       remark = '排水所属片区：0城南 1城北 2太常（与 DistrictEnum 对齐）'
+ WHERE dict_type = 'drainage_district';
+
+-- 2. 字典数据（幂等重写；dict_code 显式取 max+序号）
+--    片区不是严重性等级，list_class 统一留空走默认色，不做颜色区分。
+DELETE FROM app_user.sys_dict_data WHERE dict_type = 'drainage_district';
+
+INSERT INTO app_user.sys_dict_data
+    (dict_code, dict_sort, dict_label, dict_value, dict_type, css_class, list_class,
+     is_default, status, create_by, create_time, remark)
+SELECT (SELECT COALESCE(MAX(dict_code), 0) FROM app_user.sys_dict_data) + v.rn,
+       v.dict_sort, v.dict_label, v.dict_value, 'drainage_district',
+       '', '', v.is_default, '0', 'admin', CURRENT_TIMESTAMP, v.remark
+  FROM (VALUES
+        (1, 1, '城南', '0', 'N', '所属片区：城南'),
+        (2, 2, '城北', '1', 'N', '所属片区：城北'),
+        (3, 3, '太常', '2', 'N', '所属片区：太常')
+       ) AS v(rn, dict_sort, dict_label, dict_value, is_default, remark)
+ ORDER BY v.rn;
+
+-- 3. 列注释对齐（幂等，可重复执行）
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.district
+    IS '所属片区（字典 drainage_district：0城南 1城北 2太常）';
+
+-- 4. 存量片区值归一：中文地名 → 编码
+--    只处理能精确匹配的三个片区；其余值（历史脏数据，如乱填的 '123123'）
+--    一律原样保留，交由步骤 6 的体检输出，由人工确认后再处理，避免误伤真实数据。
+UPDATE app_user.drainage_waterlogging_point
+   SET district = CASE TRIM(district)
+                    WHEN '城南' THEN '0'
+                    WHEN '城北' THEN '1'
+                    WHEN '太常' THEN '2'
+                  END
+ WHERE TRIM(district) IN ('城南', '城北', '太常');
+
+-- 5. 自增计数器校准（放在插入之后：此时 max 已含新插入的 3 行）
+--    三种方式依次尝试，任一成功即可；都不成功只提示，不影响本脚本已插入的数据。
+--      方式一 pg_get_serial_sequence（serial / identity）
+--      方式二 从列默认值里解析 nextval('...')（KES 常见形态）
+--      方式三 pg_depend 关联序列（兜底）
+DO $$
+DECLARE
+  t         record;
+  v_seq     text;
+  v_max     bigint;
+  v_default text;
+BEGIN
+  FOR t IN
+    SELECT * FROM (VALUES ('sys_dict_data', 'dict_code'),
+                          ('sys_dict_type', 'dict_id')) AS x(tbl, col)
+  LOOP
+    v_seq := NULL;
+    v_default := NULL;
+
+    -- 方式一
+    BEGIN
+      v_seq := pg_get_serial_sequence(format('app_user.%I', t.tbl), t.col);
+    EXCEPTION WHEN OTHERS THEN
+      v_seq := NULL;
+    END;
+
+    -- 方式二：解析列默认值里的 nextval('xxx')
+    IF v_seq IS NULL THEN
+      BEGIN
+        SELECT pg_get_expr(ad.adbin, ad.adrelid) INTO v_default
+          FROM pg_attribute a
+          JOIN pg_attrdef ad ON ad.adrelid = a.attrelid AND ad.adnum = a.attnum
+         WHERE a.attrelid = format('app_user.%I', t.tbl)::regclass
+           AND a.attname::text = t.col
+           AND a.attnum > 0
+           AND NOT a.attisdropped;
+      EXCEPTION WHEN OTHERS THEN
+        v_default := NULL;
+      END;
+
+      IF v_default LIKE '%nextval%' THEN
+        v_seq := substring(v_default from 'nextval\(''([^'']+)''');
+      END IF;
+    END IF;
+
+    -- 方式三
+    IF v_seq IS NULL THEN
+      BEGIN
+        SELECT d.objid::regclass::text INTO v_seq
+          FROM pg_depend d
+          JOIN pg_attribute a ON a.attrelid = d.refobjid AND a.attnum = d.refobjsubid
+          JOIN pg_class sc ON sc.oid = d.objid AND sc.relkind = 'S'
+         WHERE d.refobjid = format('app_user.%I', t.tbl)::regclass
+           AND a.attname::text = t.col
+           AND d.deptype IN ('a', 'i')
+         LIMIT 1;
+      EXCEPTION WHEN OTHERS THEN
+        v_seq := NULL;
+      END;
+    END IF;
+
+    IF v_seq IS NULL THEN
+      RAISE NOTICE '%.% 没有关联序列（应为 MySQL 兼容 auto_increment），跳过校准。本次数据已显式指定主键插入，不影响本脚本；但后续在该表手工新增行仍可能撞主键。',
+                   t.tbl, t.col;
+      CONTINUE;
+    END IF;
+
+    EXECUTE format('SELECT COALESCE(max(%I), 0) FROM app_user.%I', t.col, t.tbl) INTO v_max;
+    BEGIN
+      PERFORM setval(v_seq::regclass, GREATEST(v_max, 1), true);
+      RAISE NOTICE '%.% 自增序列已校准: % (max=%, next=%)', t.tbl, t.col, v_seq, v_max, v_max + 1;
+    EXCEPTION WHEN OTHERS THEN
+      RAISE NOTICE '%.% 序列校准失败: %，跳过', t.tbl, t.col, SQLERRM;
+    END;
+  END LOOP;
+END $$;
+
+-- 6. 存量数据体检（只提示不改数据）：片区值不在 0/1/2 的记录
+--    出现提示说明该行是历史脏数据，前端下拉已由字典接管不会再产生新脏值，
+--    存量部分需要人工判定归属后单独 UPDATE。
+DO $$
+DECLARE
+  r      record;
+  v_rows bigint := 0;
+BEGIN
+  FOR r IN
+    SELECT district, COUNT(*) AS cnt
+      FROM app_user.drainage_waterlogging_point
+     WHERE del_flag = '0'
+       AND (district IS NULL OR TRIM(district) = '' OR TRIM(district) NOT IN ('0', '1', '2'))
+     GROUP BY district
+     ORDER BY COUNT(*) DESC
+  LOOP
+    IF r.district IS NULL OR TRIM(r.district) = '' THEN
+      RAISE NOTICE '片区待核对: (空值) —— % 条', r.cnt;
+    ELSE
+      RAISE NOTICE '片区待核对: % —— % 条', r.district, r.cnt;
+    END IF;
+    v_rows := v_rows + r.cnt;
+  END LOOP;
+
+  IF v_rows = 0 THEN
+    RAISE NOTICE '片区值全部合法（0/1/2），无需处理。';
+  ELSE
+    RAISE NOTICE '共 % 条记录片区值不合法，请人工核对后处理。', v_rows;
+  END IF;
+END $$;
+
+-- 7. 结果核验（可选，取消注释执行）：应返回 3 行，dict_value 依次 0/1/2
+--    SELECT dict_value, dict_label FROM app_user.sys_dict_data
+--     WHERE dict_type = 'drainage_district' ORDER BY dict_sort;
+
+COMMIT;

pipe-network-service/sql/migration_drainage_waterlogging_duty.sql

@@ -0,0 +1,114 @@
+-- 易积水点：责任单位 / 责任人改为关联系统部门与用户
+--
+-- 背景：原表只存 duty_unit / duty_person / duty_phone 三个自由文本，
+-- 与 sys_dept / sys_user 没有任何关联。带来的问题是：
+--   · 同一个部门被手输成多种写法，统计口径无法收敛；
+--   · 人员调岗、改名后台账失真，也点不进去看这个人是谁；
+--   · 联系电话靠手填，容易与人事信息不一致。
+--
+-- 本脚本做两件事：
+--   1. 幂等补两列：duty_dept_id（sys_dept.dept_id）、duty_user_id（sys_user.user_id）；
+--   2. 尽力回填：按名称精确匹配 sys_dept.dept_name / sys_user.nick_name，
+--      匹配上的补 ID，匹配不上的保持 NULL（不猜、不改已有文本）。
+--
+-- 保留原有三列作为「快照」：
+--   duty_unit / duty_person / duty_phone 继续存当时的名称与号码。
+--   列表、地图、详情直接读快照，不必回表 join；人员改名也不影响历史记录。
+--
+-- 前置：结束上一次失败执行遗留的 aborted 事务（事务外执行只会告警，不报错）
+ROLLBACK;
+
+BEGIN;
+
+-- =====================================================================
+-- 1. 幂等补列
+-- =====================================================================
+DO $$
+DECLARE
+  col record;
+BEGIN
+  IF to_regclass('app_user.drainage_waterlogging_point') IS NULL THEN
+    RAISE NOTICE '表 app_user.drainage_waterlogging_point 不存在，请先执行 migration_drainage_waterlogging_point.sql';
+    RETURN;
+  END IF;
+
+  FOR col IN
+    SELECT * FROM (VALUES
+        ('duty_dept_id', 'bigint'),
+        ('duty_user_id', 'bigint')
+      ) AS x(name, type)
+  LOOP
+    IF NOT EXISTS (
+        SELECT 1 FROM information_schema.columns
+         WHERE table_schema = 'app_user'
+           AND table_name = 'drainage_waterlogging_point'
+           AND column_name = col.name
+    ) THEN
+      EXECUTE format('ALTER TABLE app_user.drainage_waterlogging_point ADD COLUMN %I %s',
+                     col.name, col.type);
+      RAISE NOTICE '已补列：%', col.name;
+    END IF;
+  END LOOP;
+END $$;
+
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_dept_id IS '责任单位（sys_dept.dept_id，duty_unit 存其名称快照）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_user_id IS '责任人（sys_user.user_id，duty_person / duty_phone 存其姓名与电话快照）';
+
+CREATE INDEX IF NOT EXISTS idx_dwp_duty_dept ON app_user.drainage_waterlogging_point (duty_dept_id);
+CREATE INDEX IF NOT EXISTS idx_dwp_duty_user ON app_user.drainage_waterlogging_point (duty_user_id);
+
+-- =====================================================================
+-- 2. 存量数据回填（只做精确匹配，匹配不上就留空）
+--    同一名称对应多个部门/用户时取 ID 最小的那个，避免一对多歧义放大。
+-- =====================================================================
+DO $$
+DECLARE
+  v_dept integer := 0;
+  v_user integer := 0;
+BEGIN
+  IF to_regclass('app_user.drainage_waterlogging_point') IS NULL THEN
+    RETURN;
+  END IF;
+
+  -- 2.1 责任单位：duty_unit 文本 → dept_id
+  UPDATE app_user.drainage_waterlogging_point p
+     SET duty_dept_id = d.dept_id
+    FROM (
+      SELECT dept_name, MIN(dept_id) AS dept_id
+        FROM app_user.sys_dept
+       WHERE del_flag = '0'
+       GROUP BY dept_name
+    ) d
+   WHERE p.duty_dept_id IS NULL
+     AND p.duty_unit IS NOT NULL
+     AND trim(p.duty_unit) = d.dept_name;
+  GET DIAGNOSTICS v_dept = ROW_COUNT;
+
+  -- 2.2 责任人：duty_person 文本 → user_id
+  UPDATE app_user.drainage_waterlogging_point p
+     SET duty_user_id = u.user_id
+    FROM (
+      SELECT nick_name, MIN(user_id) AS user_id
+        FROM app_user.sys_user
+       WHERE del_flag = '0'
+       GROUP BY nick_name
+    ) u
+   WHERE p.duty_user_id IS NULL
+     AND p.duty_person IS NOT NULL
+     AND trim(p.duty_person) = u.nick_name;
+  GET DIAGNOSTICS v_user = ROW_COUNT;
+
+  RAISE NOTICE '存量回填：责任单位匹配 % 条，责任人匹配 % 条', v_dept, v_user;
+END $$;
+
+-- =====================================================================
+-- 3. 结果核验（可选，取消注释执行）
+-- =====================================================================
+-- 还没关联上的记录（需要人工在页面上重新选一次部门/人员）
+--    SELECT point_code, point_name, duty_unit, duty_person
+--      FROM app_user.drainage_waterlogging_point
+--     WHERE del_flag = '0'
+--       AND ((duty_unit IS NOT NULL AND duty_dept_id IS NULL)
+--         OR (duty_person IS NOT NULL AND duty_user_id IS NULL));
+
+COMMIT;

pipe-network-service/sql/migration_drainage_waterlogging_point.sql

@@ -0,0 +1,239 @@
+-- 易积水点台账表（KingbaseES, schema app_user）
+--
+-- 背景：旧表 app_user.psgw_waterlogging_point 混了「监测点」口径的字段，存在三类问题：
+--   1) time（数据时间）、waterlogging_count（易积水点数量）属于运行数据/派生值，
+--      不该落在基础档案表里；
+--   2) waterlogging_points 存多节点坐标 JSON，但一个易积水点在地图上就是一个位置，
+--      应直接落 WGS84 经纬度；
+--   3) equipment_id 实际存的是「视频设备 ID（equipment_base.equipment_type_id=20）」，
+--      易积水点关联的是摄像头，不是通用监测设备，字段名应表达真实语义。
+--
+-- 本脚本做四件事：
+--   1. 建新表 app_user.drainage_waterlogging_point；
+--   2. 对「已建过表」的库幂等补列（见步骤 1.1），两种执行顺序都能到位；
+--   3. 登记字典「风险等级」（积水程度沿用已有字典 drainage_waterlogging_level）；
+--   4. 从旧表迁移数据（坐标取 waterlogging_points 的第一个点；解析失败置 NULL）。
+--
+-- 字段口径刻意保持精简（不做全量台账）：
+--   基本     point_code / point_name / district / location / longitude / latitude
+--   积水     waterlogging_level / water_depth / water_volume / water_area
+--   管理     risk_level / duty_unit / duty_person / duty_phone
+--            （后三项为名称快照，另有 duty_dept_id / duty_user_id 指向 sys_dept / sys_user，
+--              由 migration_drainage_waterlogging_duty.sql 维护）
+--   关联其他 camera_ids / remark / 审计字段 / del_flag
+--
+-- 说明：
+--   · 旧表的 equipment_id 直接搬进 camera_ids。若历史数据里曾混入过普通监测设备 ID，
+--     迁移后需要人工核对（旧页面早期版本确实调用过通用设备关联）。
+--   · 幂等：建表/补列/索引/字典均可重复执行；数据迁移仅在「新表为空」时执行一次。
+--
+-- 前置：结束上一次失败执行遗留的 aborted 事务（事务外执行只会告警，不报错）
+ROLLBACK;
+
+BEGIN;
+
+-- =====================================================================
+-- 1. 建新表
+-- =====================================================================
+CREATE TABLE IF NOT EXISTS app_user.drainage_waterlogging_point (
+    point_id           varchar(64)    NOT NULL,
+    point_code         varchar(64),
+    point_name         varchar(200)   NOT NULL,
+    district           varchar(100),
+    location           varchar(500),
+    longitude          numeric(12, 8),
+    latitude           numeric(12, 8),
+    waterlogging_level int2           DEFAULT 0,
+    water_depth        numeric(10, 2),
+    water_volume       numeric(14, 2),
+    water_area         numeric(14, 2),
+    risk_level         varchar(8),
+    duty_unit          varchar(200),
+    duty_person        varchar(64),
+    duty_phone         varchar(32),
+    duty_dept_id       bigint,
+    duty_user_id       bigint,
+    camera_ids         varchar(500),
+    remark             varchar(500),
+    create_by          varchar(64),
+    create_time        timestamp,
+    update_by          varchar(64),
+    update_time        timestamp,
+    del_flag           char(1)        DEFAULT '0',
+    CONSTRAINT pk_drainage_waterlogging_point PRIMARY KEY (point_id)
+);
+
+-- ---------------------------------------------------------------------
+-- 1.1 幂等补列：兼容「表已按旧版本建过」的库
+--     逐列判断后 ALTER，已存在的列直接跳过。
+-- ---------------------------------------------------------------------
+DO $$
+DECLARE
+  col record;
+BEGIN
+  IF to_regclass('app_user.drainage_waterlogging_point') IS NULL THEN
+    RETURN;
+  END IF;
+
+  FOR col IN
+    SELECT * FROM (VALUES
+        ('water_depth', 'numeric(10,2)'),
+        ('water_area',  'numeric(14,2)'),
+        ('risk_level',  'varchar(8)'),
+        ('duty_unit',   'varchar(200)'),
+        ('duty_person', 'varchar(64)'),
+        ('duty_phone',  'varchar(32)'),
+        ('duty_dept_id', 'bigint'),
+        ('duty_user_id', 'bigint')
+      ) AS x(name, type)
+  LOOP
+    IF NOT EXISTS (
+        SELECT 1 FROM information_schema.columns
+         WHERE table_schema = 'app_user'
+           AND table_name = 'drainage_waterlogging_point'
+           AND column_name = col.name
+    ) THEN
+      EXECUTE format('ALTER TABLE app_user.drainage_waterlogging_point ADD COLUMN %I %s',
+                     col.name, col.type);
+      RAISE NOTICE '已补列：%', col.name;
+    END IF;
+  END LOOP;
+END $$;
+
+COMMENT ON TABLE  app_user.drainage_waterlogging_point IS '易积水点台账';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.point_id           IS '点位ID（主键）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.point_code         IS '点位编码';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.point_name         IS '点位名称';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.district           IS '所属区域';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.location           IS '位置描述';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.longitude          IS 'WGS84 经度';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.latitude           IS 'WGS84 纬度';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.waterlogging_level IS '积水程度（字典 drainage_waterlogging_level：0未积水 1轻度 2中度 3严重）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.water_depth        IS '积水深度（cm）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.water_volume       IS '积水量（m³）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.water_area         IS '积水面积（m²）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.risk_level         IS '风险等级（字典 drainage_waterlogging_risk_level：0低风险 1中风险 2高风险）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_unit          IS '责任单位（名称快照，对应 duty_dept_id）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_person        IS '责任人（姓名快照，对应 duty_user_id）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_phone         IS '联系电话（快照，选人时取 sys_user.phonenumber）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_dept_id       IS '责任单位ID（sys_dept.dept_id）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.duty_user_id       IS '责任人ID（sys_user.user_id）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.camera_ids         IS '关联摄像头ID（equipment_base.equipment_id，多个用逗号分隔）';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.remark             IS '备注';
+COMMENT ON COLUMN app_user.drainage_waterlogging_point.del_flag           IS '删除标志（0正常 2删除）';
+
+CREATE INDEX IF NOT EXISTS idx_dwp_point_code ON app_user.drainage_waterlogging_point (point_code);
+CREATE INDEX IF NOT EXISTS idx_dwp_point_name ON app_user.drainage_waterlogging_point (point_name);
+CREATE INDEX IF NOT EXISTS idx_dwp_district   ON app_user.drainage_waterlogging_point (district);
+CREATE INDEX IF NOT EXISTS idx_dwp_level      ON app_user.drainage_waterlogging_point (waterlogging_level);
+CREATE INDEX IF NOT EXISTS idx_dwp_risk_level ON app_user.drainage_waterlogging_point (risk_level);
+
+-- =====================================================================
+-- 2. 字典：风险等级
+--    （积水程度沿用已有字典 drainage_waterlogging_level，本脚本不重复建）
+--    注意：本库 sys_dict_data / sys_dict_type 是 MySQL 兼容 auto_increment，
+--    自增计数器未随迁移数据推进，因此主键一律显式取 max+序号。
+-- =====================================================================
+INSERT INTO app_user.sys_dict_type
+    (dict_id, dict_name, dict_type, status, create_by, create_time, remark)
+SELECT (SELECT COALESCE(MAX(dict_id), 0) FROM app_user.sys_dict_type) + 1,
+       '易积水点风险等级', 'drainage_waterlogging_risk_level', '0', 'admin', CURRENT_TIMESTAMP,
+       '风险等级：0低风险 1中风险 2高风险'
+ WHERE NOT EXISTS (SELECT 1 FROM app_user.sys_dict_type WHERE dict_type = 'drainage_waterlogging_risk_level');
+
+UPDATE app_user.sys_dict_type
+   SET dict_name = '易积水点风险等级',
+       status = '0',
+       update_by = 'admin',
+       update_time = CURRENT_TIMESTAMP,
+       remark = '风险等级：0低风险 1中风险 2高风险'
+ WHERE dict_type = 'drainage_waterlogging_risk_level';
+
+DELETE FROM app_user.sys_dict_data WHERE dict_type = 'drainage_waterlogging_risk_level';
+
+INSERT INTO app_user.sys_dict_data
+    (dict_code, dict_sort, dict_label, dict_value, dict_type, css_class, list_class,
+     is_default, status, create_by, create_time, remark)
+SELECT (SELECT COALESCE(MAX(dict_code), 0) FROM app_user.sys_dict_data) + v.rn,
+       v.dict_sort, v.dict_label, v.dict_value, 'drainage_waterlogging_risk_level',
+       '', v.list_class, v.is_default, '0', 'admin', CURRENT_TIMESTAMP, v.remark
+  FROM (VALUES
+        (1, 1, '低风险', '0', 'success', 'Y', '风险等级：低风险'),
+        (2, 2, '中风险', '1', 'warning', 'N', '风险等级：中风险'),
+        (3, 3, '高风险', '2', 'danger',  'N', '风险等级：高风险')
+       ) AS v(rn, dict_sort, dict_label, dict_value, list_class, is_default, remark)
+ ORDER BY v.rn;
+
+-- =====================================================================
+-- 3. 旧表数据迁移（仅当新表为空时执行一次）
+--    可迁移的只有旧表真实存在的语义：名称/区域/积水量/积水程度/摄像头/坐标。
+-- =====================================================================
+DO $$
+DECLARE
+  v_idx integer := 0;
+  r     record;
+  v_lng numeric(12, 8);
+  v_lat numeric(12, 8);
+BEGIN
+  IF to_regclass('app_user.psgw_waterlogging_point') IS NULL THEN
+    RAISE NOTICE '旧表 app_user.psgw_waterlogging_point 不存在，跳过数据迁移';
+    RETURN;
+  END IF;
+
+  IF EXISTS (SELECT 1 FROM app_user.drainage_waterlogging_point) THEN
+    RAISE NOTICE '新表 app_user.drainage_waterlogging_point 已有数据，跳过迁移（避免重复导入）';
+    RETURN;
+  END IF;
+
+  FOR r IN SELECT * FROM app_user.psgw_waterlogging_point LOOP
+    v_idx := v_idx + 1;
+    v_lng := NULL;
+    v_lat := NULL;
+
+    BEGIN
+      IF r.waterlogging_points IS NOT NULL AND r.waterlogging_points <> '' THEN
+        v_lng := (regexp_match(r.waterlogging_points, '"lng"\s*:\s*(-?[0-9]+(\.[0-9]+)?)'))[1]::numeric(12, 8);
+        v_lat := (regexp_match(r.waterlogging_points, '"lat"\s*:\s*(-?[0-9]+(\.[0-9]+)?)'))[1]::numeric(12, 8);
+      END IF;
+    EXCEPTION WHEN OTHERS THEN
+      v_lng := NULL;
+      v_lat := NULL;
+    END;
+
+    INSERT INTO app_user.drainage_waterlogging_point (
+        point_id, point_code, point_name, district, location,
+        longitude, latitude, waterlogging_level, water_volume, camera_ids,
+        create_by, create_time, update_by, update_time, del_flag
+    ) VALUES (
+        r.id,
+        'JSD-' || lpad(v_idx::text, 4, '0'),
+        COALESCE(NULLIF(trim(r.name), ''), '未命名易积水点'),
+        r.area,
+        NULL,
+        v_lng,
+        v_lat,
+        COALESCE(r.waterlogging_level, 0),
+        r.water_volume,
+        r.equipment_id,
+        'admin', COALESCE(r.create_time, CURRENT_TIMESTAMP),
+        'admin', COALESCE(r.update_time, r.create_time, CURRENT_TIMESTAMP),
+        '0'
+    );
+  END LOOP;
+
+  RAISE NOTICE '旧表数据迁移完成，共迁移 % 条', v_idx;
+EXCEPTION WHEN OTHERS THEN
+  RAISE NOTICE '旧表数据迁移跳过：%', SQLERRM;
+END $$;
+
+-- =====================================================================
+-- 4. 结果核验（可选，取消注释执行）
+-- =====================================================================
+-- 风险等级分布
+--    SELECT risk_level, count(*) FROM app_user.drainage_waterlogging_point
+--     WHERE del_flag = '0' GROUP BY risk_level;
+-- 风险等级字典（应返回 3 行）
+--    SELECT dict_value, dict_label, list_class FROM app_user.sys_dict_data
+--     WHERE dict_type = 'drainage_waterlogging_risk_level' ORDER BY dict_sort;
+
+COMMIT;

pipe-network-service/sql/migration_early_warning_columns.sql

@@ -0,0 +1,146 @@
+-- ============================================================================
+-- 预警表（app_user.early_warning）列补齐 / 类型归一
+--
+-- 背景
+--   排水「设备运行实时监测」页发布预警时插入失败，报：
+--     ERROR: 关系 "early_warning" 的 "warning_name" 字段不存在
+--   实体 DrainageWarning.equipmentName 原先映射到 warning_name，
+--   而早期建表脚本（sql/early_warning_system.sql）里设备名称列是 device_name。
+--   实体已改为 device_name；本脚本负责把历史库的列形态补齐/统一。
+--
+-- 覆盖三种情况
+--   1) 有 device_name        → 跳过
+--   2) 只有 warning_name     → 新增 device_name，并把数据搬过去
+--   3) 两者都没有（异常库）  → 新增 device_name
+--   并把 warning_level 由字符型统一为 integer（实体、查询、前端均为整数）。
+--
+-- 注意：不给 early_warning 加坐标列。坐标只在查询时从设备档案补齐后返回前端，
+--   属于展示派生字段，落库会让预警历史随设备改动而失真。
+--
+-- 幂等：可重复执行，已到位的环境只会输出 NOTICE，不做任何改动。
+-- ============================================================================
+
+-- 首句独立 ROLLBACK：清掉上次失败执行残留的 aborted 事务，
+-- 否则重跑只会得到「当前事务被终止，事务块结束之前的查询被忽略」
+ROLLBACK;
+
+BEGIN;
+
+SET search_path TO app_user, public;
+
+-- ----------------------------------------------------------------------------
+-- 1. device_name：实体 DrainageWarning.equipmentName 的映射列
+-- ----------------------------------------------------------------------------
+DO $$
+DECLARE
+    has_device_name  boolean;
+    has_warning_name boolean;
+BEGIN
+    SELECT EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'app_user' AND table_name = 'early_warning' AND column_name = 'device_name'
+    ) INTO has_device_name;
+
+    SELECT EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'app_user' AND table_name = 'early_warning' AND column_name = 'warning_name'
+    ) INTO has_warning_name;
+
+    IF has_device_name THEN
+        RAISE NOTICE 'early_warning.device_name 已存在，跳过';
+    ELSE
+        ALTER TABLE app_user.early_warning ADD COLUMN device_name varchar(256);
+        IF has_warning_name THEN
+            EXECUTE 'UPDATE app_user.early_warning SET device_name = warning_name WHERE device_name IS NULL';
+            RAISE NOTICE '已新增 device_name，并把 warning_name 的数据搬迁过来';
+        ELSE
+            RAISE NOTICE '已新增 device_name（库中原本没有设备名称列）';
+        END IF;
+    END IF;
+END $$;
+
+COMMENT ON COLUMN app_user.early_warning.device_name IS '关联设备名称';
+
+-- ----------------------------------------------------------------------------
+-- 2. warning_level：字符型统一为 integer
+--    实体字段是 Integer、查询条件也是整数；character varying 列在 KES 下
+--    既可能插入报类型不符，也可能让 warning_level = 1 的等值查询走不到索引。
+--    非数字历史值（如误填的中文）统一归一为 1，避免转换失败中断脚本。
+-- ----------------------------------------------------------------------------
+DO $$
+DECLARE
+    v_type text;
+BEGIN
+    SELECT data_type INTO v_type
+    FROM information_schema.columns
+    WHERE table_schema = 'app_user' AND table_name = 'early_warning' AND column_name = 'warning_level';
+
+    IF v_type IS NULL THEN
+        RAISE NOTICE 'early_warning.warning_level 不存在，跳过';
+    ELSIF v_type IN ('integer', 'smallint', 'bigint', 'numeric', 'decimal') THEN
+        RAISE NOTICE 'early_warning.warning_level 已是数值类型（%），跳过', v_type;
+    ELSE
+        ALTER TABLE app_user.early_warning ALTER COLUMN warning_level DROP DEFAULT;
+        ALTER TABLE app_user.early_warning
+            ALTER COLUMN warning_level TYPE integer
+            USING (CASE WHEN trim(warning_level) ~ '^[0-9]+$' THEN trim(warning_level)::integer ELSE 1 END);
+        RAISE NOTICE 'early_warning.warning_level 已由 % 转为 integer（非数字历史值已归一为 1）', v_type;
+    END IF;
+END $$;
+
+COMMENT ON COLUMN app_user.early_warning.warning_level IS '预警级别（1严重 2重要 3一般 4常态）';
+
+-- ----------------------------------------------------------------------------
+-- 3. psgw_warning_scope_user：指定人员发布时的接收人列表
+--    排水预警（DrainageWarningServiceImpl.publishWarning）在 publish_scope=3 时写入该表，
+--    早期环境可能没有建表，导致「指定人员」发布直接失败。
+-- ----------------------------------------------------------------------------
+CREATE TABLE IF NOT EXISTS app_user.psgw_warning_scope_user (
+    id           varchar(64) NOT NULL,
+    warning_id   varchar(64),
+    user_id      bigint,
+    user_name    varchar(128),
+    approved     integer DEFAULT 0,
+    approve_time timestamp,
+    create_time  timestamp,
+    PRIMARY KEY (id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_psgw_warning_scope_user_warning ON app_user.psgw_warning_scope_user (warning_id);
+
+COMMENT ON TABLE app_user.psgw_warning_scope_user IS '排水预警接收人员';
+
+-- ----------------------------------------------------------------------------
+-- 4. 旧版实体遗留列：current_value / threshold / unit / warning_time
+--    这些列曾经被旧版 DrainageWarning 写入，现已被实体标为不落库。
+--    若历史环境把它们建成了 NOT NULL 且无默认值，新代码不写就会触发非空约束，
+--    这里统一放开非空限制（列本身保留，历史数据不动）。
+-- ----------------------------------------------------------------------------
+DO $$
+DECLARE
+    col_name text;
+    exists_col boolean;
+BEGIN
+    FOREACH col_name IN ARRAY ARRAY['current_value', 'threshold', 'unit', 'warning_time'] LOOP
+        SELECT EXISTS (
+            SELECT 1 FROM information_schema.columns
+            WHERE table_schema = 'app_user' AND table_name = 'early_warning' AND column_name = col_name
+        ) INTO exists_col;
+
+        IF exists_col THEN
+            EXECUTE format('ALTER TABLE app_user.early_warning ALTER COLUMN %I DROP NOT NULL', col_name);
+            RAISE NOTICE 'early_warning.% 已改为可空（该列不再由程序写入）', col_name;
+        END IF;
+    END LOOP;
+END $$;
+
+COMMIT;
+
+-- ----------------------------------------------------------------------------
+-- 核验（可选，取消注释后单独执行）
+-- ----------------------------------------------------------------------------
+-- SELECT column_name, data_type, is_nullable
+-- FROM information_schema.columns
+-- WHERE table_schema = 'app_user' AND table_name = 'early_warning'
+--   AND column_name IN ('device_name','warning_name','warning_level')
+-- ORDER BY column_name;

pipe-network-service/zksy-admin/src/main/java/com/zksy/web/controller/base/EquipmentBaseController.java

@@ -40,9 +40,13 @@ public class EquipmentBaseController {
                            @RequestParam(required = false) Integer rootTypeId,
                            @ApiParam("设备类型ID集合") @RequestParam(required = false) List<Integer> equipmentTypeIds,
                            @ApiParam("在线状态（0-离线，1-在线）") @RequestParam(required = false) Integer onlineStatus,
-                           @ApiParam("报警状态（1-报警，仅勾选报警时传）") @RequestParam(required = false) Integer alarmStatus) throws Exception {
+                           @ApiParam("报警状态（0-正常，1-报警）") @RequestParam(required = false) Integer alarmStatus,
+                           @ApiParam("安装位置（模糊匹配）") @RequestParam(required = false) String equipmentLocation,
+                           @ApiParam("所属片区（0-城南，1-城北，2-太常）") @RequestParam(required = false) Integer district,
+                           @ApiParam("设备状态（1-在用，2-闲置，3-维修，4-报废，5-待入库）")
+                           @RequestParam(required = false) Integer currentStatus) throws Exception {
         return service.findByPage(pageNum, pageSize, equipmentCode, equipmentName, equipmentTypeId,
-                rootTypeId, equipmentTypeIds, onlineStatus, alarmStatus);
+                rootTypeId, equipmentTypeIds, onlineStatus, alarmStatus, equipmentLocation, district, currentStatus);
     }
 
 

pipe-network-service/zksy-admin/src/main/java/com/zksy/web/controller/drainage/DrainageRiskController.java

@@ -0,0 +1,76 @@
+package com.zksy.web.controller.drainage;
+
+import com.zksy.common.annotation.Anonymous;
+import com.zksy.common.annotation.Log;
+import com.zksy.common.core.domain.AjaxResult;
+import com.zksy.common.enums.BusinessType;
+import com.zksy.drainage.dto.DrainageRiskPageInDTO;
+import com.zksy.drainage.service.DrainageRiskService;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * 排水管网综合风险评估
+ *
+ * 评估对象固定为 pipe_network_base 中 network_type=1 的排水管网，
+ * 关联数据（设备、报警、预警、易积水点、工程设施）也只取排水口径，不含燃气与供水。
+ */
+@Slf4j
+@RestController
+@RequestMapping("/api/drainage/risk")
+@Api(tags = "排水管网综合风险评估")
+public class DrainageRiskController {
+
+    @Autowired
+    private DrainageRiskService drainageRiskService;
+
+    @Anonymous
+    @GetMapping("/page")
+    @ApiOperation("排水管网综合风险分页查询")
+    public AjaxResult page(DrainageRiskPageInDTO dto) {
+        return AjaxResult.success(drainageRiskService.queryPage(dto));
+    }
+
+    @Anonymous
+    @GetMapping("/statistics")
+    @ApiOperation("排水管网综合风险全量统计")
+    public AjaxResult statistics(DrainageRiskPageInDTO dto) {
+        return AjaxResult.success(drainageRiskService.statistics(dto));
+    }
+
+    @Anonymous
+    @GetMapping("/filterOptions")
+    @ApiOperation("排水管网综合风险筛选下拉项")
+    public AjaxResult filterOptions() {
+        return AjaxResult.success(drainageRiskService.filterOptions());
+    }
+
+    @Anonymous
+    @GetMapping("/map")
+    @ApiOperation("排水管网风险等级分布图数据")
+    public AjaxResult map(DrainageRiskPageInDTO dto) {
+        return AjaxResult.success(drainageRiskService.mapData(dto));
+    }
+
+    @Anonymous
+    @GetMapping("/{networkId}/detail")
+    @ApiOperation("排水管网综合风险评估详情")
+    public AjaxResult detail(@PathVariable String networkId) {
+        return AjaxResult.success(drainageRiskService.detail(networkId));
+    }
+
+    @PostMapping("/refresh")
+    @ApiOperation("重新计算排水管网风险快照")
+    @Log(title = "排水管网风险评估重算", businessType = BusinessType.OTHER)
+    public AjaxResult refresh() {
+        drainageRiskService.refresh();
+        return AjaxResult.success("已重新计算");
+    }
+}

pipe-network-service/zksy-admin/src/main/java/com/zksy/web/controller/drainage/DrainageWarningController.java

@@ -132,8 +132,19 @@ public class DrainageWarningController {
     @Log(title = "发布预警", businessType = BusinessType.INSERT)
     @Transactional
     public AjaxResult publish(@RequestBody WarningPublishRequest request) {
-        service.publishWarning(request.getWarning(), request.getUserIds());
-        return AjaxResult.success("发布成功");
+        try {
+            service.publishWarning(request.getWarning(), request.getUserIds());
+            return AjaxResult.success("发布成功");
+        } catch (Exception e) {
+            // 直接抛到全局处理器会给前端弹一整屏堆栈，这里转成可读的业务提示，
+            // 表结构缺列时也会明确告诉用户该执行哪个脚本。
+            log.error("发布排水预警失败", e);
+            String message = e.getMessage() == null ? "发布失败" : e.getMessage();
+            if (message.contains("字段不存在") || message.contains("does not exist")) {
+                message += "（请执行 sql/migration_early_warning_columns.sql 补齐 early_warning 表结构）";
+            }
+            return AjaxResult.error("发布失败：" + message);
+        }
     }
 
 

pipe-network-service/zksy-admin/src/main/java/com/zksy/web/controller/drainage/PipeNetworkInfoController.java

@@ -16,11 +16,13 @@ import org.springframework.web.bind.annotation.RestController;
 
 /**
  * 排水管网基础数据查询（旧表 psgw_pipe_network_info）
- * <p>
- * 仅保留分页查询：综合风险区域分析、综合风险统计分析、管网运行情况三个页面
- * 仍依赖本表的 equipment_id（关联设备）与 points（多节点坐标），
- * 这两项在 pipe_network_base 中不存在，故暂不能迁到新接口。
- * <p>
+ *
+ * 仅保留分页查询：管网运行情况页（gwyxqk）仍依赖本表的 equipment_id（关联设备）
+ * 与 points（多节点坐标），这两项在 pipe_network_base 中不存在，故暂不能迁到新接口。
+ *
+ * 综合风险统计分析、综合风险详情分析已改用 {@link DrainageRiskController}
+ * （/api/drainage/risk，数据源 pipe_network_base + equipment_base + alarm_data），不再依赖旧表。
+ *
  * 新增 / 修改 / 删除 / 批量删除 / 列表 / 详情 / 设备绑定等端点已无调用方，已于本次清理移除。
  * 新页面请改用 {@link DrainagePipeNetworkController}（/api/drainage/pipeNetworkBase，数据源 pipe_network_base）。
  */

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

@@ -32,6 +32,19 @@ public interface EquipmentBaseService extends IService<EquipmentBase> {
                                    List<Integer> equipmentTypeIds,
                                    Integer onlineStatus, Integer alarmStatus);
 
+    /**
+     * 设备分页查询（扩展筛选）
+     *
+     * 在基础查询之上补安装位置、所属片区、设备状态三个条件，供运行监测类页面使用。
+     * 参数全部可选，为空即不参与过滤。
+     */
+    Page<EquipmentBaseVO> findByPage(long pageNum, long pageSize,
+                                   String equipmentCode, String equipmentName,
+                                   Integer equipmentTypeId, Integer rootTypeId,
+                                   List<Integer> equipmentTypeIds,
+                                   Integer onlineStatus, Integer alarmStatus,
+                                   String equipmentLocation, Integer district, Integer currentStatus);
+
     /** 移动端日常设备查询：按当前用户公司范围过滤设备类型。 */
     Page<EquipmentBaseVO> findMobileByPage(long pageNum, long pageSize,
                                            String keyword, Integer rootTypeId, Integer onlineStatus,

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

@@ -94,6 +94,17 @@ public class EquipmentBaseServiceImpl extends ServiceImpl<EquipmentBaseMapper, E
                                           Integer equipmentTypeId, Integer rootTypeId,
                                           List<Integer> equipmentTypeIds,
                                           Integer onlineStatus, Integer alarmStatus) {
+        return findByPage(pageNum, pageSize, equipmentCode, equipmentName, equipmentTypeId, rootTypeId,
+                equipmentTypeIds, onlineStatus, alarmStatus, null, null, null);
+    }
+
+    @Override
+    public Page<EquipmentBaseVO> findByPage(long pageNum, long pageSize,
+                                          String equipmentCode, String equipmentName,
+                                          Integer equipmentTypeId, Integer rootTypeId,
+                                          List<Integer> equipmentTypeIds,
+                                          Integer onlineStatus, Integer alarmStatus,
+                                          String equipmentLocation, Integer district, Integer currentStatus) {
         Page<EquipmentBase> page = new Page<>(pageNum, pageSize);
         LambdaQueryWrapper<EquipmentBase> queryWrapper = new LambdaQueryWrapper<>();
         if (equipmentCode != null && !equipmentCode.isEmpty()) {
@@ -111,14 +122,23 @@ public class EquipmentBaseServiceImpl extends ServiceImpl<EquipmentBaseMapper, E
             }
             queryWrapper.in(EquipmentBase::getEquipmentTypeId, equipmentTypeIds);
         }
-        // 在线/报警条件位于 equipment_status，先取出命中的设备ID再回主表过滤
-        if (onlineStatus != null || (alarmStatus != null && alarmStatus == 1)) {
+        if (equipmentLocation != null && !equipmentLocation.isEmpty()) {
+            queryWrapper.like(EquipmentBase::getEquipmentLocation, equipmentLocation);
+        }
+        if (district != null) {
+            queryWrapper.eq(EquipmentBase::getDistrict, district);
+        }
+        // 在线 / 报警 / 设备状态都落在 equipment_status，先取出命中的设备ID再回主表过滤
+        if (onlineStatus != null || alarmStatus != null || currentStatus != null) {
             LambdaQueryWrapper<EquipmentStatus> statusWrapper = new LambdaQueryWrapper<>();
             if (onlineStatus != null) {
                 statusWrapper.eq(EquipmentStatus::getOnlineStatus, onlineStatus);
             }
-            if (alarmStatus != null && alarmStatus == 1) {
-                statusWrapper.eq(EquipmentStatus::getAlarmStatus, 1);
+            if (alarmStatus != null) {
+                statusWrapper.eq(EquipmentStatus::getAlarmStatus, alarmStatus);
+            }
+            if (currentStatus != null) {
+                statusWrapper.eq(EquipmentStatus::getCurrentStatus, currentStatus);
             }
             List<String> matchedIds = equipmentStatusMapper.selectList(statusWrapper).stream()
                     .map(EquipmentStatus::getEquipmentId)

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/domain/DrainageWarning.java

@@ -38,7 +38,7 @@ public class DrainageWarning {
     private String deviceCode;
 
     @ApiModelProperty("设备名称")
-    @TableField("warning_name")
+    @TableField("device_name")
     private String equipmentName;
 
     @ApiModelProperty("预警级别(1严重,2重要,3一般)")
@@ -61,10 +61,19 @@ public class DrainageWarning {
     @ApiModelProperty("预警位置")
     private String location;
 
+    @ApiModelProperty("权属单位")
+    private String ownershipUnit;
+
+    /**
+     * early_warning 表没有坐标列，坐标只在查询时从设备档案补齐后返回给前端定位，
+     * 不能进入 INSERT/UPDATE 语句。
+     */
     @ApiModelProperty("经度")
+    @TableField(exist = false)
     private BigDecimal longitude;
 
     @ApiModelProperty("纬度")
+    @TableField(exist = false)
     private BigDecimal latitude;
 
     @ApiModelProperty("预警内容")
@@ -79,6 +88,9 @@ public class DrainageWarning {
     @ApiModelProperty("发布人")
     private String publisher;
 
+    @ApiModelProperty("创建人")
+    private String createBy;
+
     @ApiModelProperty("发布时间")
     @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
     private LocalDateTime publishTime;

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/domain/WaterloggingPoint.java

@@ -16,8 +16,11 @@ import java.time.LocalDateTime;
 
 /**
  * 易积水点台账
- * <p>一个易积水点在地图上就是一个位置，直接落 WGS84 经纬度；
- * 关联的是视频摄像头（equipment_base.equipment_id），不是通用监测设备。</p>
+ *
+ * 一个易积水点在地图上就是一个位置，直接落 WGS84 经纬度；
+ * 关联的是视频摄像头（equipment_base.equipment_id），不是通用监测设备。
+ * 责任单位 / 责任人通过 duty_dept_id、duty_user_id 关联系统部门与用户，
+ * duty_unit、duty_person、duty_phone 存当时的名称与号码快照。
  */
 @Data
 @AllArgsConstructor
@@ -35,7 +38,7 @@ public class WaterloggingPoint {
     @ApiModelProperty("点位名称")
     private String pointName;
 
-    @ApiModelProperty("所属区域")
+    @ApiModelProperty("所属片区（字典 drainage_district：0城南 1城北 2太常）")
     private String district;
 
     @ApiModelProperty("位置描述")
@@ -67,15 +70,23 @@ public class WaterloggingPoint {
     @TableField("risk_level")
     private String riskLevel;
 
-    @ApiModelProperty("责任单位")
+    @ApiModelProperty("责任单位ID（sys_dept.dept_id）")
+    @TableField("duty_dept_id")
+    private Long dutyDeptId;
+
+    @ApiModelProperty("责任单位（部门名称快照）")
     @TableField("duty_unit")
     private String dutyUnit;
 
-    @ApiModelProperty("责任人")
+    @ApiModelProperty("责任人ID（sys_user.user_id）")
+    @TableField("duty_user_id")
+    private Long dutyUserId;
+
+    @ApiModelProperty("责任人（姓名快照）")
     @TableField("duty_person")
     private String dutyPerson;
 
-    @ApiModelProperty("联系电话")
+    @ApiModelProperty("联系电话（快照，选人时取 sys_user.phonenumber）")
     @TableField("duty_phone")
     private String dutyPhone;
 

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/dto/DrainageRiskItemVO.java

@@ -0,0 +1,91 @@
+package com.zksy.drainage.dto;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+
+/**
+ * 排水管网综合风险列表行
+ *
+ * 一条记录 = 一段排水管网 + 其当前的综合风险评估结果。
+ * 风险评分与等级由 {@code DrainageRiskService} 实时计算，不依赖外部录入。
+ */
+@Data
+@ApiModel(value = "排水管网综合风险列表行")
+public class DrainageRiskItemVO implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @ApiModelProperty("管网设施ID")
+    private String networkId;
+
+    @ApiModelProperty("管线编号")
+    private String networkCode;
+
+    @ApiModelProperty("管网名称")
+    private String networkName;
+
+    @ApiModelProperty("管线类别")
+    private String networkLevel;
+
+    @ApiModelProperty("所在道路 / 位置描述")
+    private String location;
+
+    @ApiModelProperty("所属区域（管养单位）")
+    private String managementUnit;
+
+    @ApiModelProperty("材质")
+    private String material;
+
+    @ApiModelProperty("管径（毫米）")
+    private BigDecimal diameter;
+
+    @ApiModelProperty("长度（米）")
+    private BigDecimal length;
+
+    @ApiModelProperty("管龄（年），按建设日期推算")
+    private Integer pipeAge;
+
+    @ApiModelProperty("管段起点经度")
+    private BigDecimal startLongitude;
+
+    @ApiModelProperty("管段起点纬度")
+    private BigDecimal startLatitude;
+
+    @ApiModelProperty("管段终点经度")
+    private BigDecimal endLongitude;
+
+    @ApiModelProperty("管段终点纬度")
+    private BigDecimal endLatitude;
+
+    @ApiModelProperty("周边关联设备数")
+    private Integer equipmentCount;
+
+    @ApiModelProperty("近30天报警次数")
+    private Integer alarmCount;
+
+    @ApiModelProperty("未处置报警数")
+    private Integer openAlarmCount;
+
+    @ApiModelProperty("周边易积水点数")
+    private Integer waterloggingCount;
+
+    @ApiModelProperty("综合风险评分（0-100）")
+    private Integer riskScore;
+
+    @ApiModelProperty("风险级别编码：high / medium / low")
+    private String riskLevel;
+
+    @ApiModelProperty("风险级别名称")
+    private String riskLevelName;
+
+    @ApiModelProperty("风险级别颜色")
+    private String riskColor;
+
+    @ApiModelProperty("评估时间")
+    private LocalDateTime assessTime;
+}

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/dto/DrainageRiskPageInDTO.java

@@ -0,0 +1,43 @@
+package com.zksy.drainage.dto;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 排水管网综合风险查询入参
+ *
+ * 对应产品要求的五个筛选维度：管线编号、管线类别、所属区域、所在道路、风险级别。
+ */
+@Data
+@ApiModel(value = "排水管网综合风险查询入参")
+public class DrainageRiskPageInDTO implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @ApiModelProperty("管线编号（模糊）")
+    private String networkCode;
+
+    @ApiModelProperty("管网名称（模糊）")
+    private String networkName;
+
+    @ApiModelProperty("管线类别，取 pipe_network_base.network_level")
+    private String networkLevel;
+
+    @ApiModelProperty("所属区域（取管养单位，项目内按区域划分）")
+    private String managementUnit;
+
+    @ApiModelProperty("所在道路（对位置描述模糊匹配）")
+    private String location;
+
+    @ApiModelProperty("风险级别：high / medium / low")
+    private String riskLevel;
+
+    @ApiModelProperty("页码")
+    private Long pageNum;
+
+    @ApiModelProperty("每页条数")
+    private Long pageSize;
+}

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/dto/WaterloggingPointPageInDTO.java

@@ -20,7 +20,7 @@ public class WaterloggingPointPageInDTO implements Serializable {
     @ApiModelProperty(value = "点位名称")
     private String pointName;
 
-    @ApiModelProperty(value = "所属区域")
+    @ApiModelProperty(value = "所属片区（字典 drainage_district：0城南 1城北 2太常，精确匹配）")
     private String district;
 
     @ApiModelProperty(value = "积水程度（0未积水 1轻度 2中度 3严重）")

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/service/DrainageRiskService.java

@@ -0,0 +1,60 @@
+package com.zksy.drainage.service;
+
+import com.zksy.drainage.dto.DrainageRiskItemVO;
+import com.zksy.drainage.dto.DrainageRiskPageInDTO;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 排水管网综合风险评估
+ *
+ * 以 pipe_network_base 中的排水管网为评估对象，结合设备台账、报警、预警、
+ * 易积水点和工程设施，实时计算综合风险评分与等级。
+ */
+public interface DrainageRiskService {
+
+    /**
+     * 分页查询综合风险列表
+     *
+     * @param dto 筛选与分页条件
+     * @return total / records
+     */
+    Map<String, Object> queryPage(DrainageRiskPageInDTO dto);
+
+    /**
+     * 全量统计（不受分页影响，口径与列表筛选一致）
+     *
+     * @param dto 筛选条件（分页参数忽略）
+     * @return 指标、等级分布、区域分布、类别分布
+     */
+    Map<String, Object> statistics(DrainageRiskPageInDTO dto);
+
+    /**
+     * 单条管网综合风险评估详情
+     *
+     * @param networkId 管网设施ID
+     * @return basic / pipeInfo / assessment / analysis 四块
+     */
+    Map<String, Object> detail(String networkId);
+
+    /**
+     * 筛选下拉项：管线类别、所属区域、风险级别
+     *
+     * @return 下拉选项
+     */
+    Map<String, Object> filterOptions();
+
+    /**
+     * 风险等级分布图数据（带起终点坐标，供 GIS 四色图使用）
+     *
+     * @param dto 筛选条件
+     * @return 带坐标的风险管网列表
+     */
+    List<DrainageRiskItemVO> mapData(DrainageRiskPageInDTO dto);
+
+    /**
+     * 立即重算并刷新快照
+     */
+    void refresh();
+}

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/service/impl/DrainageRiskServiceImpl.java

@@ -0,0 +1,1106 @@
+package com.zksy.drainage.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.zksy.base.alarm.domain.AlarmData;
+import com.zksy.base.alarm.mapper.AlarmDataMapper;
+import com.zksy.base.domain.EquipmentBase;
+import com.zksy.base.domain.EquipmentType;
+import com.zksy.base.domain.PipeNetworkBase;
+import com.zksy.base.mapper.EquipmentBaseMapper;
+import com.zksy.base.mapper.EquipmentTypeMapper;
+import com.zksy.base.mapper.PipeNetworkBaseMapper;
+import com.zksy.drainage.domain.DrainageWarning;
+import com.zksy.drainage.domain.EngineeringFacility;
+import com.zksy.drainage.domain.WaterloggingPoint;
+import com.zksy.drainage.dto.DrainageRiskItemVO;
+import com.zksy.drainage.dto.DrainageRiskPageInDTO;
+import com.zksy.drainage.mapper.DrainageWarningMapper;
+import com.zksy.drainage.mapper.EngineeringFacilityMapper;
+import com.zksy.drainage.mapper.WaterloggingPointMapper;
+import com.zksy.drainage.service.DrainageRiskService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.ArrayDeque;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * 排水管网综合风险评估实现
+ *
+ * 评估对象为 pipe_network_base 中 network_type=1 的排水管网，输入数据全部来自本系统已有业务表：
+ * 设备台账（equipment_base）、报警（alarm_data）、预警（early_warning）、
+ * 易积水点（drainage_waterlogging_point）、工程设施（drainage_engineering_facility）。
+ *
+ * 评分模型为 4 个维度合计 100 分：管网本体 35、运行报警 35、环境叠加 20、数据质量 10。
+ * 结果整体走内存快照，避免逐行查库，也保证列表、统计、地图三处口径一致。
+ */
+@Slf4j
+@Service
+public class DrainageRiskServiceImpl implements DrainageRiskService {
+
+    /** 快照有效期，超时后重新计算 */
+    private static final long SNAPSHOT_TTL_MILLIS = 5 * 60 * 1000L;
+
+    /** 报警/预警统计窗口（天） */
+    private static final int RUNNING_WINDOW_DAYS = 30;
+
+    /** 设备视为管网沿线的最大距离（米） */
+    private static final double EQUIPMENT_RADIUS_METERS = 300d;
+
+    /** 易积水点视为管网周边的最大距离（米） */
+    private static final double WATERLOGGING_RADIUS_METERS = 500d;
+
+    /** 泵站、闸门等设施视为管网周边的最大距离（米） */
+    private static final double FACILITY_RADIUS_METERS = 300d;
+
+    private static final String LEVEL_HIGH = "high";
+    private static final String LEVEL_MEDIUM = "medium";
+    private static final String LEVEL_LOW = "low";
+
+    /** 排水设备根类型ID，见 EquipmentTypeEnum.ROOT_DRAINAGE_DEVICE */
+    private static final int DRAINAGE_ROOT_TYPE_ID = 2;
+
+    @Autowired
+    private PipeNetworkBaseMapper pipeNetworkBaseMapper;
+
+    @Autowired
+    private EquipmentBaseMapper equipmentBaseMapper;
+
+    @Autowired
+    private EquipmentTypeMapper equipmentTypeMapper;
+
+    @Autowired
+    private AlarmDataMapper alarmDataMapper;
+
+    @Autowired
+    private DrainageWarningMapper drainageWarningMapper;
+
+    @Autowired
+    private WaterloggingPointMapper waterloggingPointMapper;
+
+    @Autowired
+    private EngineeringFacilityMapper engineeringFacilityMapper;
+
+    private volatile Snapshot snapshot;
+
+    /* ==================== 对外接口 ==================== */
+
+    @Override
+    public Map<String, Object> queryPage(DrainageRiskPageInDTO dto) {
+        long pageNum = dto == null || dto.getPageNum() == null || dto.getPageNum() < 1 ? 1L : dto.getPageNum();
+        long pageSize = dto == null || dto.getPageSize() == null || dto.getPageSize() < 1 ? 10L : dto.getPageSize();
+
+        List<DrainageRiskItemVO> filtered = filter(snapshot().rows, dto);
+        int from = (int) Math.min((pageNum - 1) * pageSize, filtered.size());
+        int to = (int) Math.min(from + pageSize, filtered.size());
+
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("total", filtered.size());
+        result.put("pageNum", pageNum);
+        result.put("pageSize", pageSize);
+        result.put("records", new ArrayList<>(filtered.subList(from, to)));
+        return result;
+    }
+
+    @Override
+    public Map<String, Object> statistics(DrainageRiskPageInDTO dto) {
+        List<DrainageRiskItemVO> rows = filter(snapshot().rows, dto);
+
+        int withAlarm = 0;
+        int totalAlarms = 0;
+        int highRisk = 0;
+        int mediumRisk = 0;
+        int lowRisk = 0;
+        int equipmentTotal = 0;
+        int waterloggingTotal = 0;
+        for (DrainageRiskItemVO row : rows) {
+            int alarmCount = row.getAlarmCount() == null ? 0 : row.getAlarmCount();
+            if (alarmCount > 0) {
+                withAlarm++;
+            }
+            totalAlarms += alarmCount;
+            equipmentTotal += row.getEquipmentCount() == null ? 0 : row.getEquipmentCount();
+            waterloggingTotal += row.getWaterloggingCount() == null ? 0 : row.getWaterloggingCount();
+            if (LEVEL_HIGH.equals(row.getRiskLevel())) {
+                highRisk++;
+            } else if (LEVEL_MEDIUM.equals(row.getRiskLevel())) {
+                mediumRisk++;
+            } else {
+                lowRisk++;
+            }
+        }
+
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("total", rows.size());
+        result.put("withAlarm", withAlarm);
+        result.put("noAlarm", rows.size() - withAlarm);
+        result.put("totalAlarms", totalAlarms);
+        result.put("highRisk", highRisk);
+        result.put("mediumRisk", mediumRisk);
+        result.put("lowRisk", lowRisk);
+        result.put("equipmentTotal", equipmentTotal);
+        result.put("waterloggingTotal", waterloggingTotal);
+        result.put("levelDistribution", levelDistribution(highRisk, mediumRisk, lowRisk));
+        result.put("areaDistribution", distribution(rows, DrainageRiskItemVO::getManagementUnit));
+        result.put("categoryDistribution", distribution(rows, DrainageRiskItemVO::getNetworkLevel));
+        return result;
+    }
+
+    @Override
+    public Map<String, Object> detail(String networkId) {
+        Snapshot current = snapshot();
+        PipeNetworkBase pipe = current.pipes.get(networkId);
+        if (pipe == null) {
+            return null;
+        }
+        Assessment assessment = current.assessments.get(networkId);
+        List<EquipmentBase> matched = matchEquipments(pipe, current.equipments, EQUIPMENT_RADIUS_METERS);
+        List<AlarmData> recentAlarms = loadRecentAlarms(matched, 20);
+
+        Map<String, Object> basic = new LinkedHashMap<>();
+        basic.put("networkId", pipe.getNetworkId());
+        basic.put("networkCode", pipe.getNetworkCode());
+        basic.put("networkName", pipe.getNetworkName());
+        basic.put("networkLevel", pipe.getNetworkLevel());
+        basic.put("managementUnit", pipe.getManagementUnit());
+        basic.put("location", pipe.getLocation());
+        basic.put("status", pipe.getStatus());
+        basic.put("createTime", pipe.getCreateTime());
+        basic.put("updateTime", pipe.getUpdateTime());
+
+        Map<String, Object> pipeInfo = new LinkedHashMap<>();
+        pipeInfo.put("material", pipe.getMaterial());
+        pipeInfo.put("diameter", pipe.getDiameter());
+        pipeInfo.put("length", pipe.getLength());
+        pipeInfo.put("buildDate", pipe.getBuildDate());
+        pipeInfo.put("pipeAge", pipeAge(pipe.getBuildDate()));
+        pipeInfo.put("networkLevel", pipe.getNetworkLevel());
+        pipeInfo.put("startLongitude", pipe.getStartLongitude());
+        pipeInfo.put("startLatitude", pipe.getStartLatitude());
+        pipeInfo.put("endLongitude", pipe.getEndLongitude());
+        pipeInfo.put("endLatitude", pipe.getEndLatitude());
+
+        Map<String, Object> assessBlock = new LinkedHashMap<>();
+        if (assessment != null) {
+            assessBlock.put("riskScore", assessment.total);
+            assessBlock.put("riskLevel", assessment.level);
+            assessBlock.put("riskLevelName", levelName(assessment.level));
+            assessBlock.put("riskColor", levelColor(assessment.level));
+            assessBlock.put("dimensions", assessment.dimensions);
+            assessBlock.put("assessmentModel", "排水管网综合风险模型 v1（本体/报警/环境/数据质量）");
+            assessBlock.put("assessmentTime", current.loadTime);
+        }
+
+        List<Map<String, Object>> equipmentItems = new ArrayList<>();
+        for (EquipmentBase equipment : matched) {
+            Map<String, Object> item = new LinkedHashMap<>();
+            item.put("equipmentId", equipment.getEquipmentId());
+            item.put("equipmentCode", equipment.getEquipmentCode());
+            item.put("equipmentName", equipment.getEquipmentName());
+            item.put("equipmentLocation", equipment.getEquipmentLocation());
+            item.put("longitude", equipment.getLongitude());
+            item.put("latitude", equipment.getLatitude());
+            equipmentItems.add(item);
+        }
+        assessBlock.put("equipmentCount", matched.size());
+        assessBlock.put("alarmCount", assessment == null ? 0 : assessment.alarmCount);
+        assessBlock.put("openAlarmCount", assessment == null ? 0 : assessment.openAlarmCount);
+        assessBlock.put("waterloggingCount", assessment == null ? 0 : assessment.waterloggingCount);
+        assessBlock.put("equipments", equipmentItems);
+
+        Map<String, Object> analysis = new LinkedHashMap<>();
+        analysis.put("riskFactors", assessment == null ? Collections.emptyList() : assessment.factors);
+        analysis.put("mitigationMeasures", assessment == null ? Collections.emptyList() : assessment.measures);
+        analysis.put("protectiveTargets", assessment == null ? null : assessment.protectiveTargets);
+        analysis.put("hazardSources", assessment == null ? null : assessment.hazardSources);
+        analysis.put("recentAlarms", recentAlarms);
+
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("basic", basic);
+        result.put("pipeInfo", pipeInfo);
+        result.put("assessment", assessBlock);
+        result.put("analysis", analysis);
+        return result;
+    }
+
+    @Override
+    public Map<String, Object> filterOptions() {
+        Snapshot current = snapshot();
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("networkLevels", distinct(current.rows, DrainageRiskItemVO::getNetworkLevel));
+        result.put("managementUnits", distinct(current.rows, DrainageRiskItemVO::getManagementUnit));
+        List<Map<String, Object>> riskLevels = new ArrayList<>();
+        riskLevels.add(option(LEVEL_HIGH, "高风险"));
+        riskLevels.add(option(LEVEL_MEDIUM, "中风险"));
+        riskLevels.add(option(LEVEL_LOW, "低风险"));
+        result.put("riskLevels", riskLevels);
+        return result;
+    }
+
+    @Override
+    public List<DrainageRiskItemVO> mapData(DrainageRiskPageInDTO dto) {
+        return filter(snapshot().rows, dto).stream()
+                .filter(row -> hasCoordinate(row.getStartLongitude(), row.getStartLatitude())
+                        || hasCoordinate(row.getEndLongitude(), row.getEndLatitude()))
+                .collect(Collectors.toList());
+    }
+
+    @Override
+    public void refresh() {
+        snapshot = buildSnapshot();
+    }
+
+    /* ==================== 快照 ==================== */
+
+    private Snapshot snapshot() {
+        Snapshot current = snapshot;
+        if (current != null && System.currentTimeMillis() - current.loadTime < SNAPSHOT_TTL_MILLIS) {
+            return current;
+        }
+        synchronized (this) {
+            if (snapshot != null && System.currentTimeMillis() - snapshot.loadTime < SNAPSHOT_TTL_MILLIS) {
+                return snapshot;
+            }
+            snapshot = buildSnapshot();
+            return snapshot;
+        }
+    }
+
+    private Snapshot buildSnapshot() {
+        long started = System.currentTimeMillis();
+        List<PipeNetworkBase> pipes = loadDrainagePipes();
+        if (pipes.isEmpty()) {
+            log.info("排水管网综合风险评估：未找到排水管网数据");
+            return new Snapshot(Collections.emptyList(), Collections.emptyMap(), Collections.emptyMap(),
+                    Collections.emptyList(), started);
+        }
+
+        List<EquipmentBase> equipments = loadDrainageEquipments();
+        // 报警按设备编码、预警按设备ID聚合，先圈定排水设备的标识集合，避免燃气/供水数据混入
+        Set<String> drainageEquipmentCodes = equipments.stream()
+                .map(EquipmentBase::getEquipmentCode)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toSet());
+        Set<String> drainageEquipmentIds = equipments.stream()
+                .map(EquipmentBase::getEquipmentId)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toSet());
+        Map<String, long[]> alarmStat = loadAlarmStat(drainageEquipmentCodes);
+        Map<String, Long> warningStat = loadWarningStat(drainageEquipmentIds);
+        List<WaterloggingPoint> waterloggingPoints = loadWaterloggingPoints();
+        List<EngineeringFacility> facilities = loadFacilities();
+
+        log.info("排水管网综合风险评估开始：管网 {} 条，设备 {} 台，报警设备 {} 个，易积水点 {} 个，设施 {} 个",
+                pipes.size(), equipments.size(), alarmStat.size(), waterloggingPoints.size(), facilities.size());
+
+        List<DrainageRiskItemVO> rows = new ArrayList<>(pipes.size());
+        Map<String, PipeNetworkBase> pipeMap = new LinkedHashMap<>();
+        Map<String, Assessment> assessmentMap = new HashMap<>();
+
+        for (PipeNetworkBase pipe : pipes) {
+            List<EquipmentBase> matched = matchEquipments(pipe, equipments, EQUIPMENT_RADIUS_METERS);
+
+            int alarmCount = 0;
+            int openAlarmCount = 0;
+            int highAlarmCount = 0;
+            long openWarningCount = 0;
+            for (EquipmentBase equipment : matched) {
+                long[] stat = alarmStat.get(equipment.getEquipmentCode());
+                if (stat != null) {
+                    alarmCount += (int) stat[0];
+                    openAlarmCount += (int) stat[1];
+                    highAlarmCount += (int) stat[2];
+                }
+                Long warning = warningStat.get(equipment.getEquipmentId());
+                if (warning != null) {
+                    openWarningCount += warning;
+                }
+            }
+
+            int waterloggingCount = countPointsWithin(pipe, waterloggingPoints, WATERLOGGING_RADIUS_METERS);
+            int facilityCount = countFacilitiesWithin(pipe, facilities, FACILITY_RADIUS_METERS);
+
+            Assessment assessment = assess(pipe, matched.size(), alarmCount, openAlarmCount,
+                    highAlarmCount, (int) openWarningCount, waterloggingCount, facilityCount);
+
+            DrainageRiskItemVO row = toRow(pipe, matched.size(), alarmCount, openAlarmCount,
+                    waterloggingCount, assessment);
+            rows.add(row);
+            pipeMap.put(pipe.getNetworkId(), pipe);
+            assessmentMap.put(pipe.getNetworkId(), assessment);
+        }
+
+        // 高风险优先，其次按评分倒序，便于首页就能看到最需要关注的管网
+        rows.sort((a, b) -> {
+            int byScore = Integer.compare(b.getRiskScore() == null ? 0 : b.getRiskScore(),
+                    a.getRiskScore() == null ? 0 : a.getRiskScore());
+            if (byScore != 0) {
+                return byScore;
+            }
+            return String.valueOf(a.getNetworkCode()).compareTo(String.valueOf(b.getNetworkCode()));
+        });
+
+        log.info("排水管网综合风险评估完成：耗时 {} ms", System.currentTimeMillis() - started);
+        return new Snapshot(rows, pipeMap, assessmentMap, equipments, started);
+    }
+
+    /* ==================== 数据加载 ==================== */
+
+    private List<PipeNetworkBase> loadDrainagePipes() {
+        LambdaQueryWrapper<PipeNetworkBase> wrapper = new LambdaQueryWrapper<>();
+        wrapper.eq(PipeNetworkBase::getNetworkType, PipeNetworkBase.NETWORK_TYPE_DRAINAGE);
+        return pipeNetworkBaseMapper.selectList(wrapper);
+    }
+
+    /**
+     * 加载排水设备（含排水根类型下所有子类型），只保留带坐标的
+     */
+    private List<EquipmentBase> loadDrainageEquipments() {
+        Set<Integer> typeIds = drainageTypeIds();
+        if (typeIds.isEmpty()) {
+            return Collections.emptyList();
+        }
+        LambdaQueryWrapper<EquipmentBase> wrapper = new LambdaQueryWrapper<>();
+        wrapper.in(EquipmentBase::getEquipmentTypeId, typeIds);
+        wrapper.isNotNull(EquipmentBase::getLongitude).isNotNull(EquipmentBase::getLatitude);
+        return equipmentBaseMapper.selectList(wrapper);
+    }
+
+    /**
+     * 解析排水设备根类型下的全部类型ID
+     */
+    private Set<Integer> drainageTypeIds() {
+        Set<Integer> result = new LinkedHashSet<>();
+        try {
+            List<EquipmentType> types = equipmentTypeMapper.selectList(null);
+            Map<Integer, List<Integer>> children = new HashMap<>();
+            for (EquipmentType type : types) {
+                if (type.getParentId() != null) {
+                    children.computeIfAbsent(type.getParentId(), key -> new ArrayList<>()).add(type.getId());
+                }
+            }
+            Deque<Integer> queue = new ArrayDeque<>();
+            queue.add(DRAINAGE_ROOT_TYPE_ID);
+            while (!queue.isEmpty()) {
+                Integer id = queue.poll();
+                if (!result.add(id)) {
+                    continue;
+                }
+                List<Integer> sub = children.get(id);
+                if (sub != null) {
+                    queue.addAll(sub);
+                }
+            }
+        } catch (Exception e) {
+            log.warn("解析排水设备类型失败，退化为仅根类型", e);
+            result.add(DRAINAGE_ROOT_TYPE_ID);
+        }
+        return result;
+    }
+
+    /**
+     * 近30天报警按设备编码聚合：[次数, 未处置数, 高等级数]
+     *
+     * 只统计传入的排水设备编码，燃气与供水设备不进统计。
+     */
+    private Map<String, long[]> loadAlarmStat(Set<String> drainageEquipmentCodes) {
+        Map<String, long[]> result = new HashMap<>();
+        if (drainageEquipmentCodes.isEmpty()) {
+            return result;
+        }
+        try {
+            QueryWrapper<AlarmData> wrapper = new QueryWrapper<>();
+            wrapper.select("device_code as device_code",
+                            "count(*) as total_cnt",
+                            "sum(case when alarm_status = 0 then 1 else 0 end) as open_cnt",
+                            "sum(case when alarm_level <= 2 then 1 else 0 end) as high_cnt")
+                    .ge("create_time", LocalDateTime.now().minusDays(RUNNING_WINDOW_DAYS))
+                    .in("device_code", drainageEquipmentCodes)
+                    .groupBy("device_code");
+            List<Map<String, Object>> rows = alarmDataMapper.selectMaps(wrapper);
+            for (Map<String, Object> row : rows) {
+                String deviceCode = str(row, "device_code");
+                if (deviceCode == null) {
+                    continue;
+                }
+                result.put(deviceCode, new long[]{num(row, "total_cnt"), num(row, "open_cnt"), num(row, "high_cnt")});
+            }
+        } catch (Exception e) {
+            log.warn("统计排水管网报警数据失败", e);
+        }
+        return result;
+    }
+
+    /**
+     * 近30天未闭环预警按设备ID聚合
+     *
+     * 只统计传入的排水设备ID，early_warning 中其他业务线的预警不进统计。
+     */
+    private Map<String, Long> loadWarningStat(Set<String> drainageEquipmentIds) {
+        Map<String, Long> result = new HashMap<>();
+        if (drainageEquipmentIds.isEmpty()) {
+            return result;
+        }
+        try {
+            QueryWrapper<DrainageWarning> wrapper = new QueryWrapper<>();
+            wrapper.select("equipment_id as equipment_id", "count(*) as total_cnt")
+                    .ge("create_time", LocalDateTime.now().minusDays(RUNNING_WINDOW_DAYS))
+                    .in("equipment_id", drainageEquipmentIds)
+                    .and(w -> w.isNull("status").or().ne("status", "CLOSED"))
+                    .groupBy("equipment_id");
+            List<Map<String, Object>> rows = drainageWarningMapper.selectMaps(wrapper);
+            for (Map<String, Object> row : rows) {
+                String equipmentId = str(row, "equipment_id");
+                if (equipmentId != null) {
+                    result.put(equipmentId, num(row, "total_cnt"));
+                }
+            }
+        } catch (Exception e) {
+            log.warn("统计排水管网预警数据失败", e);
+        }
+        return result;
+    }
+
+    private List<WaterloggingPoint> loadWaterloggingPoints() {
+        try {
+            LambdaQueryWrapper<WaterloggingPoint> wrapper = new LambdaQueryWrapper<>();
+            wrapper.isNotNull(WaterloggingPoint::getLongitude).isNotNull(WaterloggingPoint::getLatitude);
+            return waterloggingPointMapper.selectList(wrapper);
+        } catch (Exception e) {
+            log.warn("加载易积水点失败", e);
+            return Collections.emptyList();
+        }
+    }
+
+    private List<EngineeringFacility> loadFacilities() {
+        try {
+            LambdaQueryWrapper<EngineeringFacility> wrapper = new LambdaQueryWrapper<>();
+            wrapper.isNotNull(EngineeringFacility::getLongitude).isNotNull(EngineeringFacility::getLatitude);
+            return engineeringFacilityMapper.selectList(wrapper);
+        } catch (Exception e) {
+            log.warn("加载工程设施失败", e);
+            return Collections.emptyList();
+        }
+    }
+
+    /**
+     * 取管网周边指定距离内的设备
+     */
+    private List<EquipmentBase> matchEquipments(PipeNetworkBase pipe, List<EquipmentBase> equipments, double radius) {
+        if (equipments.isEmpty() || !hasAnyCoordinate(pipe)) {
+            return Collections.emptyList();
+        }
+        List<EquipmentBase> matched = new ArrayList<>();
+        for (EquipmentBase equipment : equipments) {
+            double lng = equipment.getLongitude().doubleValue();
+            double lat = equipment.getLatitude().doubleValue();
+            if (distanceToPipe(pipe, lng, lat) <= radius) {
+                matched.add(equipment);
+            }
+        }
+        return matched;
+    }
+
+    private int countPointsWithin(PipeNetworkBase pipe, List<WaterloggingPoint> points, double radius) {
+        if (points.isEmpty() || !hasAnyCoordinate(pipe)) {
+            return 0;
+        }
+        int count = 0;
+        for (WaterloggingPoint point : points) {
+            if (distanceToPipe(pipe, point.getLongitude().doubleValue(), point.getLatitude().doubleValue()) <= radius) {
+                count++;
+            }
+        }
+        return count;
+    }
+
+    private int countFacilitiesWithin(PipeNetworkBase pipe, List<EngineeringFacility> facilities, double radius) {
+        if (facilities.isEmpty() || !hasAnyCoordinate(pipe)) {
+            return 0;
+        }
+        int count = 0;
+        for (EngineeringFacility facility : facilities) {
+            if (distanceToPipe(pipe, facility.getLongitude().doubleValue(), facility.getLatitude().doubleValue()) <= radius) {
+                count++;
+            }
+        }
+        return count;
+    }
+
+    private List<AlarmData> loadRecentAlarms(List<EquipmentBase> equipments, int limit) {
+        if (equipments.isEmpty()) {
+            return Collections.emptyList();
+        }
+        List<String> codes = equipments.stream()
+                .map(EquipmentBase::getEquipmentCode)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toList());
+        if (codes.isEmpty()) {
+            return Collections.emptyList();
+        }
+        try {
+            LambdaQueryWrapper<AlarmData> wrapper = new LambdaQueryWrapper<>();
+            wrapper.in(AlarmData::getDeviceCode, codes);
+            wrapper.orderByDesc(AlarmData::getCreateTime);
+            wrapper.last("limit " + Math.max(1, limit));
+            return alarmDataMapper.selectList(wrapper);
+        } catch (Exception e) {
+            log.warn("加载管网关联报警失败", e);
+            return Collections.emptyList();
+        }
+    }
+
+    /* ==================== 评分模型 ==================== */
+
+    /**
+     * 四维评分：管网本体 35 + 运行报警 35 + 环境叠加 20 + 数据质量 10
+     */
+    private Assessment assess(PipeNetworkBase pipe, int equipmentCount, int alarmCount, int openAlarmCount,
+                              int highAlarmCount, int openWarningCount, int waterloggingCount, int facilityCount) {
+        int age = pipeAge(pipe.getBuildDate());
+
+        int ageScore = age <= 5 ? 3 : age <= 10 ? 8 : age <= 15 ? 12 : age <= 20 ? 14 : 15;
+        int materialScore = materialScore(pipe.getMaterial());
+        int diameterScore = diameterScore(pipe.getDiameter());
+        int levelScore = levelScore(pipe.getNetworkLevel());
+        int pipeScore = ageScore + materialScore + diameterScore + levelScore;
+
+        int alarmScore = alarmCount <= 0 ? 0 : alarmCount <= 2 ? 8 : alarmCount <= 5 ? 12 : 15;
+        int openScore = openAlarmCount <= 0 ? 0 : openAlarmCount == 1 ? 5 : openAlarmCount == 2 ? 8 : 10;
+        int highScore = highAlarmCount <= 0 ? 0 : highAlarmCount == 1 ? 5 : 10;
+        int runningScore = alarmScore + openScore + highScore;
+
+        int waterScore = waterloggingCount <= 0 ? 0 : waterloggingCount == 1 ? 5 : waterloggingCount == 2 ? 7 : 10;
+        int warningScore = openWarningCount <= 0 ? 0 : openWarningCount == 1 ? 4 : 6;
+        int facilityScore = facilityCount > 0 ? 4 : 0;
+        int environmentScore = waterScore + warningScore + facilityScore;
+
+        int missingCoordinate = hasAnyCoordinate(pipe) ? 0 : 4;
+        int missingMaterial = isBlank(pipe.getMaterial()) ? 3 : 0;
+        int missingDimension = isBlank(pipe.getLength() == null ? null : pipe.getLength().toPlainString())
+                || pipe.getDiameter() == null ? 3 : 0;
+        int qualityScore = missingCoordinate + missingMaterial + missingDimension;
+
+        int total = Math.min(100, pipeScore + runningScore + environmentScore + qualityScore);
+        String level = total >= 70 ? LEVEL_HIGH : total >= 40 ? LEVEL_MEDIUM : LEVEL_LOW;
+
+        Assessment assessment = new Assessment();
+        assessment.total = total;
+        assessment.level = level;
+        assessment.alarmCount = alarmCount;
+        assessment.openAlarmCount = openAlarmCount;
+        assessment.waterloggingCount = waterloggingCount;
+        assessment.dimensions = dimensions(pipeScore, ageScore, materialScore, diameterScore, levelScore,
+                runningScore, alarmScore, openScore, highScore,
+                environmentScore, waterScore, warningScore, facilityScore,
+                qualityScore, missingCoordinate, missingMaterial, missingDimension,
+                age, pipe.getMaterial(), pipe.getDiameter(), waterloggingCount, facilityCount);
+        assessment.factors = factors(pipe, age, alarmCount, openAlarmCount, highAlarmCount,
+                waterloggingCount, facilityCount);
+        assessment.measures = measures(level, age, alarmCount, openAlarmCount, waterloggingCount);
+        assessment.protectiveTargets = protectiveTargets(waterloggingCount, facilityCount);
+        assessment.hazardSources = hazardSources(equipmentCount, alarmCount, openAlarmCount);
+        return assessment;
+    }
+
+    private List<Map<String, Object>> dimensions(int pipeScore, int ageScore, int materialScore, int diameterScore,
+                                                 int levelScore, int runningScore, int alarmScore, int openScore,
+                                                 int highScore, int environmentScore, int waterScore, int warningScore,
+                                                 int facilityScore, int qualityScore, int missingCoordinate,
+                                                 int missingMaterial, int missingDimension, int age, String material,
+                                                 BigDecimal diameter, int waterloggingCount, int facilityCount) {
+        List<Map<String, Object>> dimensions = new ArrayList<>();
+
+        Map<String, Object> pipeDim = dimension("pipe", "管网本体", pipeScore, 35);
+        addItem(pipeDim, "管龄", ageScore, age <= 0 ? "未登记建设日期" : "已运行 " + age + " 年");
+        addItem(pipeDim, "材质", materialScore, isBlank(material) ? "未登记材质" : material);
+        addItem(pipeDim, "管径", diameterScore, diameter == null ? "未登记管径" : diameter.stripTrailingZeros().toPlainString() + " mm");
+        addItem(pipeDim, "管线类别", levelScore, "按类别影响范围计分");
+        dimensions.add(pipeDim);
+
+        Map<String, Object> runningDim = dimension("running", "运行报警", runningScore, 35);
+        addItem(runningDim, "近30天报警", alarmScore, alarmScore + " 分");
+        addItem(runningDim, "未处置报警", openScore, openScore + " 分");
+        addItem(runningDim, "高等级报警", highScore, highScore + " 分");
+        dimensions.add(runningDim);
+
+        Map<String, Object> envDim = dimension("environment", "环境叠加", environmentScore, 20);
+        addItem(envDim, "周边易积水点", waterScore, waterloggingCount + " 处");
+        addItem(envDim, "周边未闭环预警", warningScore, "按关联设备预警计分");
+        addItem(envDim, "周边工程设施", facilityScore, facilityCount + " 处");
+        dimensions.add(envDim);
+
+        Map<String, Object> qualityDim = dimension("quality", "数据质量", qualityScore, 10);
+        addItem(qualityDim, "坐标完整性", missingCoordinate, missingCoordinate > 0 ? "缺起终点坐标" : "正常");
+        addItem(qualityDim, "材质完整性", missingMaterial, missingMaterial > 0 ? "缺材质" : "正常");
+        addItem(qualityDim, "规格完整性", missingDimension, missingDimension > 0 ? "缺管径或长度" : "正常");
+        dimensions.add(qualityDim);
+
+        return dimensions;
+    }
+
+    private Map<String, Object> dimension(String key, String label, int score, int max) {
+        Map<String, Object> dim = new LinkedHashMap<>();
+        dim.put("key", key);
+        dim.put("label", label);
+        dim.put("score", score);
+        dim.put("max", max);
+        dim.put("items", new ArrayList<Map<String, Object>>());
+        return dim;
+    }
+
+    @SuppressWarnings("unchecked")
+    private void addItem(Map<String, Object> dimension, String label, int score, String detail) {
+        Map<String, Object> item = new LinkedHashMap<>();
+        item.put("label", label);
+        item.put("score", score);
+        item.put("detail", detail);
+        ((List<Map<String, Object>>) dimension.get("items")).add(item);
+    }
+
+    /**
+     * 风险构成要素，直接给出可读原因
+     */
+    private List<String> factors(PipeNetworkBase pipe, int age, int alarmCount, int openAlarmCount,
+                                 int highAlarmCount, int waterloggingCount, int facilityCount) {
+        List<String> factors = new ArrayList<>();
+        if (age > 20) {
+            factors.add("管网已运行 " + age + " 年，超过设计使用年限，结构性缺陷概率高");
+        } else if (age > 10) {
+            factors.add("管网已运行 " + age + " 年，进入老化加速期");
+        }
+        if (isBlank(pipe.getMaterial())) {
+            factors.add("未登记管材，无法判断材质风险");
+        } else if (pipe.getMaterial().contains("混凝土") || pipe.getMaterial().contains("砼") || pipe.getMaterial().contains("水泥")) {
+            factors.add("管材为「" + pipe.getMaterial() + "」，存在腐蚀与接口渗漏风险");
+        }
+        if (pipe.getDiameter() != null && pipe.getDiameter().doubleValue() < 300) {
+            factors.add("管径偏小（" + pipe.getDiameter().stripTrailingZeros().toPlainString() + " mm），易淤积堵塞");
+        }
+        if (alarmCount > 0) {
+            factors.add("近 " + RUNNING_WINDOW_DAYS + " 天关联设备产生 " + alarmCount + " 次报警");
+        }
+        if (openAlarmCount > 0) {
+            factors.add("存在 " + openAlarmCount + " 条未处置报警");
+        }
+        if (highAlarmCount > 0) {
+            factors.add("存在 " + highAlarmCount + " 条高等级报警");
+        }
+        if (waterloggingCount > 0) {
+            factors.add("周边 " + (int) WATERLOGGING_RADIUS_METERS + " 米内存在 " + waterloggingCount + " 处易积水点");
+        }
+        if (facilityCount > 0) {
+            factors.add("周边 " + (int) FACILITY_RADIUS_METERS + " 米内存在 " + facilityCount + " 处泵站/闸门等设施");
+        }
+        if (!hasAnyCoordinate(pipe)) {
+            factors.add("缺少起终点坐标，未能完成空间关联，实际风险可能被低估");
+        }
+        if (factors.isEmpty()) {
+            factors.add("管网本体状态良好，近 " + RUNNING_WINDOW_DAYS + " 天无关联报警");
+        }
+        return factors;
+    }
+
+    /**
+     * 按等级给出可执行的处置建议
+     */
+    private List<String> measures(String level, int age, int alarmCount, int openAlarmCount, int waterloggingCount) {
+        List<String> measures = new ArrayList<>();
+        if (LEVEL_HIGH.equals(level)) {
+            measures.add("列入重点管控清单，巡检频次提高至每周不少于 1 次");
+            measures.add("安排管道内窥检测（CCTV/声呐），核实结构性缺陷");
+        } else if (LEVEL_MEDIUM.equals(level)) {
+            measures.add("纳入常规巡检计划，巡检频次不低于每两周 1 次");
+        } else {
+            measures.add("维持现有巡检频次，按季度复核风险等级");
+        }
+        if (openAlarmCount > 0) {
+            measures.add("优先处置 " + openAlarmCount + " 条未闭环报警，完成现场核实");
+        }
+        if (alarmCount > 0) {
+            measures.add("核查报警设备阈值配置，排除误报并同步维护设备");
+        }
+        if (waterloggingCount > 0) {
+            measures.add("汛期前疏通该管段并联动周边易积水点排水设施");
+        }
+        if (age > 20) {
+            measures.add("纳入老旧管网改造计划，评估翻建或非开挖修复方案");
+        }
+        return measures;
+    }
+
+    private String protectiveTargets(int waterloggingCount, int facilityCount) {
+        if (waterloggingCount == 0 && facilityCount == 0) {
+            return null;
+        }
+        List<String> parts = new ArrayList<>();
+        if (waterloggingCount > 0) {
+            parts.add("周边易积水点 " + waterloggingCount + " 处");
+        }
+        if (facilityCount > 0) {
+            parts.add("周边泵站/闸门等设施 " + facilityCount + " 处");
+        }
+        return String.join("；", parts);
+    }
+
+    private String hazardSources(int equipmentCount, int alarmCount, int openAlarmCount) {
+        if (equipmentCount == 0) {
+            return "未关联监测设备";
+        }
+        return "关联监测设备 " + equipmentCount + " 台，近 " + RUNNING_WINDOW_DAYS + " 天报警 "
+                + alarmCount + " 次，其中未处置 " + openAlarmCount + " 条";
+    }
+
+    private int materialScore(String material) {
+        if (isBlank(material)) {
+            return 5;
+        }
+        String value = material.trim();
+        if (value.contains("混凝土") || value.contains("砼") || value.contains("水泥")) {
+            return 8;
+        }
+        if (value.contains("球墨")) {
+            return 4;
+        }
+        if (value.contains("铸铁") || value.contains("铁")) {
+            return 6;
+        }
+        if (value.contains("钢")) {
+            return 5;
+        }
+        if (value.contains("PVC") || value.contains("PE") || value.contains("塑料") || value.contains("HDPE")) {
+            return 2;
+        }
+        return 4;
+    }
+
+    private int diameterScore(BigDecimal diameter) {
+        if (diameter == null) {
+            return 7;
+        }
+        double value = diameter.doubleValue();
+        if (value < 300) {
+            return 7;
+        }
+        if (value < 600) {
+            return 4;
+        }
+        if (value < 1000) {
+            return 2;
+        }
+        return 1;
+    }
+
+    private int levelScore(String networkLevel) {
+        if (isBlank(networkLevel)) {
+            return 2;
+        }
+        String value = networkLevel.trim();
+        if (value.contains("主干") || value.contains("高") || "high".equalsIgnoreCase(value)) {
+            return 5;
+        }
+        if (value.contains("次干") || value.contains("中") || "medium".equalsIgnoreCase(value)) {
+            return 3;
+        }
+        if (value.contains("支") || value.contains("低") || "low".equalsIgnoreCase(value)) {
+            return 1;
+        }
+        return 2;
+    }
+
+    /* ==================== 空间计算 ==================== */
+
+    private boolean hasAnyCoordinate(PipeNetworkBase pipe) {
+        return hasCoordinate(pipe.getStartLongitude(), pipe.getStartLatitude())
+                || hasCoordinate(pipe.getEndLongitude(), pipe.getEndLatitude());
+    }
+
+    private boolean hasCoordinate(BigDecimal lng, BigDecimal lat) {
+        return lng != null && lat != null;
+    }
+
+    /**
+     * 点到管段的最短距离（米）
+     *
+     * 城市尺度下把经纬度按当地纬度折算成米做平面近似，精度足以支撑 300/500 米的关联判定。
+     */
+    private double distanceToPipe(PipeNetworkBase pipe, double lng, double lat) {
+        boolean hasStart = hasCoordinate(pipe.getStartLongitude(), pipe.getStartLatitude());
+        boolean hasEnd = hasCoordinate(pipe.getEndLongitude(), pipe.getEndLatitude());
+        if (hasStart && hasEnd) {
+            return distancePointToSegment(lng, lat,
+                    pipe.getStartLongitude().doubleValue(), pipe.getStartLatitude().doubleValue(),
+                    pipe.getEndLongitude().doubleValue(), pipe.getEndLatitude().doubleValue());
+        }
+        if (hasStart) {
+            return distancePointToSegment(lng, lat,
+                    pipe.getStartLongitude().doubleValue(), pipe.getStartLatitude().doubleValue(),
+                    pipe.getStartLongitude().doubleValue(), pipe.getStartLatitude().doubleValue());
+        }
+        if (hasEnd) {
+            return distancePointToSegment(lng, lat,
+                    pipe.getEndLongitude().doubleValue(), pipe.getEndLatitude().doubleValue(),
+                    pipe.getEndLongitude().doubleValue(), pipe.getEndLatitude().doubleValue());
+        }
+        return Double.MAX_VALUE;
+    }
+
+    private double distancePointToSegment(double px, double py, double ax, double ay, double bx, double by) {
+        double meterPerLat = 110540d;
+        double meterPerLng = 111320d * Math.cos(Math.toRadians((ay + by) / 2));
+        // 以目标点为原点建立局部平面坐标
+        double x1 = (ax - px) * meterPerLng;
+        double y1 = (ay - py) * meterPerLat;
+        double x2 = (bx - px) * meterPerLng;
+        double y2 = (by - py) * meterPerLat;
+        double dx = x2 - x1;
+        double dy = y2 - y1;
+        double lengthSquared = dx * dx + dy * dy;
+        double t = lengthSquared == 0 ? 0 : -(x1 * dx + y1 * dy) / lengthSquared;
+        t = Math.max(0, Math.min(1, t));
+        return Math.hypot(x1 + t * dx, y1 + t * dy);
+    }
+
+    /* ==================== 组装与工具 ==================== */
+
+    private DrainageRiskItemVO toRow(PipeNetworkBase pipe, int equipmentCount, int alarmCount,
+                                      int openAlarmCount, int waterloggingCount, Assessment assessment) {
+        DrainageRiskItemVO row = new DrainageRiskItemVO();
+        row.setNetworkId(pipe.getNetworkId());
+        row.setNetworkCode(pipe.getNetworkCode());
+        row.setNetworkName(pipe.getNetworkName());
+        row.setNetworkLevel(pipe.getNetworkLevel());
+        row.setLocation(pipe.getLocation());
+        row.setManagementUnit(pipe.getManagementUnit());
+        row.setMaterial(pipe.getMaterial());
+        row.setDiameter(pipe.getDiameter());
+        row.setLength(pipe.getLength());
+        row.setPipeAge(pipeAge(pipe.getBuildDate()));
+        row.setStartLongitude(pipe.getStartLongitude());
+        row.setStartLatitude(pipe.getStartLatitude());
+        row.setEndLongitude(pipe.getEndLongitude());
+        row.setEndLatitude(pipe.getEndLatitude());
+        row.setEquipmentCount(equipmentCount);
+        row.setAlarmCount(alarmCount);
+        row.setOpenAlarmCount(openAlarmCount);
+        row.setWaterloggingCount(waterloggingCount);
+        row.setRiskScore(assessment.total);
+        row.setRiskLevel(assessment.level);
+        row.setRiskLevelName(levelName(assessment.level));
+        row.setRiskColor(levelColor(assessment.level));
+        row.setAssessTime(LocalDateTime.now());
+        return row;
+    }
+
+    private List<DrainageRiskItemVO> filter(List<DrainageRiskItemVO> rows, DrainageRiskPageInDTO dto) {
+        if (dto == null) {
+            return new ArrayList<>(rows);
+        }
+        String networkCode = trim(dto.getNetworkCode());
+        String networkName = trim(dto.getNetworkName());
+        String networkLevel = trim(dto.getNetworkLevel());
+        String managementUnit = trim(dto.getManagementUnit());
+        String location = trim(dto.getLocation());
+        String riskLevel = trim(dto.getRiskLevel());
+
+        List<DrainageRiskItemVO> result = new ArrayList<>();
+        for (DrainageRiskItemVO row : rows) {
+            if (networkCode != null && !containsIgnoreCase(row.getNetworkCode(), networkCode)) {
+                continue;
+            }
+            if (networkName != null && !containsIgnoreCase(row.getNetworkName(), networkName)) {
+                continue;
+            }
+            if (networkLevel != null && !Objects.equals(row.getNetworkLevel(), networkLevel)) {
+                continue;
+            }
+            if (managementUnit != null && !Objects.equals(row.getManagementUnit(), managementUnit)) {
+                continue;
+            }
+            if (location != null && !containsIgnoreCase(row.getLocation(), location)) {
+                continue;
+            }
+            if (riskLevel != null && !Objects.equals(row.getRiskLevel(), riskLevel)) {
+                continue;
+            }
+            result.add(row);
+        }
+        return result;
+    }
+
+    private List<Map<String, Object>> levelDistribution(int high, int medium, int low) {
+        List<Map<String, Object>> list = new ArrayList<>();
+        list.add(distributionItem("高风险", high, levelColor(LEVEL_HIGH)));
+        list.add(distributionItem("中风险", medium, levelColor(LEVEL_MEDIUM)));
+        list.add(distributionItem("低风险", low, levelColor(LEVEL_LOW)));
+        return list;
+    }
+
+    private Map<String, Object> distributionItem(String name, int value, String color) {
+        Map<String, Object> item = new LinkedHashMap<>();
+        item.put("name", name);
+        item.put("value", value);
+        item.put("color", color);
+        return item;
+    }
+
+    private List<Map<String, Object>> distribution(List<DrainageRiskItemVO> rows,
+                                                   java.util.function.Function<DrainageRiskItemVO, String> getter) {
+        Map<String, Integer> counter = new LinkedHashMap<>();
+        for (DrainageRiskItemVO row : rows) {
+            String key = getter.apply(row);
+            if (isBlank(key)) {
+                key = "未填写";
+            }
+            counter.merge(key, 1, Integer::sum);
+        }
+        List<Map<String, Object>> list = new ArrayList<>();
+        counter.forEach((name, value) -> {
+            Map<String, Object> item = new LinkedHashMap<>();
+            item.put("name", name);
+            item.put("value", value);
+            list.add(item);
+        });
+        list.sort((a, b) -> Integer.compare(((Number) b.get("value")).intValue(),
+                ((Number) a.get("value")).intValue()));
+        return list;
+    }
+
+    private List<String> distinct(List<DrainageRiskItemVO> rows,
+                                  java.util.function.Function<DrainageRiskItemVO, String> getter) {
+        return rows.stream()
+                .map(getter)
+                .filter(value -> !isBlank(value))
+                .distinct()
+                .sorted()
+                .collect(Collectors.toList());
+    }
+
+    private Map<String, Object> option(String value, String label) {
+        Map<String, Object> item = new LinkedHashMap<>();
+        item.put("value", value);
+        item.put("label", label);
+        return item;
+    }
+
+    private String levelName(String level) {
+        if (LEVEL_HIGH.equals(level)) {
+            return "高风险";
+        }
+        if (LEVEL_MEDIUM.equals(level)) {
+            return "中风险";
+        }
+        return "低风险";
+    }
+
+    private String levelColor(String level) {
+        if (LEVEL_HIGH.equals(level)) {
+            return "#F56C6C";
+        }
+        if (LEVEL_MEDIUM.equals(level)) {
+            return "#E6A23C";
+        }
+        return "#67C23A";
+    }
+
+    private int pipeAge(java.util.Date buildDate) {
+        if (buildDate == null) {
+            return 0;
+        }
+        int buildYear = buildDate.toInstant().atZone(ZoneId.systemDefault()).getYear();
+        return Math.max(0, LocalDate.now().getYear() - buildYear);
+    }
+
+    private String trim(String value) {
+        if (value == null) {
+            return null;
+        }
+        String trimmed = value.trim();
+        return trimmed.isEmpty() ? null : trimmed;
+    }
+
+    private boolean containsIgnoreCase(String source, String keyword) {
+        if (source == null) {
+            return false;
+        }
+        return source.toLowerCase(Locale.ROOT).contains(keyword.toLowerCase(Locale.ROOT));
+    }
+
+    private boolean isBlank(String value) {
+        return value == null || value.trim().isEmpty();
+    }
+
+    /**
+     * 分组查询结果取值：兼容不同数据库对列别名大小写的处理
+     */
+    private String str(Map<String, Object> row, String key) {
+        for (Map.Entry<String, Object> entry : row.entrySet()) {
+            if (entry.getKey() != null && entry.getKey().equalsIgnoreCase(key) && entry.getValue() != null) {
+                return entry.getValue().toString();
+            }
+        }
+        return null;
+    }
+
+    private long num(Map<String, Object> row, String key) {
+        String value = str(row, key);
+        if (value == null || value.isEmpty()) {
+            return 0L;
+        }
+        try {
+            return (long) Double.parseDouble(value);
+        } catch (NumberFormatException e) {
+            return 0L;
+        }
+    }
+
+    /* ==================== 内部结构 ==================== */
+
+    private static class Assessment {
+        private int total;
+        private String level;
+        private int alarmCount;
+        private int openAlarmCount;
+        private int waterloggingCount;
+        private List<Map<String, Object>> dimensions;
+        private List<String> factors;
+        private List<String> measures;
+        private String protectiveTargets;
+        private String hazardSources;
+    }
+
+    private static class Snapshot {
+        private final List<DrainageRiskItemVO> rows;
+        private final Map<String, PipeNetworkBase> pipes;
+        private final Map<String, Assessment> assessments;
+        private final List<EquipmentBase> equipments;
+        private final long loadTime;
+
+        private Snapshot(List<DrainageRiskItemVO> rows, Map<String, PipeNetworkBase> pipes,
+                         Map<String, Assessment> assessments, List<EquipmentBase> equipments,
+                         long loadTime) {
+            this.rows = rows;
+            this.pipes = pipes;
+            this.assessments = assessments;
+            this.equipments = equipments;
+            this.loadTime = loadTime;
+        }
+    }
+}

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/service/impl/DrainageWarningServiceImpl.java

@@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.zksy.common.exception.ServiceException;
+import com.zksy.common.utils.SecurityUtils;
 import com.zksy.base.domain.EquipmentBase;
 import com.zksy.base.enums.EquipmentTypeEnum;
 import com.zksy.base.mapper.EquipmentBaseMapper;
@@ -44,6 +45,8 @@ public class DrainageWarningServiceImpl extends ServiceImpl<DrainageWarningMappe
 
     private static final List<String> CURRENT_STATUSES = Arrays.asList("DRAFT", "PENDING", "RECEIVED", "PROCESSING");
     private static final List<String> HISTORY_STATUSES = Arrays.asList("HANDLED", "CLOSED");
+    /** warning_type 非空兜底值：设备侧发起、无预警类型时的默认值 */
+    private static final String DEFAULT_WARNING_TYPE = "DRAINAGE_DEVICE";
 
     @Override
     public Page<DrainageWarning> findByPage(long pageNum, long pageSize,
@@ -159,7 +162,8 @@ public class DrainageWarningServiceImpl extends ServiceImpl<DrainageWarningMappe
         update.setId(id);
         update.setStatus(status);
         update.setProcessor(processor);
-        update.setProcessTime(LocalDateTime.now());
+        // processTime 表里没有列，处置时间统一落 update_time，历史预警详情靠它展示
+        update.setUpdateTime(LocalDateTime.now());
         return this.updateById(update);
     }
 
@@ -176,12 +180,50 @@ public class DrainageWarningServiceImpl extends ServiceImpl<DrainageWarningMappe
     @Transactional
     public void publishWarning(DrainageWarning warning, List<Long> userIds) {
         // 自动生成预警编号
+        // warning_no 是唯一键，formatter 只到秒，批量发布时会撞号，补 3 位序号
         if (warning.getWarningNo() == null || warning.getWarningNo().isEmpty()) {
-            warning.setWarningNo("WRN" + LocalDateTime.now().format(WARNING_NO_FORMATTER));
+            warning.setWarningNo("WRN" + LocalDateTime.now().format(WARNING_NO_FORMATTER)
+                    + String.format("%03d", Math.abs((int) (System.nanoTime() % 1000))));
+        }
+        // early_warning.warning_type 非空且无默认值：设备侧发起预警时没有预警类型概念，
+        // 用设备编码兜底，保证插入不违反非空约束。
+        if (warning.getWarningType() == null || warning.getWarningType().trim().isEmpty()) {
+            String fallbackType = warning.getDeviceCode();
+            warning.setWarningType(fallbackType == null || fallbackType.trim().isEmpty()
+                    ? DEFAULT_WARNING_TYPE : fallbackType);
         }
         warning.setPublishTime(LocalDateTime.now());
         // 默认状态为PENDING（待办）
         warning.setStatus("PENDING");
+        // 发布人 / 创建人：接口是匿名的，取不到登录上下文时沿用前端传入的发布人，
+        // 再退化为 system，避免这两列在列表里始终为空。
+        String operator = currentOperator();
+        if (warning.getPublisher() == null || warning.getPublisher().trim().isEmpty()) {
+            warning.setPublisher(operator);
+        }
+        if (warning.getCreateBy() == null || warning.getCreateBy().trim().isEmpty()) {
+            warning.setCreateBy(operator);
+        }
+        // 预警发生时间表里没有独立列，用它覆盖创建时间，保证记录时间与业务时间一致
+        if (warning.getWarningTime() != null) {
+            warning.setCreateTime(warning.getWarningTime());
+        }
+        // 当前值/阈值同样是前端采集信息，表里无独立列，落到备注避免用户填写后丢失
+        if (warning.getRemark() == null || warning.getRemark().trim().isEmpty()) {
+            StringBuilder extra = new StringBuilder();
+            if (warning.getCurrentValue() != null && !warning.getCurrentValue().trim().isEmpty()) {
+                extra.append("当前值：").append(warning.getCurrentValue().trim());
+            }
+            if (warning.getThresholdValue() != null && !warning.getThresholdValue().trim().isEmpty()) {
+                if (extra.length() > 0) {
+                    extra.append("；");
+                }
+                extra.append("阈值：").append(warning.getThresholdValue().trim());
+            }
+            if (extra.length() > 0) {
+                warning.setRemark(extra.toString());
+            }
+        }
         this.save(warning);
 
         // publish_scope=3 指定人员：保存接收人员记录
@@ -197,6 +239,16 @@ public class DrainageWarningServiceImpl extends ServiceImpl<DrainageWarningMappe
         }
     }
 
+    /** 当前操作人：排水预警接口是匿名开放的，取不到登录上下文时退化为 system。 */
+    private String currentOperator() {
+        try {
+            String username = SecurityUtils.getUsername();
+            return username == null || username.trim().isEmpty() ? "system" : username.trim();
+        } catch (Exception e) {
+            return "system";
+        }
+    }
+
     @Override
     public Page<DrainageWarning> findVisibleCurrentPage(long pageNum, long pageSize,
                                                          Long userId, Long deptId, boolean isAdmin,

pipe-network-service/zksy-system/src/main/java/com/zksy/drainage/service/impl/WaterloggingPointServiceImpl.java

@@ -55,7 +55,8 @@ public class WaterloggingPointServiceImpl extends ServiceImpl<WaterloggingPointM
     @Override
     public Map<String, List<String>> getFilterOptions() {
         Map<String, List<String>> options = new LinkedHashMap<>();
-        options.put("district", distinctValues(WaterloggingPoint::getDistrict));
+        // district 不再参与：所属片区改由字典 drainage_district（0城南/1城北/2太常）提供选项。
+        // 用存量值去重当下拉项会把历史脏数据回灌到界面，这条路径已废弃。
         options.put("waterloggingLevel", distinctValues(WaterloggingPoint::getWaterloggingLevel));
         options.put("riskLevel", distinctValues(WaterloggingPoint::getRiskLevel));
         return options;
@@ -68,7 +69,8 @@ public class WaterloggingPointServiceImpl extends ServiceImpl<WaterloggingPointM
         LambdaQueryWrapper<WaterloggingPoint> wrapper = new LambdaQueryWrapper<>();
         wrapper.like(hasText(query.getPointCode()), WaterloggingPoint::getPointCode, query.getPointCode());
         wrapper.like(hasText(query.getPointName()), WaterloggingPoint::getPointName, query.getPointName());
-        wrapper.like(hasText(query.getDistrict()), WaterloggingPoint::getDistrict, query.getDistrict());
+        // 片区存的是字典编码（0/1/2），必须精确匹配；like 会让 '0' 命中 '10' 这类越界值
+        wrapper.eq(hasText(query.getDistrict()), WaterloggingPoint::getDistrict, query.getDistrict());
         wrapper.eq(query.getWaterloggingLevel() != null,
                 WaterloggingPoint::getWaterloggingLevel, query.getWaterloggingLevel());
         wrapper.eq(hasText(query.getRiskLevel()), WaterloggingPoint::getRiskLevel, query.getRiskLevel());

pipe-network-service/zksy-system/src/main/java/com/zksy/ops/service/impl/WorkOrderServiceImpl.java

@@ -875,15 +875,73 @@ public class WorkOrderServiceImpl extends ServiceImpl<WorkOrderMapper, WorkOrder
 
     private boolean isDeviceInCurrentCompany(String deviceId) {
         if (StringUtils.isEmpty(deviceId)) return true;
+        // 超级管理员按部门路径推断会落到某个具体公司（如 admin 挂在研发部门下），
+        // 只有该公司设备能派单，其余子系统全部报「无权操作其他公司范围」，与实际授权不符。
+        if (assignScopeSupport.isAdmin()) return true;
         EquipmentBase equipment = equipmentBaseMapper.selectById(deviceId);
         if (equipment == null || equipment.getEquipmentTypeId() == null) return false;
-        EquipmentType type = equipmentTypeMapper.selectById(equipment.getEquipmentTypeId());
-        return type != null && type.getParentId() != null && currentCompanyParentTypeIds().contains(type.getParentId());
+        Integer rootTypeId = rootEquipmentTypeId(equipment.getEquipmentTypeId());
+        if (rootTypeId == null) return false;
+        Set<Integer> scope = currentCompanyParentTypeIds();
+        // 当前部门不属于任何一家公司（集团/管理端账号）时公司边界不适用：
+        // 这类账号能看到全部子系统的设备，若按空范围判定会导致排水页面完全无法派单。
+        if (scope.isEmpty()) {
+            log.warn("当前部门不在供水/排水/燃气公司范围内，跳过设备公司范围校验，设备ID={}", deviceId);
+            return true;
+        }
+        return scope.contains(rootTypeId);
+    }
+
+    /**
+     * 沿设备类型的 parent_id 向上追溯到一级根类型。
+     *
+     * 设备类型表允许三级扩展（如 排水设备 → 液位计 → 平升液位流量终端），
+     * 直接取 parent_id 判断公司范围会把三级类型误判成其他公司。
+     */
+    private Integer rootEquipmentTypeId(Integer typeId) {
+        Integer current = typeId;
+        Set<Integer> visited = new HashSet<>();
+        while (current != null && visited.add(current)) {
+            EquipmentType type = equipmentTypeMapper.selectById(current);
+            if (type == null) return null;
+            Integer parentId = type.getParentId();
+            if (parentId == null || parentId == 0) return type.getId();
+            current = parentId;
+        }
+        return null;
     }
 
     private void requireDeviceCompanyScope(WorkOrder order) {
-        if (order != null && StringUtils.isNotEmpty(order.getDeviceId()) && !isDeviceInCurrentCompany(order.getDeviceId())) {
-            throw new ServiceException("无权操作其他公司范围的工单");
+        if (order == null || StringUtils.isEmpty(order.getDeviceId())) return;
+        if (isDeviceInCurrentCompany(order.getDeviceId())) return;
+        throw new ServiceException("无权操作其他公司范围的工单（设备属于" + deviceCompanyName(order.getDeviceId())
+                + "，当前账号属于" + currentCompanyName() + "）");
+    }
+
+    /** 设备所属公司名称（按设备类型一级根节点推断），仅用于错误提示。 */
+    private String deviceCompanyName(String deviceId) {
+        EquipmentBase equipment = equipmentBaseMapper.selectById(deviceId);
+        Integer rootTypeId = equipment == null || equipment.getEquipmentTypeId() == null
+                ? null : rootEquipmentTypeId(equipment.getEquipmentTypeId());
+        return companyNameOf(rootTypeId);
+    }
+
+    /** 当前账号所属公司名称，仅用于错误提示。 */
+    private String currentCompanyName() {
+        Set<Integer> scope = currentCompanyParentTypeIds();
+        if (scope.isEmpty()) return "未归属任一子公司";
+        return scope.stream().filter(id -> id != 4).sorted().findFirst()
+                .map(this::companyNameOf).orElse("未归属任一子公司");
+    }
+
+    private String companyNameOf(Integer rootTypeId) {
+        if (rootTypeId == null) return "未知设备类型";
+        switch (rootTypeId) {
+            case 1: return "供水公司";
+            case 2: return "排水公司";
+            case 3: return "燃气公司";
+            case 4: return "窨井共享设备";
+            default: return "未知设备类型";
         }
     }
 
@@ -904,13 +962,24 @@ public class WorkOrderServiceImpl extends ServiceImpl<WorkOrderMapper, WorkOrder
     /** 无设备及共享类型不按候选人的部门限制；关联损坏不可当作无设备放宽。 */
     private Long assignmentCompanyDeptId(WorkOrder order) {
         if (StringUtils.isEmpty(order.getDeviceId())) return null;
+        if (assignScopeSupport.isAdmin()) return null;
         EquipmentBase device = equipmentBaseMapper.selectById(order.getDeviceId());
-        EquipmentType type = device == null || device.getEquipmentTypeId() == null ? null
-                : equipmentTypeMapper.selectById(device.getEquipmentTypeId());
-        Integer parent = type == null ? null : type.getParentId();
-        if (Objects.equals(parent, 4)) return null;
-        if (parent != null && parent >= 1 && parent <= 3) return 100L + parent;
-        throw new ServiceException("工单关联设备或设备类型无效，无法选择人员");
+        if (device == null || device.getEquipmentTypeId() == null) {
+            throw new ServiceException("工单关联设备或设备类型无效，无法选择人员");
+        }
+        Integer rootTypeId = rootEquipmentTypeId(device.getEquipmentTypeId());
+        if (rootTypeId == null) {
+            throw new ServiceException("工单关联设备或设备类型无效，无法选择人员");
+        }
+        // 窨井为三家公司共享类型
+        if (Objects.equals(rootTypeId, 4)) return null;
+        if (rootTypeId < 1 || rootTypeId > 3) {
+            throw new ServiceException("工单关联设备或设备类型无效，无法选择人员");
+        }
+        long companyDeptId = 100L + rootTypeId;
+        // 公司部门未按 101/102/103 建账时，按部门过滤只会得到空候选人；
+        // 此时退化为不限部门，由角色权限与可指派范围继续约束。
+        return assignmentDeptMapper.selectDeptById(companyDeptId) == null ? null : companyDeptId;
     }
 
     private void requireAssignmentCompany(WorkOrder order, Long userId, String label) {

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

@@ -122,7 +122,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
 	</select>
 	
     <select id="selectAllEnabledUser" resultMap="SysUserResult">
-        select u.user_id, u.dept_id, u.nick_name, u.user_name, d.dept_name
+        select u.user_id, u.dept_id, u.nick_name, u.user_name, u.phonenumber, d.dept_name
         from app_user.sys_user u
         left join app_user.sys_dept d on d.dept_id = u.dept_id
         where u.del_flag = '0' and u.status = '0'