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

-- 工程设施基础数据（闸门 / 调蓄设施 / 泵站）表重建与迁移（KingbaseES, schema app_user）
--
-- 需求依据：《排水管网安全运行监测系统》5.3.3.1.3 工程设施基础数据管理
--   「支持对闸门、调蓄设施、泵站等工程设施的查看、查询、筛选等管理操作，
--     并可将相关要素在 GIS 地图中叠加展示。」
--
-- 背景：旧表 app_user.psgw_engineering_facility 是「监测点」口径的临时表结构，
--   字段名与语义都不对：name 注释写成「监测点名称/设施名称」、area 实为行政区域、
--   time 是「数据时间」、point_count 由前端现算却落库、points 存多节点 JSON，
--   而一个工程设施在地图上就是一个点要素，应直接落经纬度。
--
-- 本脚本做三件事：
--   1. 建新表 app_user.drainage_engineering_facility（按基础档案口径设计）；
--   2. 登记两个字典：drainage_facility_type（设施类型）、drainage_facility_status（运行状态）；
--   3. 把旧表数据搬到新表（坐标取 points 的第一个点；旧表保留不删，确认无误后再自行下线）。
--
-- 幂等：建表/索引/字典均可重复执行；数据迁移仅在「新表为空」时执行一次。
--
-- =====================================================================
-- 为什么字典 INSERT 要「显式指定主键」
-- ---------------------------------------------------------------------
-- 本库由 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 都拿不到序列。
-- 因此这里统一写成 max(主键) + n，插入后再尽力校准序列。
-- =====================================================================
--
-- 前置（必须先执行）：结束上一次失败执行遗留的 aborted 事务
--   上一次语句报错后，PG/KES 会把该事务标记为已失败，此时同一连接上后续任何语句都只返回
--   「当前事务被终止，事务块结束之前的查询被忽略」，必须先回滚才能继续。
--   注意：本句会回滚该连接上所有未提交事务；若你有别的未提交手工操作，请先单独提交再跑本脚本。
ROLLBACK;

BEGIN;

-- =====================================================================
-- 1. 建新表
-- =====================================================================
CREATE TABLE IF NOT EXISTS app_user.drainage_engineering_facility (
    facility_id      varchar(64)    NOT NULL,
    facility_code    varchar(64),
    facility_name    varchar(200)   NOT NULL,
    facility_type    varchar(8)     NOT NULL,
    facility_level   varchar(8),
    management_unit  varchar(200),
    location         varchar(500),
    longitude        numeric(12, 8),
    latitude         numeric(12, 8),
    status           varchar(8)     DEFAULT '0',
    build_date       date,
    design_flow      numeric(12, 2),
    design_head      numeric(10, 2),
    storage_capacity numeric(14, 2),
    equipment_id     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_engineering_facility PRIMARY KEY (facility_id)
);

COMMENT ON TABLE  app_user.drainage_engineering_facility IS '工程设施基础数据（闸门/调蓄设施/泵站）';
COMMENT ON COLUMN app_user.drainage_engineering_facility.facility_id      IS '设施ID（主键）';
COMMENT ON COLUMN app_user.drainage_engineering_facility.facility_code    IS '设施编码';
COMMENT ON COLUMN app_user.drainage_engineering_facility.facility_name    IS '设施名称';
COMMENT ON COLUMN app_user.drainage_engineering_facility.facility_type    IS '设施类型（字典 drainage_facility_type：0闸门 1调蓄设施 2泵站）';
COMMENT ON COLUMN app_user.drainage_engineering_facility.facility_level   IS '设施等级';
COMMENT ON COLUMN app_user.drainage_engineering_facility.management_unit  IS '管理单位';
COMMENT ON COLUMN app_user.drainage_engineering_facility.location        IS '位置描述';
COMMENT ON COLUMN app_user.drainage_engineering_facility.longitude       IS 'WGS84 经度';
COMMENT ON COLUMN app_user.drainage_engineering_facility.latitude        IS 'WGS84 纬度';
COMMENT ON COLUMN app_user.drainage_engineering_facility.status          IS '运行状态（字典 drainage_facility_status：0正常 1检修 2停用 3故障）';
COMMENT ON COLUMN app_user.drainage_engineering_facility.build_date      IS '建设日期';
COMMENT ON COLUMN app_user.drainage_engineering_facility.design_flow     IS '设计流量（m³/s）';
COMMENT ON COLUMN app_user.drainage_engineering_facility.design_head     IS '设计扬程（m）';
COMMENT ON COLUMN app_user.drainage_engineering_facility.storage_capacity IS '调蓄容积（m³）';
COMMENT ON COLUMN app_user.drainage_engineering_facility.equipment_id    IS '关联设备ID（多个用逗号分隔，沿用旧口径）';
COMMENT ON COLUMN app_user.drainage_engineering_facility.remark          IS '备注';
COMMENT ON COLUMN app_user.drainage_engineering_facility.del_flag        IS '删除标志（0正常 2删除）';

CREATE INDEX IF NOT EXISTS idx_def_facility_type   ON app_user.drainage_engineering_facility (facility_type);
CREATE INDEX IF NOT EXISTS idx_def_facility_status ON app_user.drainage_engineering_facility (status);
CREATE INDEX IF NOT EXISTS idx_def_management_unit ON app_user.drainage_engineering_facility (management_unit);
CREATE INDEX IF NOT EXISTS idx_def_facility_name   ON app_user.drainage_engineering_facility (facility_name);
CREATE UNIQUE INDEX IF NOT EXISTS uk_def_facility_code ON app_user.drainage_engineering_facility (facility_code);

