"""#3063: бедный re-scrape больше не стирает признаки продавца.

Апсерт listings присваивал шесть полей напрямую (`= EXCLUDED.X`):
phones, is_homeowner, is_pro_seller, bargain_allowed, sale_type, metro_stations.
Комментарий рядом («Cian-specific: обновляем при каждом re-scrape») был верен,
пока в listings писал один cian, который отдаёт их на каждом проходе. Сейчас в ту
же таблицу пишут yandex/avito/domklik, чей SERP их не отдаёт вовсе — и NULL молча
затирал значение, добытое detail-обогащением. Ни ошибки, ни лога, ни изменения
статуса прогона: колонка просто откатывалась к NULL на обычном переобходе.

Прод 23-24.08: у yandex is_homeowner NULL у 98.6 % (16 232 из 16 460) активных
листингов, is_pro_seller — у 76.6 %.

Соседние поля (address, city, kitchen_area_m2, ceiling_height_m,
mortgage_available, is_apartments, is_rosreestr_checked) уже защищены COALESCE
ровно от этого — с явными объяснениями #2007/#2594/#2777. Эти шесть в защиту
просто не попали.
"""

from __future__ import annotations

import os
import re

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

import inspect
import uuid
from typing import Any
from unittest.mock import MagicMock

import pytest
import scraper_kit.base as base_module
from scraper_kit.base import ScrapedLot, save_listings
from sqlalchemy import text

# Поля, которые detail добывает, а SERP чужого источника не отдаёт.
ERODED_FIELDS = [
    "phones",
    "is_homeowner",
    "is_pro_seller",
    "bargain_allowed",
    "sale_type",
    "metro_stations",
]


def _upsert_sql() -> str:
    return inspect.getsource(base_module)


# ── 1. Статические гарды: работают всегда, даже без БД ───────────────────────


@pytest.mark.parametrize("field", ERODED_FIELDS)
def test_field_is_coalesce_protected_in_set(field: str) -> None:
    """SET присваивает поле через COALESCE, а не сырым EXCLUDED.

    Фальсификация: на коде до правки каждый кейс падает — там стоит
    `is_homeowner = EXCLUDED.is_homeowner` без COALESCE.
    """
    src = _upsert_sql()
    assert not re.search(rf"^\s*{field} = EXCLUDED\.{field},", src, re.M), (
        f"{field} присваивается сырым EXCLUDED — бедный re-scrape затрёт значение"
    )
    assert re.search(
        rf"{field} = COALESCE\(\s*EXCLUDED\.{field}, listings\.{field}\s*\)", src
    ), f"{field} должен присваиваться через COALESCE(EXCLUDED.{field}, listings.{field})"


def test_unchanged_gate_mirrors_the_set_clause() -> None:
    """Правая часть `IS DISTINCT FROM` зеркалит то, что реально запишется.

    Это НЕ стилистика. Гейт #2992 решает, переписывать ли строку, сравнивая
    текущие значения с ИТОГОВЫМИ (post-COALESCE). Если SET станет COALESCE, а
    гейт останется на сыром EXCLUDED, они разъедутся: гейт увидит «NULL против
    значения» и посчитает строку изменившейся там, где она не меняется —
    вернутся ровно те лишние UPDATE и TOAST-чанки, ради которых #2992 делался.

    Гард структурный: сравниваем два кортежа поэлементно, поэтому он поймает и
    будущее расхождение по ЛЮБОЙ колонке, не только по шести из #3063.
    """
    src = _upsert_sql()
    m = re.search(
        r"WHERE \(\n(.*?)\n\s*\) IS DISTINCT FROM \(\n(.*?)\n\s*\)\n\s*OR \(listings\.last_seen_at",
        src,
        re.S,
    )
    assert m is not None, "не найден гейт `IS DISTINCT FROM` — тест устарел вместе с кодом"

    def items(block: str) -> list[str]:
        block = re.sub(r"--[^\n]*", "", block)  # комментарии внутри кортежа
        out: list[str] = []
        depth, cur = 0, ""
        for ch in block:
            if ch == "(":
                depth += 1
            if ch == ")":
                depth -= 1
            if ch == "," and depth == 0:
                out.append(cur.strip())
                cur = ""
            else:
                cur += ch
        if cur.strip():
            out.append(cur.strip())
        return [" ".join(x.split()) for x in out]

    left, right = items(m.group(1)), items(m.group(2))
    assert len(left) == len(right), (
        f"кортежи гейта разной длины: слева {len(left)}, справа {len(right)} — "
        "сравнение поехало бы по колонкам молча"
    )
    # strict=True безопасен: равенство длин уже проверено assert выше.
    for i, (lhs, rhs) in enumerate(zip(left, right, strict=True)):
        column = lhs.split(".")[-1]
        if column == "is_active":
            continue  # SET пишет литерал true, справа он же — намеренно
        assert column in rhs, f"позиция {i}: слева {lhs}, справа {rhs} — колонки разъехались"
        if column in ERODED_FIELDS:
            assert rhs.startswith("COALESCE("), (
                f"{column} в гейте сравнивается сырым EXCLUDED, а SET пишет COALESCE"
            )


