"""Тесты ГАР-loader'а знаменателя (app/services/gar_flats_loader.py, мигр. 143).

Структура (mirror'ит test_cadastral_geo_match):
  - ДЕТЕРМИНИРОВАННОЕ ядро (без БД): синтетические крошечные ГАР-XML фикстуры в tmp_path →
    парс → assert apartment-count на дом, сборка адреса, фильтрация ISACTUAL/ISACTIVE,
    дом с 0 квартир → flat_count 0, мульти-сегментная сборка адреса.
  - СТАТИЧЕСКИЕ asserts по SQL (_INSERT_SQL / _MATCH_SQL): ON CONFLICT, psycopg-v3 CAST,
    DISTINCT ON tie-break, идемпотентный gate.
  - ОПЦИОНАЛЬНЫЙ real-Postgres тест upsert+match → houses.gar_flat_count (self-SKIP без БД,
    создаёт TEMP-таблицы + fn в транзакции и откатывает).
"""

from __future__ import annotations

import inspect
import os
import re
from pathlib import Path
from typing import Any

import pytest

# settings нужен DSN на импорте (как в sibling-тестах); ядро статично/без БД.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.services import gar_flats_loader as gfl

# ─────────────────────────────────────────────────────────────────────────────
# Фикстуры: крошечный синтетический ГАР-дамп
# ─────────────────────────────────────────────────────────────────────────────
# Иерархия:
#   1 регион «обл Свердловская»
#     └─ 2 город «г Екатеринбург»
#          ├─ 3 улица «ул Шаумяна»
#          │    ├─ 100 дом «10»   → квартиры 1001,1002,1003 (3 активные) + 1004(ISACTIVE=0)
#          │    └─ 200 дом «5 к1» → квартиры 2001,2002 (2 активные)
#          └─ 30 улица «ул Ленина»
#               └─ 300 дом «7»    → 0 квартир
#   inactive: адр-объект 9 (ISACTIVE=0), дом 900 (ISACTUAL=0), кв 1005 (ISACTUAL=0)

_ADDR_OBJ_XML = """<?xml version="1.0" encoding="utf-8"?>
<ADDRESSOBJECTS>
  <OBJECT OBJECTID="1" OBJECTGUID="g-region" NAME="Свердловская" TYPENAME="обл"
          LEVEL="1" ISACTUAL="1" ISACTIVE="1"/>
  <OBJECT OBJECTID="2" OBJECTGUID="g-city" NAME="Екатеринбург" TYPENAME="г"
          LEVEL="5" ISACTUAL="1" ISACTIVE="1"/>
  <OBJECT OBJECTID="3" OBJECTGUID="g-shaumyana" NAME="Шаумяна" TYPENAME="ул"
          LEVEL="8" ISACTUAL="1" ISACTIVE="1"/>
  <OBJECT OBJECTID="30" OBJECTGUID="g-lenina" NAME="Ленина" TYPENAME="ул"
          LEVEL="8" ISACTUAL="1" ISACTIVE="1"/>
  <OBJECT OBJECTID="9" OBJECTGUID="g-dead" NAME="Снесённая" TYPENAME="ул"
          LEVEL="8" ISACTUAL="1" ISACTIVE="0"/>
</ADDRESSOBJECTS>
"""

_HOUSES_XML = """<?xml version="1.0" encoding="utf-8"?>
<HOUSES>
  <HOUSE OBJECTID="100" OBJECTGUID="h-a" HOUSENUM="10" HOUSETYPE="2"
         ISACTUAL="1" ISACTIVE="1"/>
  <HOUSE OBJECTID="200" OBJECTGUID="h-b" HOUSENUM="5" ADDNUM1="1" ADDTYPE1="1"
         HOUSETYPE="2" ISACTUAL="1" ISACTIVE="1"/>
  <HOUSE OBJECTID="300" OBJECTGUID="h-c" HOUSENUM="7" HOUSETYPE="2"
         ISACTUAL="1" ISACTIVE="1"/>
  <HOUSE OBJECTID="900" OBJECTGUID="h-dead" HOUSENUM="99" HOUSETYPE="2"
         ISACTUAL="0" ISACTIVE="1"/>
</HOUSES>
"""