-- =====================================================================
-- 2. 字典：设施类型、运行状态
-- =====================================================================
-- 2.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_facility_type', '0', 'admin', CURRENT_TIMESTAMP,
       '工程设施类型：0闸门 1调蓄设施 2泵站'
 WHERE NOT EXISTS (SELECT 1 FROM app_user.sys_dict_type WHERE dict_type = 'drainage_facility_type');

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_facility_type';

DELETE FROM app_user.sys_dict_data WHERE dict_type = 'drainage_facility_type';

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_facility_type',
       '', v.list_class, v.is_default, '0', 'admin', CURRENT_TIMESTAMP, v.remark
  FROM (VALUES
        (1, 1, '闸门',     '0', 'primary', 'Y', '工程设施类型：闸门'),
        (2, 2, '调蓄设施', '1', 'warning', 'N', '工程设施类型：调蓄设施'),
        (3, 3, '泵站',     '2', 'success', 'N', '工程设施类型：泵站')
       ) AS v(rn, dict_sort, dict_label, dict_value, list_class, is_default, remark)
 ORDER BY v.rn;

-- 2.2 运行状态
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_facility_status', '0', 'admin', CURRENT_TIMESTAMP,
       '设施运行状态：0正常 1检修 2停用 3故障'
 WHERE NOT EXISTS (SELECT 1 FROM app_user.sys_dict_type WHERE dict_type = 'drainage_facility_status');

UPDATE app_user.sys_dict_type
   SET dict_name = '设施运行状态',
       status = '0',
       update_by = 'admin',
       update_time = CURRENT_TIMESTAMP,
       remark = '设施运行状态：0正常 1检修 2停用 3故障'
 WHERE dict_type = 'drainage_facility_status';

DELETE FROM app_user.sys_dict_data WHERE dict_type = 'drainage_facility_status';

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_facility_status',
       '', 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', 'info',    'N', '设施运行状态：停用'),
        (4, 4, '故障', '3', 'danger',  'N', '设施运行状态：故障')
       ) AS v(rn, dict_sort, dict_label, dict_value, list_class, is_default, remark)
 ORDER BY v.rn;

-- =====================================================================
-- 3. 自增计数器校准（放在插入之后：此时 max 已含新插入行）
--    三种方式依次尝试，任一成功即可；都不成功只提示，不影响已插入的数据。
-- =====================================================================
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 $$;

-- =====================================================================
-- 4. 旧表数据迁移（仅当新表为空时执行一次）
--    坐标：旧表 points 是 JSON 数组，取第一个元素的 lng/lat；
--          解析失败或为空时置 NULL，该设施不会出现在地图上，但列表仍可见。
--    旧表字段若有缺失，整段迁移跳过并提示，不影响建表与字典。
-- =====================================================================
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_engineering_facility') IS NULL THEN
    RAISE NOTICE '旧表 app_user.psgw_engineering_facility 不存在，跳过数据迁移';
    RETURN;
  END IF;

  IF EXISTS (SELECT 1 FROM app_user.drainage_engineering_facility) THEN
    RAISE NOTICE '新表 app_user.drainage_engineering_facility 已有数据，跳过迁移（避免重复导入）';
    RETURN;
  END IF;

  FOR r IN SELECT * FROM app_user.psgw_engineering_facility LOOP
    v_idx := v_idx + 1;
    v_lng := NULL;
    v_lat := NULL;

    BEGIN
      IF r.points IS NOT NULL AND r.points <> '' THEN
        v_lng := (regexp_match(r.points, '"lng"\s*:\s*(-?[0-9]+(\.[0-9]+)?)'))[1]::numeric(12, 8);
        v_lat := (regexp_match(r.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_engineering_facility (
        facility_id, facility_code, facility_name, facility_type, facility_level,
        management_unit, location, longitude, latitude, status, build_date,
        design_flow, design_head, storage_capacity, equipment_id, remark,
        create_by, create_time, update_by, update_time, del_flag
    ) VALUES (
        r.id,
        'GCSS-' || lpad(v_idx::text, 4, '0'),
        COALESCE(NULLIF(trim(r.name), ''), '未命名设施'),
        CASE r.facility_type
            WHEN '闸门'     THEN '0'
            WHEN '调蓄设施' THEN '1'
            WHEN '泵站'     THEN '2'
            WHEN '0'        THEN '0'
            WHEN '1'        THEN '1'
            WHEN '2'        THEN '2'
            ELSE '0'
        END,
        NULL,
        NULL,
        r.area,
        v_lng,
        v_lat,
        '0',
        NULL,
        NULL, NULL, NULL,
        r.equipment_id,
        NULL,
        '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 $$;

-- =====================================================================
-- 5. 结果核验（可选，取消注释执行）
-- =====================================================================
-- 设施类型与数量
--    SELECT facility_type, count(*) FROM app_user.drainage_engineering_facility
--     WHERE del_flag = '0' GROUP BY facility_type ORDER BY facility_type;
-- 能在地图上展示的条数（有坐标）
--    SELECT count(*) FROM app_user.drainage_engineering_facility
--     WHERE del_flag = '0' AND longitude IS NOT NULL AND latitude IS NOT NULL;
-- 字典是否齐全（应分别返回 3 行、4 行）
--    SELECT dict_type, dict_value, dict_label, list_class FROM app_user.sys_dict_data
--     WHERE dict_type IN ('drainage_facility_type', 'drainage_facility_status')
--     ORDER BY dict_type, dict_sort;

COMMIT;