"""#3256: сделочная сторона НЕ фильтруется по `deals.rooms`.

`deals.rooms` Росреестра — синтетика из площади: import-rosreestr.sh пишет туда тот
же CASE 30/44/62/85, что `asking_to_sold_ratio.area_bucket`. Прод-замер 2026-09-11:
321 559 из 321 560 сделок удовлетворяют `rooms == area_bucket(area_m2)`,
`max(rooms) = 4`. Значит `d.rooms = X` тождественно `d.area_m2 ∈ [граница_X,
граница_X+1)` — ЛЮБОЙ предикат по rooms был вторым, ступенчатым фильтром по площади
поверх полосы `area_m2 BETWEEN :area_min AND :area_max`, стоящей рядом.

Прод-замер по 1179 реальным запросам (trade_in_estimates, 2026-09-12) — какая доля
полосы ±15% переживает предикат:
    d.rooms = комнаты клиента   → медиана 77.8%, у 180 запросов полоса вырезана
                                  ЦЕЛИКОМ (пересечение пусто ⇒ коридора нет никогда);
    d.rooms = area_bucket(area) → медиана 90.0%, пустых нет, но у 902 из 1179 полоса
                                  всё ещё усечена границей бакета;
    без предиката               → 100% по построению.

Тесты ПО ЗНАЧЕНИЮ: фейковая сессия перехватывает SQL и bind-параметры, уходящие в
Postgres. На origin/main в каждом из трёх мест есть и предикат `rooms = ...`, и
bind-параметр `rooms` — тесты краснеют ФАКТОМ фильтра, а не отсутствием атрибута.
"""

from __future__ import annotations

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

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

from app.api.v1.trade_in import get_street_deals
from app.services.estimator import _fetch_deals, _fetch_dkp_corridor
from app.tasks.asking_to_sold_ratio import area_bucket

_EKB_ADDRESS = "Екатеринбург, улица Ленина, 5"

# Путь к импортёру: tests/ → backend/ → tradein-mvp/ → deploy/
_IMPORTER = Path(__file__).resolve().parents[2] / "deploy" / "import-rosreestr.sh"


class _Result:
    """Пустой результат: и `.mappings().all()`, и итерация дают ноль строк."""

    def mappings(self) -> _Result:
        return self

    def all(self) -> list[Any]:
        return []

    def __iter__(self):
        return iter(())


class _RecordingDB:
    """Сессия-перехватчик: копит (sql, params) каждого execute, строк не отдаёт."""

    def __init__(self) -> None:
        self.calls: list[tuple[str, dict[str, Any]]] = []

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _Result:
        self.calls.append((str(getattr(stmt, "text", stmt)), dict(params or {})))
        return _Result()


def _rooms_predicates(db: _RecordingDB) -> list[str]:
    """Строки SQL, где по rooms стоит СРАВНЕНИЕ (предикат), из всех запросов к deals.

    Ищем `rooms =` / `rooms IN` / `rooms BETWEEN` (с необязательным алиасом `d.`).
    Колонка `rooms` в SELECT-списке витрины — не предикат и сюда не попадает.
    """
    rx = re.compile(r"\b(?:\w+\.)?rooms\s*(?:=|<|>|IN\b|BETWEEN\b)", re.IGNORECASE)
    hits: list[str] = []
    for sql, _params in db.calls:
        if "deals" not in sql.lower():
            continue
        hits += [line.strip() for line in sql.splitlines() if rx.search(line)]
    return hits


def _rooms_binds(db: _RecordingDB) -> list[Any]:
    """Значения bind-параметра `rooms`, ушедшие в запросы к deals (должно быть пусто)."""
    return [p["rooms"] for sql, p in db.calls if "rooms" in p and "deals" in sql.lower()]


def _area_bands(db: _RecordingDB) -> list[tuple[float, float]]:
    return [
        (p["area_min"], p["area_max"])
        for sql, p in db.calls
        if "area_min" in p and "deals" in sql.lower()
    ]


# ── ДКП-коридор (клампит headline, ставит radius-floor, кормит deals-fallback) ──


def test_dkp_corridor_has_no_rooms_predicate() -> None:
    """«3 комнаты, 100 м²»: в deals уходит ТОЛЬКО полоса площади, без rooms.

    С предикатом `rooms=3` (≡ площадь 62-85) пересечение с полосой [85, 115] ПУСТО —
    коридор не строился вовсе. С предикатом `rooms=area_bucket(100)=4` (≡ площадь
    ≥85) полоса усекалась до [85, 115] (нижние 8.5 м² отрезаны без причины).
    """
    db = _RecordingDB()
    _fetch_dkp_corridor(db, address=_EKB_ADDRESS, rooms=3, area=100.0, city="екатеринбург")
    assert db.calls, "запрос к deals вообще не ушёл"
    assert _rooms_predicates(db) == []
    assert _rooms_binds(db) == []


def test_dkp_corridor_area_band_is_the_only_similarity_filter() -> None:
    """Полоса ±15% доходит до SQL целиком И остаётся ЕДИНСТВЕННЫМ фильтром похожести.

    Вторая половина утверждения — та, что краснеет: на origin/main рядом с полосой
    стоит `d.rooms = ...`, т.е. фильтров по площади там два (полоса и бакет), и
    итоговое окно — их пересечение.
    """
    db = _RecordingDB()
    _fetch_dkp_corridor(db, address=_EKB_ADDRESS, rooms=2, area=50.0, city="екатеринбург")
    assert _area_bands(db) == [(50.0 * 0.85, 50.0 * 1.15)]
    assert _rooms_predicates(db) == []


