gendesign/data/sql/178b_load_cadastral_value.py

"""Загрузчик кварталов dataset_КАДАСТРСТОИМОСТЬ (кадастровая стоимость) в gendesign.

Region-66-scoped: при INSERT фильтруем region_code='66' (Свердловская обл.), хотя
исходный CSV покрывает регионы 60-92. Источник — открытый квартальный датасет
Росреестра. Заливается через psycopg v3 (staging COPY → typed INSERT).

Назначение: cost_per_area (кадастровая ₽/m²) даёт «пол» цены на квартал, по которому
denoise-миграция 179 отбрасывает зашумлённые сделки rosreestr_deals дешевле
0.7x кадастрового медианного ₽/m².

Idempotent: пропускает source_quarter, уже залитый в rosreestr_cadastral_value.
Запускается ВРУЧНУЮ оператором после деплоя; затем REFRESH двух MV (см. 179).

Usage:
    cd backend
    uv run python ../data/sql/178b_load_cadastral_value.py 2026Q1
    uv run python ../data/sql/178b_load_cadastral_value.py            # все из JOBS
"""

from __future__ import annotations

import os
import sys
import time
from pathlib import Path

import psycopg

PG_DSN = os.environ.get(
    "TUNNEL_DSN",
    "postgresql://gendesign:2J2SBPMKuS998fiwhtQqDhMI@localhost:15432/gendesign",
)

RAW = Path(__file__).resolve().parent.parent / "raw"

# Датасет всегда UTF-8, delimiter '~', header columns в порядке staging-таблицы.
DELIMITER = "~"

# (source_quarter, csv_filename)
JOBS: list[tuple[str, str]] = [
    ("2026Q1", "dataset_КАДАСТРСТОИМОСТЬ_r-r_60-92_y_2026_q_1.csv"),
]


# Region-66-scoped typed INSERT. Staging — всё TEXT, поэтому каждое числовое поле
# проходит regex-guard + NULLIF перед ::cast (мусорные строки → NULL, не падаем).
INSERT_SQL = """
INSERT INTO rosreestr_cadastral_value (
    source_quarter,
    number, realestate_type_code, purpose_code, area, region_code,
    district, city, quarter_cad_number, street,
    registration_quarter, application_quarter, cost, cost_per_area, period
)
SELECT
    %s,
    CASE WHEN number ~ '^[0-9]+$' THEN number::INTEGER ELSE NULL END,
    NULLIF(realestate_type_code, ''),
    NULLIF(purpose_code, ''),
    CASE WHEN area ~ '^[0-9]+(\\.[0-9]+)?$' AND area::NUMERIC > 0
         THEN area::NUMERIC ELSE NULL END,
    region_code::SMALLINT,
    NULLIF(district, ''),
    NULLIF(city, ''),
    NULLIF(quarter_cad_number, ''),
    NULLIF(street, ''),
    NULLIF(registration_quarter, ''),
    NULLIF(application_quarter, ''),
    CASE WHEN cost ~ '^[0-9]+(\\.[0-9]+)?$' AND cost::NUMERIC > 0
         THEN cost::NUMERIC ELSE NULL END,
    CASE WHEN cost_per_area ~ '^[0-9]+(\\.[0-9]+)?$' AND cost_per_area::NUMERIC > 0
         THEN cost_per_area::NUMERIC ELSE NULL END,
    NULLIF(period, '')
FROM rosreestr_cadastral_value_staging
WHERE region_code = '66'
  AND NULLIF(quarter_cad_number, '') IS NOT NULL
"""


def main() -> int:
    targets = sys.argv[1:] or [j[0] for j in JOBS]
    selected = [j for j in JOBS if j[0] in targets]
    if not selected:
        print(f"Нет matching кварталов в JOBS. Targets: {targets}")
        return 1

    conn = psycopg.connect(PG_DSN, connect_timeout=15)
    try:
        for sq, fname in selected:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT COUNT(*) FROM rosreestr_cadastral_value "
                    "WHERE source_quarter = %s",
                    (sq,),
                )
                already = cur.fetchone()[0]
            if already:
                print(
                    f"=== {sq} SKIP — уже {already} строк "
                    f"(DELETE FROM rosreestr_cadastral_value "
                    f"WHERE source_quarter='{sq}' чтобы перезалить)"
                )
                continue
            csv_path = RAW / fname
            if not csv_path.exists():
                print(f"=== {sq} SKIP — нет файла {csv_path}")
                continue
            size_mb = csv_path.stat().st_size / 1e6
            print(f"=== {sq}  ({fname}, sep='{DELIMITER}', {size_mb:.1f} MB) ===")
            t0 = time.time()
            with conn.cursor() as cur:
                cur.execute("TRUNCATE rosreestr_cadastral_value_staging")
                copy_sql = (
                    f"COPY rosreestr_cadastral_value_staging FROM STDIN "
                    f"(FORMAT csv, HEADER true, DELIMITER '{DELIMITER}', "
                    f"QUOTE '\"', NULL '')"
                )
                with cur.copy(copy_sql) as copy:
                    with open(csv_path, "rb") as f:
                        while True:
                            chunk = f.read(8 * 1024 * 1024)
                            if not chunk:
                                break
                            copy.write(chunk)
                t_copy = time.time() - t0
                cur.execute(INSERT_SQL, (sq,))
                inserted = cur.rowcount
            conn.commit()
            t_total = time.time() - t0
            print(
                f"  COPY {t_copy:.0f}s, INSERT (region 66) {inserted} rows, "
                f"total {t_total:.0f}s"
            )

        # Final summary
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT source_quarter,
                       COUNT(*) AS rows,
                       COUNT(*) FILTER (
                           WHERE realestate_type_code = '002001003000'
                       ) AS apt_rows
                  FROM rosreestr_cadastral_value
                 GROUP BY source_quarter
                 ORDER BY source_quarter
                """
            )
            print("\n=== Итоги по rosreestr_cadastral_value (region 66) ===")
            for sq, rows, apt_rows in cur.fetchall():
                print(f"  {sq}  rows={rows:>9}  apartments={apt_rows:>9}")
    finally:
        conn.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())