_APARTMENTS_XML = """<?xml version="1.0" encoding="utf-8"?>
<APARTMENTS>
  <APARTMENT OBJECTID="1001" NUMBER="1" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
  <APARTMENT OBJECTID="1002" NUMBER="2" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
  <APARTMENT OBJECTID="1003" NUMBER="3" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
  <APARTMENT OBJECTID="1004" NUMBER="4" APARTTYPE="2" ISACTUAL="1" ISACTIVE="0"/>
  <APARTMENT OBJECTID="1005" NUMBER="5" APARTTYPE="2" ISACTUAL="0" ISACTIVE="1"/>
  <APARTMENT OBJECTID="2001" NUMBER="1" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
  <APARTMENT OBJECTID="2002" NUMBER="2" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
</APARTMENTS>
"""

# Иерархия: пара (OBJECTID лежит под PARENTOBJID).
_MUN_HIER_XML = """<?xml version="1.0" encoding="utf-8"?>
<ITEMS>
  <ITEM OBJECTID="2" PARENTOBJID="1" ISACTIVE="1"/>
  <ITEM OBJECTID="3" PARENTOBJID="2" ISACTIVE="1"/>
  <ITEM OBJECTID="30" PARENTOBJID="2" ISACTIVE="1"/>
  <ITEM OBJECTID="100" PARENTOBJID="3" ISACTIVE="1"/>
  <ITEM OBJECTID="200" PARENTOBJID="3" ISACTIVE="1"/>
  <ITEM OBJECTID="300" PARENTOBJID="30" ISACTIVE="1"/>
  <ITEM OBJECTID="1001" PARENTOBJID="100" ISACTIVE="1"/>
  <ITEM OBJECTID="1002" PARENTOBJID="100" ISACTIVE="1"/>
  <ITEM OBJECTID="1003" PARENTOBJID="100" ISACTIVE="1"/>
  <ITEM OBJECTID="1004" PARENTOBJID="100" ISACTIVE="1"/>
  <ITEM OBJECTID="2001" PARENTOBJID="200" ISACTIVE="1"/>
  <ITEM OBJECTID="2002" PARENTOBJID="200" ISACTIVE="1"/>
  <ITEM OBJECTID="9999" PARENTOBJID="100" ISACTIVE="0"/>
</ITEMS>
"""


@pytest.fixture
def gar_dir(tmp_path: Path) -> str:
    """Пишет синтетический ГАР-каталог, возвращает путь."""
    (tmp_path / "AS_ADDR_OBJ_20260601_x.XML").write_text(_ADDR_OBJ_XML, encoding="utf-8")
    (tmp_path / "AS_HOUSES_20260601_x.XML").write_text(_HOUSES_XML, encoding="utf-8")
    (tmp_path / "AS_APARTMENTS_20260601_x.XML").write_text(_APARTMENTS_XML, encoding="utf-8")
    (tmp_path / "AS_MUN_HIERARCHY_20260601_x.XML").write_text(_MUN_HIER_XML, encoding="utf-8")
    return str(tmp_path)


def _rows_by_guid(rows: list[gfl.GarHouseRow]) -> dict[str, gfl.GarHouseRow]:
    return {r.house_guid: r for r in rows}


# ─────────────────────────────────────────────────────────────────────────────
# Парс: отдельные файлы
# ─────────────────────────────────────────────────────────────────────────────
def test_parse_addr_objects_filters_inactive(gar_dir: str) -> None:
    paths = gfl._find_files(gar_dir, gfl.ADDR_OBJ_PREFIX)
    addr = gfl.parse_addr_objects(paths)
    assert set(addr.keys()) == {1, 2, 3, 30}  # снесённая (9, ISACTIVE=0) исключена
    assert addr[3].name == "Шаумяна" and addr[3].typename == "ул"


