"""#2189 — канон-ключ адреса не теряет дробь в номере дома («Сулимова, 3/2» ≠ «Сулимова, 32»).

Проверяется функция, которую поставили миграции (322), а не её Python-копия, и оба её
потребителя: матч ГАР↔houses (gar_flats_loader) и схлопывание дублей (house_dedup_merge).
Без БД тесты пропускаются (записи в tests/skip_allowlist.txt), в CI идут на схеме из data/sql.
"""

from __future__ import annotations

import os
from pathlib import Path
from typing import Any

import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from sqlalchemy import text

from app.services import gar_flats_loader as gfl
from app.services import house_dedup_merge as hdm


def _live_session() -> Any | None:
    try:
        from sqlalchemy import create_engine
        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)
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


needs_db = pytest.mark.skipif(_live_session() is None, reason="нет доступной Postgres test-БД")


@needs_db
def test_real_canon_keeps_slash_between_digits() -> None:
    db = _live_session()
    assert db is not None

    def canon(s: str) -> str:
        return db.execute(text("SELECT tradein_canon_addr(:s)"), {"s": s}).scalar()

    try:
        # живые пары с прода (ревизия DaData 02.07): это разные дома
        assert canon("Екатеринбург, ул. Сулимова, 3/2") == "сулимова3/2"
        assert canon("Екатеринбург, ул. Сулимова, 32") == "сулимова32"
        assert canon("улица Краснолесья, 10/7") == "краснолесья10/7"
        assert canon("ул. Краснолесья,107") == "краснолесья107"
        assert canon("улица Машиностроителей, 4/1") != canon("ул. Машиностроителей, 41")
        # пробелы вокруг дроби и формат ГАР дают тот же ключ, что и слитная запись
        assert canon("ул. Сулимова, 3 / 2") == canon(
            "р-н Кировский, мкр. Пионерский, ул. Сулимова,3/2"
        )
        # «/» не между цифрами — как раньше: угловой адрес и литера перед дробью
        assert canon(
            "Екатеринбург, Втузгородок м-н, улица Гагарина, 27 / Первомайская улица, 81"
        ) == ("гагарина27первомайская81")
        assert canon("ул. 8 Марта,204Г/1") == "8марта204г1"
        # остальная нормализация не задета
        assert canon("ул. Шаумяна,20") == canon("ул. Шаумяна, 20") == "шаумяна20"
        assert (
            canon("Екатеринбург, ул. Евгения Савкова, 17Ар-н Академический") == "евгениясавкова17а"
        )
    finally:
        db.rollback()
        db.close()


@needs_db
def test_real_gar_match_picks_the_house_with_the_same_number() -> None:
    """Прод: «Сулимова 32» получал ГАР-дом «3/2» (414 квартир) — max(flat_count) в общем ключе."""
    db = _live_session()
    assert db is not None
    try:
        db.execute(
            text(
                "INSERT INTO gar_house_flats "
                "(house_guid, region_code, flat_count, full_address, norm_address) VALUES "
                "('g2189-3-2', '66', 414, 'г. Екатеринбург, ул. Сулимова2189, 3/2', "
                " 'ул. Сулимова2189, 3/2'),"
                "('g2189-32', '66', 120, 'г. Екатеринбург, ул. Сулимова2189, 32', "
                " 'ул. Сулимова2189, 32')"
            )
        )
        # region_code обязателен: с #2583 H5 матчер даёт guid только ГАР-строке региона дома,
        # дом без региона не матчится вовсе.
        db.execute(
            text(
                "INSERT INTO houses (id, source, ext_house_id, url, address, region_code) VALUES "
                "(900189, 'avito', 'EXT-2189-32', 'u', 'Екатеринбург, ул. Сулимова2189, 32', 66),"
                "(900190, 'cian', 'EXT-2189-3-2', 'u', 'ул. Сулимова2189, 3 / 2', 66)"
            )
        )

        gfl.match_houses_to_gar(db, region_code="66")

        got = db.execute(
            text(
                "SELECT id, gar_house_guid, gar_flat_count, gar_match_method FROM houses "
                "WHERE id IN (900189, 900190) ORDER BY id"
            )
        ).all()
        assert [tuple(r) for r in got] == [
            (900189, "g2189-32", 120, "canon_addr"),
            (900190, "g2189-3-2", 414, "canon_addr"),
        ]
    finally:
        db.rollback()
        db.close()


