gendesign/data/sql/73_complexes_master.sql

-- ЖК как canonical master entity + cross-ref на источники.
--
-- Зачем:
--   До этого жил мир из 5+ таблиц с дублирующейся семантикой "ЖК":
--     - domrf_kn_objects (4548, ДОМ.РФ kn-API)
--     - objective_lots.project_name (350 уникальных, Objective API)
--     - objective_corpus_room_month.project_name (тот же, 350)
--     - objective_complex_mapping (114 fuzzy matches → domrf_obj_id)
--     - cad_buildings.cad_num (косвенно через NSPD)
--     - rosreestr_deals.complex_name (текстовое поле в сделках)
--   Каждый источник называет ЖК по-своему ("ЖК \"Парк Культуры\"" vs
--   "Парк Культуры" vs domrf_obj_id=66845). recommend_mix /
--   velocity-anomaly / cross-source analytics требуют одного entity.
--
-- Архитектура:
--   complexes (master) — одна строка = один реальный ЖК
--   complex_sources (cross-ref) — many-to-many ЖК ↔ источник + source_id
--
-- Стратегия backfill:
--   1. Каждый domrf_kn_objects = entity в complexes (4548 ЖК)
--      + complex_sources(source='domrf_kn', source_id=obj_id)
--   2. Сматченные objective project_name (114) → существующий complex по
--      objective_complex_mapping.domrf_obj_id
--      + complex_sources(source='objective', source_id=project_name)
--   3. Unmatched objective project_name (236) → новые entity в complexes
--      + complex_sources(source='objective')
--
-- Идемпотентно через ON CONFLICT.

BEGIN;

-- ── 1. complexes (canonical master) ─────────────────────────────────────────