def test_parse_houses_filters_non_actual(gar_dir: str) -> None:
    paths = gfl._find_files(gar_dir, gfl.HOUSES_PREFIX)
    houses = gfl.parse_houses(paths)
    assert set(houses.keys()) == {100, 200, 300}  # 900 (ISACTUAL=0) исключён
    assert houses[200].housenum == "5" and houses[200].addnum1 == "1"


def test_parse_apartment_ids_excludes_inactive_and_non_actual(gar_dir: str) -> None:
    paths = gfl._find_files(gar_dir, gfl.APARTMENTS_PREFIX)
    apts = gfl.parse_apartment_ids(paths)
    # 1004 (ISACTIVE=0) и 1005 (ISACTUAL=0) исключены.
    assert apts == {1001, 1002, 1003, 2001, 2002}


def test_parse_hierarchy_counts_apartments_per_house(gar_dir: str) -> None:
    addr = gfl.parse_addr_objects(gfl._find_files(gar_dir, gfl.ADDR_OBJ_PREFIX))
    houses = gfl.parse_houses(gfl._find_files(gar_dir, gfl.HOUSES_PREFIX))
    apts = gfl.parse_apartment_ids(gfl._find_files(gar_dir, gfl.APARTMENTS_PREFIX))
    hier_paths = gfl._find_files(gar_dir, gfl.MUN_HIER_PREFIX)
    apt_count, child_to_parent = gfl.parse_hierarchy(hier_paths, apts, set(houses), set(addr))
    # ISACTIVE=0 строка иерархии (9999) не учитывается; 1004 не в apartment_ids → не считается.
    assert apt_count.get(100) == 3
    assert apt_count.get(200) == 2
    assert apt_count.get(300, 0) == 0  # дом без квартир
    # child_to_parent заполнен ТОЛЬКО для домов и адр-объектов (не для квартир).
    assert child_to_parent[100] == 3
    assert child_to_parent[3] == 2
    assert 1001 not in child_to_parent


# ─────────────────────────────────────────────────────────────────────────────
# Сборка адреса + строки
# ─────────────────────────────────────────────────────────────────────────────
def test_build_rows_apartment_counts_and_zero_house(gar_dir: str) -> None:
    rows, stats = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
    by_guid = _rows_by_guid(rows)
    assert by_guid["h-a"].flat_count == 3
    assert by_guid["h-b"].flat_count == 2
    assert by_guid["h-c"].flat_count == 0  # дом с 0 квартир → flat_count 0
    assert stats.houses == 3
    assert stats.apartments == 5  # суммарно учтено под домами
    assert stats.rows == 3


def test_multi_segment_address_assembly(gar_dir: str) -> None:
    rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
    a = _rows_by_guid(rows)["h-a"]
    # street_name = непосредственный родитель (улица), «<TYPENAME> <NAME>».
    assert a.street_name == "ул Шаумяна"
    assert a.house_num == "10"
    # full_address — мульти-сегмент регион→город→улица + номер (регион — name-first).
    assert a.full_address == "Свердловская обл, г Екатеринбург, ул Шаумяна, 10"
    # norm_src (вход нормализатора) — «улица, номер».
    assert a.norm_src == "ул Шаумяна, 10"


def test_house_num_with_korpus(gar_dir: str) -> None:
    rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
    b = _rows_by_guid(rows)["h-b"]
    # ADDNUM1=1 ADDTYPE1=1 (Корпус) → «5 к1».
    assert b.house_num == "5 к1"
    assert b.street_name == "ул Шаумяна"


def test_region_and_version_propagated(gar_dir: str) -> None:
    rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
    assert all(r.region_code == "66" for r in rows)
    assert all(r.gar_version == "2026-06-01" for r in rows)


