"""Тесты импорта московского сырья `msk_raw` в listings.

Изоляция — как у соседей (tests/tasks/test_backfill_house_coords_from_listings.py):
DATABASE_URL проставляется до импорта модуля, в сеть и в прод-БД тест не ходит.
Сессия — заглушка, отдающая заранее заготовленные payload'ы; писатель
(`save_listings`) подменяется monkeypatch'ем, так что проверяется ровно наш отбор
и пересчёт метки, а не поведение кита.

Главные риски задачи, которые тут и караулятся:
  * в регион 77 уедет Московская область (адрес без округа);
  * артефакт парсера («Екатеринбург (Cian)») растворится в «не Москва»;
  * метка кита 'novostroyki' будет скопирована как есть и выкинет 29 000 карточек
    из подбора аналогов (гвард estimator'а берёт только NULL/'vtorichka').
"""

from __future__ import annotations

import os
from typing import Any

import pytest

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

from app.tasks import msk_raw_import as mod


def _payload(**over: Any) -> dict[str, Any]:
    """Минимальный сериализованный ScrapedLot, как он лежит в msk_raw.payload."""
    base: dict[str, Any] = {
        "source": "cian",
        "source_url": "https://www.cian.ru/sale/flat/330200428/",
        "source_id": "330200428",
        "address": "ЦАО, р-н Арбат, улица Арбат, 10",
        "price_rub": 25_000_000,
        "rooms": 2,
        "area_m2": 55.0,
        "listing_segment": "novostroyki",  # метка кита — её и пересчитываем
        "raw_payload": {"is_from_developer": False},
    }
    base.update(over)
    return base


# ── отбор по адресу ──────────────────────────────────────────────────────────


@pytest.mark.parametrize("okrug", mod.MOSCOW_OKRUGS)
def test_every_okrug_prefix_is_moscow(okrug: str) -> None:
    assert mod.is_moscow_address(f"{okrug}, р-н Хамовники, улица Льва Толстого, 16")


def test_oblast_address_without_okrug_is_not_moscow() -> None:
    """Адрес без округа — Московская область (регион 50), его в импорт не берём."""
    assert not mod.is_moscow_address("Химки, микрорайон Новогорск, Ивановская улица, 5")
    assert not mod.is_moscow_address("городской округ Мытищи, Олимпийский проспект, 29")


def test_okrug_must_be_a_whole_token() -> None:
    """«ЗАО» в начале улицы — не округ; иначе область просочилась бы в регион 77."""
    assert not mod.is_moscow_address("ЗАОзёрная улица, 25")


def test_ekaterinburg_artifact_is_its_own_reason() -> None:
    assert mod.is_artifact_address("Екатеринбург (Cian)")
    assert not mod.is_moscow_address("Екатеринбург (Cian)")


# ── пересчёт listing_segment ─────────────────────────────────────────────────


def test_segment_recomputed_to_vtorichka_when_developer_false() -> None:
    lot = mod.build_lot(_payload())
    assert lot.listing_segment == "vtorichka"


def test_segment_recomputed_to_vtorichka_when_flag_absent() -> None:
    """Отсутствие флага — не признак новостройки (у 31 464 карточек его нет вовсе)."""
    lot = mod.build_lot(_payload(raw_payload={}))
    assert lot.listing_segment == "vtorichka"
    assert mod.build_lot(_payload(raw_payload=None)).listing_segment == "vtorichka"


def test_segment_is_novostroyki_only_when_developer_sells() -> None:
    lot = mod.build_lot(_payload(raw_payload={"is_from_developer": True}))
    assert lot.listing_segment == "novostroyki"


def test_unknown_payload_keys_are_dropped() -> None:
    """В payload может приехать ключ, которого в модели нет — это не должно ронять импорт."""
    lot = mod.build_lot(_payload(some_future_field="x"))
    assert lot.source_id == "330200428"


# ── прогон целиком ───────────────────────────────────────────────────────────


class _FakeResult:
    def __init__(self, rows: list[dict[str, Any]]) -> None:
        self._rows = rows

    def mappings(self) -> _FakeResult:
        return self

    def all(self) -> list[dict[str, Any]]:
        return self._rows

    def scalar(self) -> Any:
        """Проба существования кэш-таблицы: None = таблицы ещё нет."""
        return self._rows[0] if self._rows else None


