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

-- 易积水点台账表（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;