def test_render_house_num_unit() -> None:
    h = gfl.GarHouse(
        object_id=1,
        guid="x",
        housenum="12",
        addnum1="3",
        addnum2="",
        addtype1="2",
        addtype2="",
    )
    assert gfl.render_house_num(h) == "12 стр3"  # ADDTYPE1=2 = Строение


def test_adm_hierarchy_fallback(tmp_path: Path) -> None:
    """Без AS_MUN_HIERARCHY использует AS_ADM_HIERARCHY (fallback)."""
    (tmp_path / "AS_ADDR_OBJ_x.XML").write_text(_ADDR_OBJ_XML, encoding="utf-8")
    (tmp_path / "AS_HOUSES_x.XML").write_text(_HOUSES_XML, encoding="utf-8")
    (tmp_path / "AS_APARTMENTS_x.XML").write_text(_APARTMENTS_XML, encoding="utf-8")
    (tmp_path / "AS_ADM_HIERARCHY_x.XML").write_text(_MUN_HIER_XML, encoding="utf-8")
    rows, _ = gfl.parse_gar_region(str(tmp_path), "66", "2026-06-01")
    assert _rows_by_guid(rows)["h-a"].flat_count == 3


# ─────────────────────────────────────────────────────────────────────────────
# Статические asserts по SQL (psycopg v3, ON CONFLICT, tie-break)
# ─────────────────────────────────────────────────────────────────────────────
_INSERT_SQL = str(gfl._INSERT_SQL.text)
_MATCH_SQL = str(gfl._MATCH_SQL.text)


def test_insert_is_idempotent_on_conflict() -> None:
    flat = re.sub(r"\s+", " ", _INSERT_SQL)
    assert "INSERT INTO gar_house_flats" in flat
    assert "ON CONFLICT (house_guid) DO UPDATE SET" in flat
    # norm_address считает SQL-fn в INSERT (= «UPDATE после insert»).
    assert "tradein_normalize_short_addr(CAST(:norm_src AS text))" in flat


def test_match_distinct_on_tiebreak_max_flat_count() -> None:
    flat = re.sub(r"\s+", " ", _MATCH_SQL)
    # Tie-break: на norm_address — строка с макс flat_count, затем меньший house_guid.
    assert "DISTINCT ON (norm_address)" in flat
    assert "ORDER BY norm_address, flat_count DESC, house_guid" in flat
    # houses-сторона: тот же нормализатор поверх COALESCE(short/full/address).
    assert (
        "tradein_normalize_short_addr( COALESCE(h.short_address, h.full_address, h.address) )"
        in flat
    )
    # Идемпотентность повторного прогона.
    assert "h.gar_flat_count IS DISTINCT FROM gp.flat_count" in flat
    assert "gar_match_method = 'norm_address'" in flat
    # Только дома с известным знаменателем матчим.
    assert "flat_count > 0" in flat


def test_no_psycopg_v3_colon_colon_cast() -> None:
    # Только исполняемый SQL (модульный docstring специально содержит анти-паттерн как памятку).
    assert not re.search(r":\w+::", _INSERT_SQL)
    assert not re.search(r":\w+::", _MATCH_SQL)


def test_upsert_uses_savepoint_per_batch() -> None:
    src = inspect.getsource(gfl.upsert_gar_houses)
    assert "with db.begin_nested():" in src
    # upsert НЕ коммитит сам (коммитит caller).
    assert "db.commit()" not in src


# ─────────────────────────────────────────────────────────────────────────────
# Опциональный real-Postgres тест upsert+match (self-skip без БД)
# ─────────────────────────────────────────────────────────────────────────────
def _live_session() -> Any | None:
    try:
        from sqlalchemy import create_engine, text
        from sqlalchemy.orm import sessionmaker

        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
        if not dsn or "localhost:5432/test" in dsn:
            return None
        engine = create_engine(dsn, future=True)
        conn = engine.connect()
        conn.execute(text("SELECT 1"))
        conn.close()
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