@needs_db
def test_real_dedup_does_not_merge_slash_and_plain_number() -> None:
    """Дома «3/2» и «32» без ФИАС в одной точке — разные здания, в один кластер не попадают."""
    db = _live_session()
    assert db is not None
    try:
        db.execute(
            text(
                "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon) VALUES "
                "(900189, 'avito', 'EXT-2189-32', 'u', 'ул. Сулимова2189, 32', 56.84, 60.6),"
                "(900190, 'cian', 'EXT-2189-3-2', 'u', 'ул. Сулимова2189, 3/2', 56.84009, 60.6)"
            )
        )
        db.commit()

        hdm.merge_duplicate_houses(db, dry_run=False)

        ids = {r.id for r in db.execute(text("SELECT id FROM houses WHERE id IN (900189, 900190)"))}
        assert ids == {900189, 900190}
    finally:
        db.rollback()
        for sql in (
            "DELETE FROM house_sources WHERE house_id IN (900189, 900190)",
            "DELETE FROM house_address_aliases WHERE house_id IN (900189, 900190)",
            "DELETE FROM house_merge_log WHERE loser_id IN (900189, 900190)",
            "DELETE FROM houses WHERE id IN (900189, 900190)",
        ):
            db.execute(text(sql))
        db.commit()
        db.close()


_MIGRATION = (
    Path(__file__).resolve().parents[1]
    / "data"
    / "sql"
    / "322_canon_keep_slash_in_house_number.sql"
)


def test_migration_322_resets_only_matches_held_by_the_lost_slash() -> None:
    """Матчер пары не снимает: «4/1», пришитый к ГАР «41», без миграции так и остался бы с ним.

    Прогон настоящего файла миграции; houses и gar_house_flats затенены временными таблицами
    (pg_temp первым в search_path), реальные таблицы не трогаются.
    """
    dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
    if not dsn or "localhost:5432/test" in dsn:
        pytest.skip("нет доступной Postgres test-БД")
    import psycopg

    with psycopg.connect(
        dsn.replace("postgresql+psycopg://", "postgresql://"), autocommit=True
    ) as c:
        c.execute(
            "CREATE TEMP TABLE gar_house_flats (house_guid text PRIMARY KEY, region_code text,"
            " flat_count int, norm_address text)"
        )
        c.execute(
            "CREATE TEMP TABLE houses (id int 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)"
        )
        c.execute(
            "INSERT INTO gar_house_flats VALUES "
            "('g-3-2', '66', 414, 'ул. Сулимова, 3/2'),"
            "('g-41', '66', 92, 'ул. Машиностроителей, 41'),"
            "('g-58', '66', 249, 'ул. Абрикосовая, 58')"
        )
        c.execute(
            "INSERT INTO houses (id, short_address, gar_house_guid, gar_flat_count, gar_matched_at,"
            " gar_match_method) VALUES "
            # держится только на потерянной дроби → сброс
            "(1, 'ул. Сулимова, 32', 'g-3-2', 414, now(), 'canon_addr'),"
            "(2, 'улица Машиностроителей, 4/1', 'g-41', 92, now(), 'canon_addr'),"
            # верная пара с дробью
            "(3, 'Сулимова ул.,3/2', 'g-3-2', 414, now(), 'canon_addr'),"
            # не канон-метод
            "(4, 'улица Машиностроителей, 4/1', 'g-41', 92, now(), 'dadata_fias_revision'),"
            # ключ разошёлся не из-за дроби (адрес сменился после матча)
            "(5, 'жилой комплекс Совушки, Абрикосовая ул.,58', 'g-58', 249, now(), 'canon_addr')"
        )

        c.execute(_MIGRATION.read_text(encoding="utf-8"))

        rows = c.execute(
            "SELECT id, gar_house_guid, gar_flat_count, gar_matched_at IS NOT NULL,"
            " gar_match_method FROM houses ORDER BY id"
        ).fetchall()
        assert rows == [
            (1, None, None, False, None),
            (2, None, None, False, None),
            (3, "g-3-2", 414, True, "canon_addr"),
            (4, "g-41", 92, True, "dadata_fias_revision"),
            (5, "g-58", 249, True, "canon_addr"),
        ]