class _FakeSession:
    """Отдаёт строки постранично по тому же keyset-условию, что и настоящая вью."""

    def __init__(self, rows: list[dict[str, Any]]) -> None:
        self.rows = rows
        self.commits = 0

    def execute(self, _sql: Any, params: dict[str, Any] | None = None) -> _FakeResult:
        if not params or "after" not in params:
            return _FakeResult([])  # DDL кэша пред-геокода — строк не отдаёт
        after, limit = params["after"], params["limit"]
        return _FakeResult([r for r in self.rows if r["id"] > after][:limit])

    def commit(self) -> None:
        self.commits += 1


def _rows() -> list[dict[str, Any]]:
    return [
        {"id": 1, "payload": _payload(source_id="1")},
        {"id": 2, "payload": _payload(source_id="2", address="Химки, Ивановская улица, 5")},
        {"id": 3, "payload": _payload(source_id="3", address="Екатеринбург (Cian)")},
        {"id": 4, "payload": _payload(source_id="4", address="СВАО, улица Милашенкова, 7")},
        {"id": 5, "payload": _payload(source_id="5", price_rub=0)},  # невалидный: цена > 0
    ]


@pytest.fixture()
def captured(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
    calls: list[dict[str, Any]] = []

    def _fake_save(db: Any, lots: list[Any], **kwargs: Any) -> tuple[int, int]:
        calls.append({"lots": lots, **kwargs})
        return len(lots), 0

    monkeypatch.setattr(mod, "save_listings", _fake_save)
    return calls


def test_only_moscow_rows_reach_the_writer(captured: list[dict[str, Any]]) -> None:
    db = _FakeSession(_rows())
    counters = mod.import_msk_raw(db, batch_size=10)

    assert [lot.source_id for lot in captured[0]["lots"]] == ["1", "4"]
    assert counters.read == 5
    assert counters.selected == 2
    assert counters.skipped_not_moscow == 1
    assert counters.skipped_artifact == 1
    assert counters.skipped_invalid == 1
    assert counters.check(), "счётчики обязаны сходиться с прочитанным"


def test_writer_gets_region_77_and_city(captured: list[dict[str, Any]]) -> None:
    mod.import_msk_raw(_FakeSession(_rows()), batch_size=10)
    assert captured[0]["region_code"] == 77
    assert captured[0]["city"] == "Москва"
    assert all(lot.listing_segment == "vtorichka" for lot in captured[0]["lots"])


def test_commit_per_batch(captured: list[dict[str, Any]]) -> None:
    """Обрыв на середине не должен терять уже перелитые батчи."""
    db = _FakeSession(_rows())
    mod.import_msk_raw(db, batch_size=1)
    assert db.commits == 2  # ровно столько батчей содержали московские строки


def test_dry_run_writes_nothing(captured: list[dict[str, Any]]) -> None:
    db = _FakeSession(_rows())
    counters = mod.import_msk_raw(db, batch_size=10, dry_run=True)
    assert captured == []
    assert db.commits == 0
    assert counters.selected == 2 and counters.written == 0


def test_limit_caps_rows_read(captured: list[dict[str, Any]]) -> None:
    counters = mod.import_msk_raw(_FakeSession(_rows()), batch_size=2, limit=3)
    assert counters.read == 3


def test_avito_refuses_to_write_without_explicit_opt_in(captured: list[dict[str, Any]]) -> None:
    """У Авито в адресе нет города, а координат нет ни у одной карточки — молча
    залить область в регион 77 нельзя."""
    rows = [{"id": 1, "payload": _payload(source="avito", address="Варшавское ш.,62к1")}]
    with pytest.raises(SystemExit):
        mod.import_msk_raw(_FakeSession(rows), source="avito", batch_size=10)

    counters = mod.import_msk_raw(
        _FakeSession(rows), source="avito", batch_size=10, allow_unfiltered=True
    )
    assert counters.selected == 1  # префиксный фильтр к Авито не применяется


# ── пред-геокод Авито ────────────────────────────────────────────────────────
#
# Риски этой ветки, которые тут и караулятся:
#   * регион возьмут из слага, а не из ответа — Новая Москва (Троицк/Щербинка/
#     Коммунарка) уедет в область, хотя это регион 77;
#   * область (регион 50) просочится в listings;
#   * слаг выпадет из ключа кэша — адреса-тёзки из разных муниципалитетов
#     («Центральная ул.,5») схлопнутся в один дом;
#   * повторный прогон заново потратит квоту по уже разрешённым адресам.


def _avito(**over: Any) -> dict[str, Any]:
    base = _payload(
        source="avito",
        source_url="https://www.avito.ru/moskva/kvartiry/2-k._kvartira_56_m_56_et._46317006",
        address="Варшавское ш.,62к1",
    )
    base.update(over)
    return base


class _Sugg:
    """Минимальный DadataSuggestion — нужны только поля, которые читает отбор."""

    def __init__(self, kind: str, lat: float | None, lon: float | None, kladr: str | None) -> None:
        self.kind, self.lat, self.lon, self.kladr_id = kind, lat, lon, kladr
        self.unrestricted_value = "г Москва, ш Варшавское, д 62 к 1"


def test_slug_comes_from_source_url() -> None:
    assert mod.avito_city_slug(_avito()) == "moskva"
    himki = _avito(source_url="https://www.avito.ru/himki/kvartiry/1")
    assert mod.avito_city_slug(himki) == "himki"
    assert mod.avito_city_slug(_avito(source_url="")) is None
    assert mod.avito_city_slug({}) is None


def test_cache_key_separates_same_address_in_different_towns() -> None:
    """1110 адресов из 21 570 живут под несколькими слагами — это разные дома."""
    assert mod.geo_cache_key("Центральная ул.,5", "moskva") != mod.geo_cache_key(
        "Центральная ул.,5", "himki"
    )
    # нормализация адреса общая с геокодером — регистр и пробелы ключ не множат
    assert mod.geo_cache_key("  Варшавское Ш.,62к1 ", "moskva") == mod.geo_cache_key(
        "варшавское ш.,62к1", "moskva"
    )


def test_moscow_slug_narrows_constraint_others_keep_both_regions() -> None:
    """Не-московский слаг может оказаться Новой Москвой — оба региона в констрейнте."""
    assert mod._dadata_regions_for("moskva") == [mod.DADATA_MOSCOW_REGION]
    assert mod._dadata_regions_for("moskovskaya_oblast_troitsk") == [
        mod.DADATA_MOSCOW_REGION,
        mod.DADATA_OBLAST_REGION,
    ]
    assert mod._dadata_regions_for(None) == [
        mod.DADATA_MOSCOW_REGION,
        mod.DADATA_OBLAST_REGION,
    ]


def test_region_is_read_from_the_answer_not_from_the_slug() -> None:
    """КЛАДР ответа 77… → регион 77, даже если слаг был подмосковным."""
    point = mod._point_from_suggestions([_Sugg("house", 55.6, 37.6, "77000000000000000")])
    assert point is not None and point.region_code == 77
    oblast = mod._point_from_suggestions([_Sugg("house", 55.9, 37.4, "50000000000000000")])
    assert oblast is not None and oblast.region_code == 50


def test_only_house_level_answers_become_points() -> None:
    """Улица/город дают точку в середине улицы — для radius-подбора это хуже, чем ничего."""
    assert mod._point_from_suggestions([_Sugg("street", 55.6, 37.6, "77000000000000000")]) is None
    assert mod._point_from_suggestions([_Sugg("house", None, None, "77000000000000000")]) is None
    # чужой регион (констрейнт пробит) — тоже не берём
    assert mod._point_from_suggestions([_Sugg("house", 56.8, 60.6, "66000000000000000")]) is None


def test_budget_stops_asking_instead_of_failing() -> None:
    budget = mod.GeocodeBudget(remaining=2)
    assert budget.take(5) == 2 and budget.exhausted and budget.spent == 2
    assert budget.take(1) == 0  # квота кончилась — просто больше не спрашиваем


def _geo_rows() -> list[dict[str, Any]]:
    return [
        {"id": 1, "payload": _avito(source_id="1")},  # Москва
        {
            "id": 2,
            "payload": _avito(
                source_id="2",
                source_url="https://www.avito.ru/himki/kvartiry/2",
                address="Ивановская ул.,5",
            ),
        },  # область
        {
            "id": 3,
            "payload": _avito(source_id="3", address="Мкр-н имени В.Н. Махалина, 33"),
        },  # не разрешился
    ]


@pytest.fixture()
def geocoded(monkeypatch: pytest.MonkeyPatch) -> list[list[dict[str, Any]]]:
    """Подменяет резолвер целиком — в сеть и в кэш-таблицу тест не ходит."""
    seen: list[list[dict[str, Any]]] = []

    def _fake_resolve(db: Any, rows: list[Any], budget: Any, **kwargs: Any) -> dict[str, Any]:
        seen.append(list(rows))
        return {
            mod.geo_cache_key("Варшавское ш.,62к1", "moskva"): mod.GeoPoint(
                55.6412, 37.6089, 77, "г Москва, ш Варшавское, д 62 к 1"
            ),
            mod.geo_cache_key("Ивановская ул.,5", "himki"): mod.GeoPoint(
                55.8970, 37.4297, 50, "Московская обл, г Химки, ул Ивановская, д 5"
            ),
        }

    monkeypatch.setattr(mod, "_resolve_points", _fake_resolve)
    return seen


def test_geocode_lets_avito_through_without_allow_unfiltered(
    captured: list[dict[str, Any]], geocoded: list[list[dict[str, Any]]]
) -> None:
    """Пред-геокод — штатный путь: аварийный флаг больше не нужен."""
    counters = mod.import_msk_raw(
        _FakeSession(_geo_rows()), source="avito", batch_size=10, geocode=True
    )
    assert [lot.source_id for lot in captured[0]["lots"]] == ["1"]
    assert counters.skipped_oblast == 1
    assert counters.skipped_ungeocoded == 1
    assert counters.check(), "счётчики обязаны сходиться с прочитанным"


def test_coordinates_reach_the_writer(
    captured: list[dict[str, Any]], geocoded: list[list[dict[str, Any]]]
) -> None:
    """Ради geom всё и затевалось — карточка не должна лечь без координат."""
    mod.import_msk_raw(_FakeSession(_geo_rows()), source="avito", batch_size=10, geocode=True)
    lot = captured[0]["lots"][0]
    assert (lot.lat, lot.lon) == (55.6412, 37.6089)
    assert lot.geo_precision == "house"
    assert captured[0]["region_code"] == 77


def test_oblast_is_never_written_as_region_77(
    captured: list[dict[str, Any]], geocoded: list[list[dict[str, Any]]]
) -> None:
    mod.import_msk_raw(_FakeSession(_geo_rows()), source="avito", batch_size=10, geocode=True)
    written = [lot.source_id for call in captured for lot in call["lots"]]
    assert "2" not in written, "химкинская карточка не должна уехать в регион 77"


def test_avito_without_geocode_demands_opt_in(captured: list[dict[str, Any]]) -> None:
    rows = [{"id": 1, "payload": _avito()}]
    with pytest.raises(SystemExit):
        mod.import_msk_raw(_FakeSession(rows), source="avito", batch_size=10)


def test_geocode_is_refused_for_cian(captured: list[dict[str, Any]]) -> None:
    """У Циана город виден по префиксу округа даром — квоту на него не тратим."""
    with pytest.raises(SystemExit):
        mod.import_msk_raw(_FakeSession(_rows()), source="cian", batch_size=10, geocode=True)


def test_dry_run_survives_missing_geocode_cache(monkeypatch: pytest.MonkeyPatch) -> None:
    """Первый `--dry-run` идёт по проду, где кэш-таблицы ещё нет.

    Таблицу создаёт только боевой прогон, поэтому SELECT по ней в сухом ронял
    ровно ту репетицию, ради которой сухой прогон и существует (замер на проде
    11.09: UndefinedTable msk_raw.avito_geocode). Резолвер обязан молча считать
    кэш пустым и идти к DaData за остатком.
    """

    class _NoCacheSession:
        def __init__(self) -> None:
            self.cache_reads = 0

        def execute(self, sql: Any, params: dict[str, Any] | None = None) -> _FakeResult:
            if "to_regclass" in str(sql):
                return _FakeResult([])  # таблицы нет
            self.cache_reads += 1
            raise AssertionError("к отсутствующей кэш-таблице ходить нельзя")

    point = mod.GeoPoint(55.6412, 37.6089, 77, "г Москва, ш Варшавское, д 62 к 1")

    async def _fake_geocode(misses: list[Any]) -> dict[str, Any]:
        return {misses[0][0]: point}

    monkeypatch.setattr(mod, "_geocode_many", _fake_geocode)

    db = _NoCacheSession()
    points = mod._resolve_points(
        db, [{"id": 1, "payload": _avito()}], mod.GeocodeBudget(10), dry_run=True
    )

    assert db.cache_reads == 0
    assert list(points.values()) == [point]


# ── Яндекс: город из второго компонента адреса ───────────────────────────────


def _yandex(**over: Any) -> dict[str, Any]:
    base = _payload(
        source="yandex",
        source_url="https://realty.yandex.ru/offer/1234567890123456789/",
        source_id="1234567890123456789",
        address="Россия, Москва, Коробейников переулок, 1",
        lat=55.737312,
        lon=37.601006,
    )
    base.update(over)
    return base


def test_yandex_moscow_is_recognised_without_geocode() -> None:
    assert mod.is_moscow_yandex_address("Россия, Москва, Коробейников переулок, 1")
    assert mod.is_moscow_yandex_address("Россия, Москва, Зеленоград, корпус 1130")


def test_yandex_oblast_is_not_moscow() -> None:
    """Ровно те два значения, что встречаются в корпусе; область писать нельзя."""
    oblast = "Россия, Московская область, Химки, Ивановская улица, 5"
    assert not mod.is_moscow_yandex_address(oblast)
    assert not mod.is_moscow_yandex_address("Москва, Коробейников переулок, 1")  # без страны
    assert not mod.is_moscow_yandex_address("")
    assert not mod.is_moscow_yandex_address(None)


def test_yandex_oblast_rows_do_not_reach_the_writer(captured: list[dict[str, Any]]) -> None:
    rows = [
        {"id": 1, "payload": _yandex(source_id="1")},
        {
            "id": 2,
            "payload": _yandex(
                source_id="2", address="Россия, Московская область, Химки, Ивановская улица, 5"
            ),
        },
    ]
    counters = mod.import_msk_raw(_FakeSession(rows), source="yandex", batch_size=10)
    assert [lot.source_id for lot in captured[0]["lots"]] == ["1"]
    assert counters.skipped_not_moscow == 1


def test_yandex_keeps_its_own_coordinates(captured: list[dict[str, Any]]) -> None:
    """Координаты у Яндекса свои — геокод ему не нужен, geom появляется сразу."""
    rows = [{"id": 1, "payload": _yandex()}]
    mod.import_msk_raw(_FakeSession(rows), source="yandex", batch_size=10)
    lot = captured[0]["lots"][0]
    assert (lot.lat, lot.lon) == (55.737312, 37.601006)


def test_geocode_is_refused_for_yandex(captured: list[dict[str, Any]]) -> None:
    """Квота нужна только Авито: у Яндекса город и точка есть даром."""
    with pytest.raises(SystemExit):
        mod.import_msk_raw(
            _FakeSession([{"id": 1, "payload": _yandex()}]), source="yandex", geocode=True
        )


def test_domclick_moscow_is_recognised_without_geocode() -> None:
    """Регион ДомКлика — первый компонент адреса; Новая Москва тоже Москва."""
    assert mod.is_moscow_domclick_address("Москва, улица Нижние Мнёвники, 16 к2")
    assert mod.is_moscow_domclick_address("Москва, пос. Птичное, Лесная улица, 82")
    assert mod.is_moscow_domclick_address("Москва, x. Ильичевка, хутор Ильичёвка, 1")


def test_domclick_non_moscow_is_rejected() -> None:
    """Вторая линия обороны: GUID в сборщике сменится — импорт не смолчит."""
    assert not mod.is_moscow_domclick_address("Московская область, Химки, 9 Мая, 21к3")
    assert not mod.is_moscow_domclick_address("Екатеринбург, улица Ясная, 2")
    # «Москва» должна быть отдельным компонентом, а не префиксом другого слова.
    assert not mod.is_moscow_domclick_address("Московский, 1-я улица, 5")
    assert not mod.is_moscow_domclick_address("")
    assert not mod.is_moscow_domclick_address(None)


def test_domclick_view_is_wired_into_source_views() -> None:
    assert mod.SOURCE_VIEWS["domclick"] == "msk_raw.domclick_latest"
    assert mod.CITY_FILTERS["domclick"] is mod.is_moscow_domclick_address