# Тот же нормализатор, что в мигр. 063 — создаём в транзакции теста, откатываем.
_NORMALIZER_FN = """
CREATE OR REPLACE FUNCTION tradein_normalize_short_addr(addr text)
RETURNS text LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$
  SELECT trim(both ' ,.' FROM
    regexp_replace(
      regexp_replace(
        regexp_replace(
          regexp_replace(
            regexp_replace(
              regexp_replace(addr, '^\\s*(?:Россия|РФ|Российская\\s+Федерация)\\s*,\\s*', '', 'i'),
              '^[А-ЯЁа-яё][А-ЯЁа-яё\\s-]+\\s+обл(?:асть|\\.)\\s*,\\s*', '', 'i'),
            '^г\\.?\\s*[А-ЯЁ][А-ЯЁа-яё-]+\\s*,\\s*', '', 'i'),
          '\\s*·\\s*.+$', '', 'i'),
        ',\\s*(?:кв\\.?|корп\\.?|к\\.?|оф\\.?|пом\\.?|подъезд)\\s*\\d+.*$', '', 'i'),
      '\\s{2,}', ' ', 'g'));
$$;
"""


@pytest.mark.skipif(_live_session() is None, reason="нет доступной Postgres test-БД")
def test_upsert_and_match_populates_gar_flat_count(gar_dir: str) -> None:
    """С реальной Postgres: upsert ГАР-строк + match → houses.gar_flat_count проставлен."""
    from sqlalchemy import text

    db = _live_session()
    assert db is not None
    try:
        db.execute(text(_NORMALIZER_FN))
        db.execute(
            text(
                "CREATE TEMP TABLE gar_house_flats ("
                "  house_guid text PRIMARY KEY, object_id bigint, region_code text NOT NULL,"
                "  flat_count int, full_address text, norm_address text, street_name text,"
                "  house_num text, loaded_at timestamptz NOT NULL DEFAULT now(), gar_version text"
                ") ON COMMIT DROP"
            )
        )
        db.execute(
            text(
                "CREATE TEMP TABLE houses ("
                "  id serial PRIMARY KEY, short_address text, full_address text, address text,"
                "  gar_house_guid text, gar_flat_count int, gar_matched_at timestamptz,"
                "  gar_match_method text"
                ") ON COMMIT DROP"
            )
        )
        # Дом «ул Шаумяна, 10» совпадёт с ГАР h-a (flat_count=3); «ул Ленина, 7» — h-c (0) не матч.
        db.execute(
            text(
                "INSERT INTO houses (short_address) VALUES " "('ул Шаумяна, 10'), ('ул Ленина, 7')"
            )
        )

        rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
        upserted = gfl.upsert_gar_houses(db, rows, batch_size=2)
        assert upserted == 3

        # norm_address посчитан SQL-fn = «ул Шаумяна, 10».
        norm = db.execute(
            text("SELECT norm_address FROM gar_house_flats WHERE house_guid='h-a'")
        ).scalar()
        assert norm == "ул Шаумяна, 10"

        matched = gfl.match_houses_to_gar(db, region_code="66")
        assert matched == 1
        # Идемпотентность: повторный матч ничего не трогает.
        assert gfl.match_houses_to_gar(db, region_code="66") == 0

        shaumyana = db.execute(
            text(
                "SELECT gar_flat_count, gar_match_method FROM houses "
                "WHERE short_address='ул Шаумяна, 10'"
            )
        ).fetchone()
        assert shaumyana is not None and shaumyana[0] == 3
        assert shaumyana[1] == "norm_address"
        # Дом без квартир (flat_count=0) НЕ матчится.
        lenina = db.execute(
            text("SELECT gar_flat_count FROM houses WHERE short_address='ул Ленина, 7'")
        ).scalar()
        assert lenina is None
    finally:
        db.rollback()
        db.close()