# ── 2. Поведенческий тест на живом Postgres (в CI есть, локально skip) ───────


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)
        conn = engine.connect()
        conn.execute(text("SELECT 1"))
        conn.close()
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


def _matcher() -> MagicMock:
    m = MagicMock()
    m.match_or_create_house.return_value = (None, 0.0, "no_address")
    m.upsert_listing_source.return_value = None
    return m


def _lot(src_id: str, *, rich: bool) -> ScrapedLot:
    """rich=True — как detail-обогащение; rich=False — как бедный SERP чужого
    источника (поля не заполнены вовсе)."""
    extra: dict[str, Any] = {}
    if rich:
        extra = {
            "is_homeowner": True,
            "is_pro_seller": False,
            # phones — jsonb вида [{countryCode, number, type}, ...]
            # (019_listings_alter_cian.sql:41), а не список строк.
            "phones": [{"countryCode": "+7", "number": "9000000000", "type": "mobile"}],
            "sale_type": "free",
        }
    return ScrapedLot(
        source="cian",
        source_url=f"https://ekb.cian.ru/sale/flat/t3063-{src_id}/",
        source_id=f"t3063-{src_id}",
        price_rub=5_000_000,
        **extra,
    )


def _cleanup(db: Any) -> None:
    try:
        db.rollback()
        db.execute(
            text(
                "DELETE FROM listings_snapshots WHERE listing_id IN "
                "(SELECT id FROM listings WHERE source='cian' AND source_id LIKE 't3063-%')"
            )
        )
        db.execute(text("DELETE FROM listing_sources WHERE ext_id LIKE 't3063-%'"))
        db.execute(text("DELETE FROM listings WHERE source='cian' AND source_id LIKE 't3063-%'"))
        db.commit()
    finally:
        db.close()


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_poor_rescrape_does_not_erase_seller_fields() -> None:
    """Головной: обогащённая строка переживает последующий бедный проход.

    На коде до правки второй save_listings обнуляет is_homeowner/phones/sale_type.
    """
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        save_listings(db, [_lot(sid, rich=True)], matcher=_matcher(), region_code=66)
        save_listings(db, [_lot(sid, rich=False)], matcher=_matcher(), region_code=66)
        row = db.execute(
            text(
                "SELECT is_homeowner, is_pro_seller, phones, sale_type FROM listings "
                "WHERE source='cian' AND source_id = :sid"
            ),
            {"sid": f"t3063-{sid}"},
        ).fetchone()
        assert row is not None, "строка исчезла"
        assert row.is_homeowner is True, "бедный re-scrape стёр is_homeowner"
        assert row.is_pro_seller is False, "бедный re-scrape стёр is_pro_seller (False -> NULL)"
        assert row.phones is not None, "бедный re-scrape стёр phones"
        assert row.sale_type == "free", "бедный re-scrape стёр sale_type"
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_change_still_overwrites() -> None:
    """Обратная сторона: COALESCE не превращает поля в write-once.

    Настоящая смена признака (частник -> агентство) приходит НЕ NULL'ом, значит
    обязана перезаписать. Без этого теста фикс мог бы «защитить» данные ценой
    того, что они перестали бы обновляться вообще.
    """
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        save_listings(db, [_lot(sid, rich=True)], matcher=_matcher(), region_code=66)
        flipped = _lot(sid, rich=True)
        flipped.is_homeowner = False
        flipped.is_pro_seller = True
        save_listings(db, [flipped], matcher=_matcher(), region_code=66)
        row = db.execute(
            text(
                "SELECT is_homeowner, is_pro_seller FROM listings "
                "WHERE source='cian' AND source_id = :sid"
            ),
            {"sid": f"t3063-{sid}"},
        ).fetchone()
        assert row.is_homeowner is False, "настоящая смена признака не записалась"
        assert row.is_pro_seller is True, "настоящая смена признака не записалась"
    finally:
        _cleanup(db)