"""Ключ города ДКП-сделки для ценовых полос (#3051 «Москва», округа).

ПРОБЛЕМА. deal_city_price_bands ключуется (region_code, city), а deals.city у
ВСЕХ 212 937 московских сделок буквально 'Москва' — на весь город получалась
ОДНА полоса 34221..718870 ₽/м². Москва неоднородна на порядок (Хамовники против
Некрасовки), поэтому единая полоса одновременно и не режет опечатки в дорогом
центре, и режет легитимный рынок на окраинах.

РЕШЕНИЕ. Ключом города становится
    COALESCE(NULLIF(raw_payload->>'src_city', ''), city)
— у московских сделок Росреестра raw_payload.src_city несёт муниципальный округ
('муниципальный округ Хамовники' и т.п.): заполнен у 198 600 из 212 937 сделок
(93.27%), 197 различных значений. Оставшиеся 6.73% (14 376 сделок) отдают
'Москва' и образуют СВОЮ строку-фолбэк (n=14376 → tier 'full', полоса
22475..772165), а не проваливаются в глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2,
откалиброванные под Екатеринбург. Ключи дизъюнктны: либо 'муниципальный округ X',
либо ровно 'Москва' — b.city = ключ_сделки всегда матчит одну строку.

ИНВАРИАНТ РЕГИОНА 66. У сделок Свердловской области src_city пуст у ВСЕХ
108 623 строк, поэтому COALESCE отдаёт city и ключ не меняется. Проверено на
проде 2026-09-10: сделок региона 66, где ключ отличается от city, — 0 штук;
пересчёт по новому выражению даёт те же 383 строки полос, все совпадают с
текущими по (ppm2_min, ppm2_max, n_deals, tier).

NULL-семантика. Если raw_payload отсутствует целиком (NULL), то
NULL->>'src_city' = NULL → NULLIF(NULL,'') = NULL → COALESCE отдаёт city.
Если src_city есть, но пустая строка — NULLIF гасит её в NULL, тот же исход.
Питоновский helper ниже повторяет эту семантику один-в-один (пустая строка
считается отсутствующей, пробелы НЕ подрезаются — SQL их тоже не подрезает).

Модуль намеренно крошечный и без зависимостей: выражение обязано быть ОДНИМ на
derivation (app/tasks/deal_city_price_bands_refresh.py) и на все читающие места
(app/services/estimator.py). Три копии выражения = три места, где полосы
разъезжаются молча.
"""

from __future__ import annotations

from collections.abc import Mapping
from typing import Any

# Имя вычисляемой колонки-ключа в SELECT'ах, которые читают сделки для питонового
# пути фильтрации (_fetch_deals → _is_plausible_deal).
DEAL_CITY_KEY_COLUMN = "city_key"


def deal_city_key_sql(alias: str = "d") -> str:
    """SQL-выражение ключа города сделки.

    alias — префикс таблицы deals в запросе ('d' для `FROM deals d`, '' для
    `FROM deals` без алиаса, как в derivation задачи полос).
    """
    prefix = f"{alias}." if alias else ""
    return f"COALESCE(NULLIF({prefix}raw_payload->>'src_city', ''), {prefix}city)"


def deal_city_key(row: Mapping[str, Any]) -> str | None:
    """Питоновский эквивалент deal_city_key_sql для уже прочитанной строки сделки.

    Порядок: готовая колонка DEAL_CITY_KEY_COLUMN (её считает SQL) → src_city из
    raw_payload, если строку прочитали вместе с payload → city. Пустая строка
    трактуется как отсутствие значения — как NULLIF(x, '') в SQL.
    """
    key = row.get(DEAL_CITY_KEY_COLUMN)
    if key:
        return str(key)
    raw = row.get("raw_payload")
    if isinstance(raw, Mapping):
        src = raw.get("src_city")
        if src:
            return str(src)
    city = row.get("city")
    return str(city) if city else None


