pipe-ner-server/pipe-network-service/sql/apply_drainage_jcsj_fix.sql

-- ============================================================================
-- 排水基础数据：一次性补齐脚本（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;