def test_dkp_corridor_five_room_client_is_reachable() -> None:
    """5+ комнат: в deals.rooms значения >4 нет, предикат `rooms=5` не нашёл бы ничего."""
    db = _RecordingDB()
    _fetch_dkp_corridor(db, address=_EKB_ADDRESS, rooms=5, area=140.0, city="екатеринбург")
    assert _rooms_predicates(db) == []
    assert _rooms_binds(db) == []


def test_dkp_corridor_query_does_not_depend_on_client_rooms() -> None:
    """Один метраж — один и тот же SQL и параметры, какой бы комнатность ни назвали.

    Прямое свойство синтетической колонки: `deals.rooms` = f(area_m2), значит выборка
    сделок не имеет права зависеть от комнатности клиента ни через что.
    """
    seen: set[tuple] = set()
    for rooms in (1, 2, 3, 4, 5, 6):
        db = _RecordingDB()
        _fetch_dkp_corridor(db, address=_EKB_ADDRESS, rooms=rooms, area=95.0, city="екатеринбург")
        seen.add(tuple((sql, tuple(sorted(p.items()))) for sql, p in db.calls))
    assert len(seen) == 1, "выборка сделок всё ещё зависит от комнатности клиента"


# ── Радиусные сделки (блок «реальные сделки рядом» + expected_sold) ───────────


def test_fetch_deals_has_no_rooms_predicate() -> None:
    """`_fetch_deals` (радиус) — тот же ключ: только полоса площади."""
    db = _RecordingDB()
    _fetch_deals(db, lat=56.83, lon=60.6, area=130.0, radius_m=1000)
    assert _rooms_predicates(db) == []
    assert _rooms_binds(db) == []
    assert _area_bands(db) == [(130.0 * 0.85, 130.0 * 1.15)]


# ── Витрина «сделки на улице» ────────────────────────────────────────────────


def test_street_deals_has_no_rooms_predicate() -> None:
    """GET /street-deals: «3 комнаты, 92 м²» — витрина не опустошается предикатом."""
    db = _RecordingDB()
    get_street_deals(address=_EKB_ADDRESS, area_m2=92.0, rooms=3, db=db)  # type: ignore[arg-type]
    assert _rooms_predicates(db) == []
    assert _rooms_binds(db) == []


# ── ЯКОРЬ: импортёр всё ещё пишет в deals.rooms бакет площади ─────────────────


def _parse_importer_case() -> list[tuple[float, int]]:
    """Границы CASE `<area → rooms` из import-rosreestr.sh + значение ветки ELSE.

    Возвращает [(граница, значение), ...] + [(inf, ELSE)]. Падает, если CASE
    исчез/переименован — это тоже сигнал «импортёр переписали, вернись в #3256».
    """
    src = _IMPORTER.read_text(encoding="utf-8")
    m = re.search(r"CASE\s+(WHEN\s+area\s*<.+?ELSE\s+\d+\s+END)\s+AS\s+rooms", src, re.S)
    assert m, "CASE ... AS rooms не найден в import-rosreestr.sh — см. якорь #3256"
    body = m.group(1)
    edges = [
        (float(a), int(b))
        for a, b in re.findall(r"WHEN\s+area\s*<\s*([\d.]+)\s+THEN\s+(\d+)", body)
    ]
    else_m = re.search(r"ELSE\s+(\d+)\s+END", body)
    assert else_m
    return [*edges, (float("inf"), int(else_m.group(1)))]


def test_importer_case_still_matches_area_bucket() -> None:
    """ЯКОРЬ #3256: границы импортёра совпадают с area_bucket() — иначе правка мертва.

    Ни один другой тест не покраснеет, если импортёр начнёт писать НАСТОЯЩУЮ
    комнатность: все потребители просто продолжат молча ключеваться площадью (точнее,
    перестанут ключеваться вовсе, а area-полоса начнёт означать не то). Этот тест —
    единственное место, которое свяжет два файла и заставит вернуться в #3256.
    """
    edges = _parse_importer_case()
    assert [e for e, _v in edges] == [30.0, 44.0, 62.0, 85.0, float("inf")], (
        "границы CASE в import-rosreestr.sh разошлись с area_bucket() — см. якорь #3256"
    )
    assert [v for _e, v in edges] == [0, 1, 2, 3, 4]


def test_importer_case_and_area_bucket_agree_pointwise() -> None:
    """Та же сверка по значениям: на границах и между ними CASE == area_bucket()."""
    edges = _parse_importer_case()

    def importer_rooms(area: float) -> int:
        for edge, value in edges:
            if area < edge:
                return value
        raise AssertionError("ELSE-ветка не покрыла area")

    for area in (0.1, 18.0, 29.9, 30.0, 43.9, 44.0, 61.9, 62.0, 84.9, 85.0, 120.0, 200.0):
        assert importer_rooms(area) == area_bucket(area), f"расхождение на area={area}"