# ── Двухступенчатый поиск полосы (#3051, разрыв на деплое) ───────────────────
#
# Таблицу deal_city_price_bands наполняет НОЧНАЯ задача, а читающая сторона
# уезжает на ключ-округ сразу с деплоем. В окне между деплоем и первым рефрешем
# по региону 77 в таблице лежит ровно ОДНА строка city='Москва': 93.27%
# московских сделок искали бы ключ 'муниципальный округ X', не находили и падали
# на глобальные DEAL_MIN_PPM2=50000 / DEAL_MAX_PPM2=800000 (калибровка ЕКБ) —
# нижняя граница прыгала бы с 34221 до 50000 и молча выбрасывала легитимно
# дешёвые сделки. Это ХУЖЕ, чем было до правки.
#
# Поэтому поиск полосы ДВУХСТУПЕНЧАТЫЙ и одинаковый во всех трёх читающих местах
# (два SQL-джойна ДКП-коридора + питоновский путь _fetch_deals):
#   ступень 1 — строка по ключу-округу (deal_city_key / deal_city_key_sql);
#   ступень 2 — строка по deals.city;
#   ступень 3 — глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2.
# Порядок деплоя перестаёт иметь значение, а округ, для которого строки ещё нет
# (свежий округ, n<10, мусорное значение src_city), деградирует в ГОРОДСКУЮ
# полосу, а не в екатеринбургскую калибровку.
#
# Ступень не расщепляется по границам: ppm2_min и ppm2_max в таблице NOT NULL
# (data/sql/178_deal_city_price_bands.sql), значит найденная строка отдаёт ОБЕ
# границы — COALESCE не может взять min из округа, а max из города.
#
# ИНВАРИАНТ РЕГИОНА 66. src_city пуст у всех его 108 623 сделок → ключ ступени 1
# равен ключу ступени 2, обе ступени находят одну и ту же строку, результат
# байт-в-байт прежний. Вторая ступень для него — тавтология, не изменение.

DEAL_CITY_BAND_ALIAS = "b"  # ступень 1: строка по ключу-округу
DEAL_CITY_BAND_FALLBACK_ALIAS = "bc"  # ступень 2: строка по deals.city


def deal_city_band_join_sql(alias: str = "d", indent: str = "") -> str:
    """Два LEFT JOIN'а к deal_city_price_bands: ступень 1 (округ) + ступень 2 (город).

    alias — префикс таблицы deals, indent — отступ строк со 2-й (косметика SQL).
    """
    prefix = f"{alias}." if alias else ""
    b, bc = DEAL_CITY_BAND_ALIAS, DEAL_CITY_BAND_FALLBACK_ALIAS
    lines = [
        f"LEFT JOIN deal_city_price_bands {b}",
        f"    ON {b}.region_code = {prefix}region_code",
        f"   AND {b}.city = {deal_city_key_sql(alias)}",
        f"LEFT JOIN deal_city_price_bands {bc}",
        f"    ON {bc}.region_code = {prefix}region_code",
        f"   AND {bc}.city = {prefix}city",
    ]
    return ("\n" + indent).join(lines)


def deal_city_band_bounds_sql(
    min_param: str = ":ppm_min", max_param: str = ":ppm_max", indent: str = ""
) -> str:
    """Границы полосы одним выражением: округ → город → глобальные константы."""
    b, bc = DEAL_CITY_BAND_ALIAS, DEAL_CITY_BAND_FALLBACK_ALIAS
    lines = [
        f"BETWEEN COALESCE({b}.ppm2_min, {bc}.ppm2_min, CAST({min_param} AS int))",
        f"    AND COALESCE({b}.ppm2_max, {bc}.ppm2_max, CAST({max_param} AS int))",
    ]
    return ("\n" + indent).join(lines)


def resolve_city_band(
    bands: Mapping[tuple[int, str], tuple[int, int]] | None,
    region_code: int,
    city_key: str | None,
    city: str | None,
    default: tuple[int, int],
) -> tuple[int, int]:
    """Питоновский эквивалент двух LEFT JOIN'ов выше: округ → город → default.

    Ступени и их порядок обязаны совпадать с SQL-версией: разъехавшийся порядок
    означал бы, что питоновский фильтр сделок судит по другой полосе, чем
    ДКП-коридор на тех же данных.
    """
    table = bands or {}
    for key in (city_key, city):
        if key is None:
            continue
        band = table.get((region_code, key))
        if band is not None:
            return band
    return default