CREATE TABLE IF NOT EXISTS complexes (
    id                  BIGSERIAL PRIMARY KEY,
    -- Display
    canonical_name      TEXT NOT NULL,            -- "Парк Культуры" / "ЖК Авангард 3.0"
    developer_name      TEXT,                     -- "PRINZIP" / "ГК КОРТРОС"
    developer_id        TEXT,                     -- FK on domrf_developers.developer_id (nullable)
    -- Geo
    region_id           INTEGER,                  -- 66 для Свердл.обл (FK on domrf_regions.region_id)
    district_name       TEXT,                     -- "Академический" / "Автовокзал"
    address             TEXT,
    latitude            NUMERIC,
    longitude           NUMERIC,
    geom                geometry(Point, 4326),    -- spatial для ST_DWithin / spatial join
    -- Spec
    obj_class           TEXT,                     -- "комфорт" / "бизнес" / "стандарт" / "элитный"
    flat_count          INTEGER,                  -- общее ПД
    square_living       NUMERIC,                  -- общая жилая площадь
    floor_min           INTEGER,
    floor_max           INTEGER,
    wall_type           TEXT,                     -- "монолит" / "кирпич" / etc.
    -- Stage
    sales_start_date    DATE,
    plan_completion_date DATE,
    actual_completion_date DATE,
    site_status         TEXT,                     -- "строящийся" / "сдан" / "архив"
    obj_status          INTEGER,                  -- 0/1/2 кодировка ДОМ.РФ
    is_problem          BOOLEAN DEFAULT FALSE,
    has_escrow          BOOLEAN,                  -- эскроу-проект
    -- Audit
    primary_source      TEXT NOT NULL DEFAULT 'domrf_kn',  -- какой источник был основой
    created_at          timestamptz NOT NULL DEFAULT NOW(),
    updated_at          timestamptz NOT NULL DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS complexes_name_idx
    ON complexes(canonical_name);
CREATE INDEX IF NOT EXISTS complexes_developer_idx
    ON complexes(developer_id) WHERE developer_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS complexes_region_district_idx
    ON complexes(region_id, district_name);
CREATE INDEX IF NOT EXISTS complexes_class_idx
    ON complexes(obj_class) WHERE obj_class IS NOT NULL;
CREATE INDEX IF NOT EXISTS complexes_geom_idx
    ON complexes USING GIST (geom) WHERE geom IS NOT NULL;
CREATE INDEX IF NOT EXISTS complexes_status_idx
    ON complexes(site_status) WHERE site_status IS NOT NULL;

COMMENT ON TABLE complexes IS
    'Canonical master entity для ЖК. Один реальный жилой комплекс = одна строка. '
    'Связи на разные источники (domrf_kn, objective, rosreestr, etc.) — через '
    'complex_sources. Все analytics-запросы должны идти через эту таблицу '
    'вместо direct lookup в источниках.';

-- ── 2. complex_sources (cross-ref) ──────────────────────────────────────────

CREATE TABLE IF NOT EXISTS complex_sources (
    id                  BIGSERIAL PRIMARY KEY,
    complex_id          BIGINT NOT NULL REFERENCES complexes(id) ON DELETE CASCADE,
    source              TEXT NOT NULL,            -- 'domrf_kn' | 'objective' | 'rosreestr' | 'cad_buildings' | 'yandex' | ...
    source_id           TEXT NOT NULL,            -- obj_id::text / project_name / cad_num / external_id
    source_external_id  BIGINT,                   -- для int-ID источников (objective.project_id, domrf.obj_id)
    -- Match audit
    match_method        TEXT,                     -- 'exact_name' | 'fuzzy_name' | 'spatial' | 'manual' | 'imported'
    match_score         NUMERIC,                  -- 0..1 для fuzzy
    is_reviewed         BOOLEAN NOT NULL DEFAULT FALSE,
    note                TEXT,
    linked_at           timestamptz NOT NULL DEFAULT NOW(),
    UNIQUE (source, source_id)                    -- один (источник, source_id) — на один ЖК
);

CREATE INDEX IF NOT EXISTS complex_sources_complex_idx
    ON complex_sources(complex_id);
CREATE INDEX IF NOT EXISTS complex_sources_source_idx
    ON complex_sources(source);
CREATE INDEX IF NOT EXISTS complex_sources_external_id_idx
    ON complex_sources(source, source_external_id) WHERE source_external_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS complex_sources_unreviewed_idx
    ON complex_sources(is_reviewed) WHERE is_reviewed = FALSE;

COMMENT ON TABLE complex_sources IS
    'Cross-ref many-to-many: ЖК (complexes) ↔ его представления в источниках. '
    'Один реальный ЖК может иметь записи в нескольких источниках — здесь '
    'хранятся все эти ссылки. UNIQUE(source, source_id) гарантирует что один '
    '(источник, ID-в-источнике) присвоен ровно одному ЖК.';

-- ── 3. Backfill из domrf_kn_objects ─────────────────────────────────────────

-- ВАЖНО: domrf_kn_objects содержит несколько snapshot'ов на один obj_id
-- (4548 rows = 1516 уникальных obj_id × ~3 snapshot_date). Поэтому
-- DISTINCT ON (obj_id) ORDER BY snapshot_date DESC — берём свежий snapshot.
--
-- ВНИМАНИЕ: domrf_kn_objects.obj_class = NULL во всех строках (известный
-- баг kn-API), поэтому obj_class останется NULL — он подтянется когда
-- сделаем UPDATE из objective_lots (там obj_class заполнен).

WITH latest_kn AS (
    SELECT DISTINCT ON (obj_id)
        obj_id, comm_name, addr, region_cd, dev_id, dev_name, district_name,
        latitude, longitude, obj_class, flat_count, square_living,
        floor_min, floor_max, wall_type, ready_dt,
        site_status, obj_status, is_problem, escrow
      FROM domrf_kn_objects
     ORDER BY obj_id, snapshot_date DESC NULLS LAST
),
inserted_complexes AS (
    INSERT INTO complexes (
        canonical_name, developer_name, developer_id, region_id, district_name,
        address, latitude, longitude, geom,
        obj_class, flat_count, square_living, floor_min, floor_max, wall_type,
        plan_completion_date, site_status, obj_status, is_problem, has_escrow,
        primary_source
    )
    SELECT
        COALESCE(comm_name, addr, 'unknown'),
        dev_name, dev_id, region_cd, district_name,
        addr, latitude, longitude,
        CASE WHEN latitude IS NOT NULL AND longitude IS NOT NULL
             THEN ST_SetSRID(ST_MakePoint(longitude, latitude), 4326) ELSE NULL END,
        obj_class, flat_count, square_living, floor_min, floor_max, wall_type,
        ready_dt, site_status, obj_status, COALESCE(is_problem, FALSE), escrow,
        'domrf_kn'
    FROM latest_kn
    RETURNING id, canonical_name
)
INSERT INTO complex_sources (
    complex_id, source, source_id, source_external_id,
    match_method, match_score, is_reviewed, note
)
SELECT
    ic.id, 'domrf_kn', lk.obj_id::text, lk.obj_id, 'imported', 1.0, TRUE,
    'auto-import from domrf_kn_objects (latest snapshot per obj_id)'
FROM latest_kn lk
JOIN inserted_complexes ic
  ON ic.canonical_name = COALESCE(lk.comm_name, lk.addr, 'unknown')
ON CONFLICT (source, source_id) DO NOTHING;

-- ── 4. Backfill из objective_complex_mapping (114 сматченных) ───────────────

-- Эти project_name уже привязаны к domrf_obj_id — линкуем к существующему complex.
INSERT INTO complex_sources (
    complex_id, source, source_id, source_external_id,
    match_method, match_score, is_reviewed, note
)
SELECT
    cs_kn.complex_id,                             -- complex по domrf_kn
    'objective',
    m.objective_complex_name,
    m.objective_project_id,
    m.match_method,
    m.match_score,
    m.is_reviewed,
    'imported from objective_complex_mapping'
FROM objective_complex_mapping m
JOIN complex_sources cs_kn
  ON cs_kn.source = 'domrf_kn'
 AND cs_kn.source_external_id = m.domrf_obj_id
WHERE m.domrf_obj_id IS NOT NULL
ON CONFLICT (source, source_id) DO NOTHING;

-- ── 5. Backfill orphan-проектов из objective_lots (236 unmatched) ───────────

-- Для каждого project_name которого нет ни в complex_sources(source=objective)
-- — создаём новый complex (берём атрибуты из любой quartiry этого проекта)
-- + добавляем complex_sources(source=objective).

WITH unmatched_objective AS (
    SELECT DISTINCT ON (project_name)
        project_name,
        objective_project_id,
        developer,
        district,
        class AS obj_class,
        city,
        address,
        sales_start_date,
        plan_completion_date,
        actual_completion_date,
        construction_stage
      FROM objective_lots ol
     WHERE NOT EXISTS (
        SELECT 1 FROM complex_sources cs
         WHERE cs.source = 'objective'
           AND cs.source_id = ol.project_name
     )
     ORDER BY project_name, snapshot_date DESC NULLS LAST
),
inserted AS (
    INSERT INTO complexes (
        canonical_name, developer_name, region_id, district_name,
        address, obj_class, sales_start_date, plan_completion_date,
        actual_completion_date, site_status, primary_source
    )
    SELECT
        project_name,
        developer,
        66,                                    -- Свердл.обл (Объектив пока только Екб + спутники)
        district,
        address,
        obj_class,
        sales_start_date,
        plan_completion_date,
        actual_completion_date,
        construction_stage,
        'objective'
    FROM unmatched_objective
    RETURNING id, canonical_name
)
INSERT INTO complex_sources (
    complex_id, source, source_id, source_external_id,
    match_method, match_score, is_reviewed, note
)
SELECT
    i.id,
    'objective',
    uo.project_name,
    uo.objective_project_id,
    'imported',
    1.0,
    FALSE,                                     -- нужно ревью: возможно есть domrf_kn-аналог
    'orphan objective project; consider reviewing for domrf_kn match'
FROM unmatched_objective uo
JOIN inserted i ON i.canonical_name = uo.project_name
ON CONFLICT (source, source_id) DO NOTHING;

-- ── 6. Helper-views для запросов ────────────────────────────────────────────

-- v_complex_full: полная карточка ЖК с метриками из всех источников.
DROP VIEW IF EXISTS v_complex_full;
CREATE VIEW v_complex_full AS
SELECT
    c.id AS complex_id,
    c.canonical_name,
    c.developer_name,
    c.region_id,
    c.district_name,
    c.obj_class,
    c.address,
    c.latitude,
    c.longitude,
    c.flat_count,
    c.square_living,
    c.site_status,
    -- сводка по источникам
    (SELECT array_agg(source ORDER BY source) FROM complex_sources WHERE complex_id = c.id) AS sources,
    (SELECT COUNT(*) FROM complex_sources WHERE complex_id = c.id) AS sources_count,
    -- domrf_kn obj_id (если есть)
    (SELECT source_external_id FROM complex_sources
       WHERE complex_id = c.id AND source = 'domrf_kn' LIMIT 1) AS domrf_obj_id,
    -- objective project_name (если есть)
    (SELECT source_id FROM complex_sources
       WHERE complex_id = c.id AND source = 'objective' LIMIT 1) AS objective_project_name,
    c.primary_source,
    c.created_at,
    c.updated_at
  FROM complexes c;

COMMENT ON VIEW v_complex_full IS
    'Карточка ЖК с агрегированной информацией о всех источниках. '
    'Используй для UI / отчётов / вместо direct lookup в domrf_kn_objects.';

-- v_complex_with_objective: ЖК + агрегаты Objective (lots count, sold pct, avg price)
DROP VIEW IF EXISTS v_complex_with_objective;
CREATE VIEW v_complex_with_objective AS
SELECT
    c.id AS complex_id,
    c.canonical_name,
    c.district_name,
    c.obj_class,
    cs.source_id AS objective_project_name,
    COUNT(ol.id) AS objective_lots_n,
    COUNT(ol.id) FILTER (WHERE ol.is_sold) AS objective_sold_n,
    ROUND(100.0 * COUNT(ol.id) FILTER (WHERE ol.is_sold)
                / NULLIF(COUNT(ol.id), 0), 1) AS objective_sold_pct,
    AVG(ol.price_per_m2_rub) FILTER (WHERE ol.is_sold AND ol.price_per_m2_rub > 0) AS avg_sold_price_per_m2,
    AVG(ol.area_pd) FILTER (WHERE ol.area_pd > 0) AS avg_area_pd,
    MIN(ol.registration_date) AS first_reg_date,
    MAX(ol.registration_date) AS last_reg_date
  FROM complexes c
  JOIN complex_sources cs
    ON cs.complex_id = c.id AND cs.source = 'objective'
  JOIN objective_lots ol
    ON ol.project_name = cs.source_id
 GROUP BY c.id, c.canonical_name, c.district_name, c.obj_class, cs.source_id;

COMMENT ON VIEW v_complex_with_objective IS
    'ЖК + агрегаты Objective per-flat: sold count/pct, средняя цена, '
    'диапазон reg_date. NULL во всех колонках = ЖК без Objective-источника.';

COMMIT;