"""Daily recompute of per-city ppm² plausible-deal guard-bands (#2576 Stage B).

ПРОБЛЕМА: deal_city_price_bands (migration 178, tier-схема — migration 194)
засеяна ON CONFLICT DO UPDATE derivation-запросом. По мере ночного импорта новых
ДКП-сделок (rosreestr_dkp_import) города переходят между tier ('region_fallback'
N<10 → 'rough' N 10-29 → 'full' N>=30), а перцентили внутри tier дрейфуют — нужен
периодический пересчёт по той же derivation.

Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается
kit-scheduler'ом через product_handlers._job_deal_city_price_bands_refresh
(run_in_executor), по образцу asking_to_sold_ratio.py / snapshot_listing_sources.

Окно расписания 07:00-08:00 UTC — ПОСЛЕ rosreestr_dkp_import (04:00-06:00 UTC) И
asking_to_sold_ratio_refresh (06:00-07:00 UTC), чтобы бэнды считались по тому же
свежему срезу deals, что и ratio-таблица того же дня.

SQL derivation ниже держит ту же трёхуровневую схему, что seed в
data/sql/298_deal_city_price_bands_region.sql (region_stats / city_stats / tiered:
full N>=30 / rough N 10-29 / region_fallback N 1-9, см. комментарий в 194/298 для
полного обоснования тиров и hard floor'а 8000), но РАСХОДИТСЯ с ним в двух местах
(#3051, округа Москвы): ключ города — COALESCE(NULLIF(raw_payload->>'src_city',''),
city) вместо голого city, и потолок ppm² — региональный region_ppm2_max вместо
литерала 800000. Полное обоснование обоих — в комментарии над _REDERIVE_SQL.
Для региона 66 обе правки тождественны прежнему поведению (src_city пуст у всех
его сделок, региональный потолок вырождается ровно в 800000) — проверено на проде
2026-09-10 пересчётом: 383 строки, все совпадают с текущими.

#3051 «Москва» (298): ключ (region_code, city) вместо (city) — region_stats и
city_stats теперь группируются ПО РЕГИОНУ (region_code), а не по всей таблице
deals целиком. Без этого пул для tier='region_fallback' одного региона
подмешивал бы сделки другого (Москва в deals region_code=77 иначе тянула бы
p1-floor малых городов Свердловской обл. region_code=66 вверх). Для
region_code=66 derivation байт-в-байт прежняя (194): фильтр
NOT (region_code = 66 AND <ключ> = 'Екатеринбург'), где <ключ> — то же
выражение, по которому идёт GROUP BY (прод 2026-09-11: у региона 66 ключ == city
у всех 108 623 сделок, обе формы исключают одни и те же 55 749 строк);
region_stats/city_stats для региона 66 видят
ТУ ЖЕ популяцию строк, что видели до появления региона 77 в deals.

Нет DELETE перед re-derive (в отличие от asking_to_sold_ratio.py true-mirror
паттерна) — множество (region_code, city) монотонно растёт
(rosreestr_dkp_import только INSERT/ON CONFLICT DO UPDATE, никогда не удаляет
сделки), поэтому merge-по-ключу (ON CONFLICT DO UPDATE) достаточен: город,
перешедший в другой tier, просто перезаписывается на следующем refresh.
Екатеринбург НЕ включён для региона 66 (WHERE NOT (region_code = 66 AND
<ключ> = 'Екатеринбург')) — estimator.py fallback на глобальные
DEAL_MIN_PPM2/DEAL_MAX_PPM2 для ЕКБ остаётся byte-identical (invariant из
178/194/298 сохранён).
"""

from __future__ import annotations

import logging

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.services import scrape_runs as runs_mod
from app.services.deal_city_key import deal_city_key_sql

logger = logging.getLogger(__name__)

# Ключ города сделки — ОДНО выражение на derivation и на читающую сторону
# (estimator.py), см. app/services/deal_city_key.py. Здесь alias пустой:
# в запросе ниже `FROM deals` без алиаса.
_CITY_KEY_SQL = deal_city_key_sql(alias="")

# ── Числа формулы регионального потолка ppm² (#3051) ─────────────────────────
# ОДИН источник и для SQL (_REGION_CEILING_SQL ниже), и для питоновского
# эквивалента region_ppm2_max(). Раньше формула жила двумя копиями (SQL +
# локальная копия в тесте): подмена множителя 6 на 3 оставляла ВСЕ тесты
# зелёными, тихо роняя потолок Москвы с 1766742 до 883371 и снова срезая дорогие
# округа. Теперь правка любого из этих чисел автоматически едет в обе стороны.
REGION_CEILING_FLOOR = 800_000  # исторический якорь: ниже потолок не падает нигде
REGION_CEILING_MEDIAN_MULT = 6  # шесть медианных ₽/м² региона — заведомо не рынок
REGION_CEILING_MEDIAN_Q = 0.5  # медиана региона
REGION_CEILING_CAP_Q = 0.9999  # шапка: одиночный мусорный выброс не раздувает потолок


def region_ppm2_max(p_cap: int, p_median: int) -> int:
    """Региональный потолок ppm²: GREATEST(floor, LEAST(p99.99, mult * медиана)).

    Питоновский эквивалент _REGION_CEILING_SQL — собран из ТЕХ ЖЕ констант, а не
    из своих чисел. Замеры прода 2026-09-10: регион 66 → (615312, 52706) = 800000
    (тот же прежний литерал), регион 77 → (1944535, 294457) = 1766742.
    """
    return max(REGION_CEILING_FLOOR, min(p_cap, REGION_CEILING_MEDIAN_MULT * p_median))


_REGION_CEILING_CAP_SQL = (
    f"round(percentile_cont({REGION_CEILING_CAP_Q}) WITHIN GROUP (ORDER BY price_per_m2))::int"
)
_REGION_CEILING_MEDIAN_SQL = (
    f"{REGION_CEILING_MEDIAN_MULT} * "
    f"round(percentile_cont({REGION_CEILING_MEDIAN_Q}) WITHIN GROUP (ORDER BY price_per_m2))::int"
)
_REGION_CEILING_SQL = (
    f"GREATEST({REGION_CEILING_FLOOR}, "
    f"LEAST({_REGION_CEILING_CAP_SQL}, {_REGION_CEILING_MEDIAN_SQL}))"
)

# ── Derivation + re-seed (region-aware; #3051 округа Москвы + региональный потолок) ──
#
# #3051 (округа). Ключ города — COALESCE(NULLIF(raw_payload->>'src_city',''), city)
# вместо голого city: у московских сделок src_city несёт муниципальный округ
# (заполнен у 93.27% из 212 937), и вместо ОДНОЙ полосы 'Москва' 34221..718870
# получается 197 ключей — 152 в тире full, 8 rough, 37 region_fallback. Остаточные
# 6.73% сделок без src_city дают собственную строку 'Москва' (n=14376, tier full,
# 22475..772165) — они не проваливаются в глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2,
# откалиброванные под ЕКБ. Регион 66 не меняется: src_city пуст у всех его сделок
# (прод 2026-09-10: 0 строк, где ключ != city).
#
# #3051 (потолок). Литерал 800000 был калибровкой Свердловской области, а в Москве
# p99 округов доходит до 1 405 882 ₽/м² — 15 округов из 197 упирались в потолок,
# т.е. он резал не опечатки, а легитимный рынок. Потолок стал РЕГИОНАЛЬНЫМ
# (region_ppm2_max в region_stats):
#     GREATEST(800000, LEAST(p9999_региона, 6 * медиана_региона))
# Три множителя, каждый со своим смыслом: 800000 — исторический якорь, ниже
# которого потолок не опускается нигде (страхует и от обвала цен); 6 * медиана —
# привязка к масштабу региона (шесть медианных ₽/м² — заведомо не рынок, а
# опечатка или доля); p99.99 — жёсткая шапка, чтобы одиночный мусорный выброс не
# раздул потолок. Замеры: регион 66 → GREATEST(800000, LEAST(615312, 316236)) =
# 800000, тот же литерал; регион 77 → GREATEST(800000, LEAST(1944535, 1766742)) =
# 1766742.
#
# Инвариант региона 66 проверен на проде 2026-09-10 пересчётом по этому же
# выражению: 383 строки против 383 текущих, все совпадают по
# (ppm2_min, ppm2_max, n_deals, tier). Запас прочности: чтобы потолок 66 сдвинулся,
# нужно ОДНОВРЕМЕННО медиане перевалить 133 333 (сейчас 52 706, x2.53) и p99.99
# перевалить 800 000 (сейчас 615 312, x1.3).
#
# Тиры (full N>=30 / rough N 10-29 / region_fallback N<10) и обоснование floor'а
# 8000 — без изменений, см. миграции 194/298.
_REDERIVE_SQL = text(
    f"""
    WITH region_stats AS (
        SELECT
            region_code,
            GREATEST(
                round(percentile_cont(0.01) WITHIN GROUP (ORDER BY price_per_m2))::int,
                8000
            ) AS region_ppm2_min,
            {_REGION_CEILING_SQL} AS region_ppm2_max
        FROM deals
        WHERE source = 'rosreestr'
          AND doc_type = 'ДКП'
          AND price_per_m2 IS NOT NULL
          -- #3051: непустоту ключа судим ТЕМ ЖЕ выражением, по которому идут GROUP BY
          -- и предикат ЕКБ ниже (было: сырая колонка city). Сделка с непустым src_city
          -- и NULL в city даёт ВАЛИДНЫЙ ключ, но сырой фильтр выбрасывал её целиком —
          -- и из её собственной городской строки, и из региональной статистики, молча
          -- занижая n_deals и перцентили региона, в том числе потолок. Прод 2026-09-11:
          -- в популяции 321 560 сделок, city IS NULL — 0 строк, поэтому обе формы
          -- сегодня тождественны: регион 66 — 52 874 строки, p1=15345, p50=52706,
          -- p99.99=615312; регион 77 — 212 937 строк, 34221 / 294457 / 1944535;
          -- расхождение 0 по всем регионам. Совпадение больше не держится на данных.
          AND {_CITY_KEY_SQL} IS NOT NULL
          AND region_code IS NOT NULL
          -- #3051: исключение ЕКБ судится ТЕМ ЖЕ выражением ключа, по которому идёт
          -- GROUP BY ниже (было: голая колонка city). Разъехавшиеся предикат и ключ
          -- держались на данных: сегодня у региона 66 src_city пуст у всех 108 623
          -- сделок, ключ == city, и обе формы дают одни и те же 55 749 исключённых
          -- строк (прод 2026-09-11: by_city=55749, by_key=55749, расхождение 0).
          -- Появись источник с src_city='Екатеринбург' у сделки с другим city —
          -- старая форма пропустила бы её в derivation и завела строку полосы
          -- 'Екатеринбург', которую ступень 1 нашла бы для настоящих ЕКБ-сделок,
          -- сломав намеренное исключение. Теперь исключение и ключ — одно выражение.
          AND NOT (region_code = 66 AND {_CITY_KEY_SQL} = 'Екатеринбург')
        GROUP BY region_code
    ),
    city_stats AS (
        SELECT
            region_code,
            {_CITY_KEY_SQL} AS city,
            GREATEST(round(percentile_cont(0.01) WITHIN GROUP (ORDER BY price_per_m2))::int, 8000)
                AS ppm2_p1,
            -- p99 сырой: клампится региональным потолком в ветке full ниже,
            -- а не литералом 800000 (см. шапку).
            round(percentile_cont(0.99) WITHIN GROUP (ORDER BY price_per_m2))::int
                AS ppm2_p99,
            count(*) AS n_deals
        FROM deals
        WHERE source = 'rosreestr'
          AND doc_type = 'ДКП'
          AND price_per_m2 IS NOT NULL
          -- #3051: непустоту ключа судим ТЕМ ЖЕ выражением, по которому идут GROUP BY
          -- и предикат ЕКБ ниже (было: сырая колонка city). Сделка с непустым src_city
          -- и NULL в city даёт ВАЛИДНЫЙ ключ, но сырой фильтр выбрасывал её целиком —
          -- и из её собственной городской строки, и из региональной статистики, молча
          -- занижая n_deals и перцентили региона, в том числе потолок. Прод 2026-09-11:
          -- в популяции 321 560 сделок, city IS NULL — 0 строк, поэтому обе формы
          -- сегодня тождественны: регион 66 — 52 874 строки, p1=15345, p50=52706,
          -- p99.99=615312; регион 77 — 212 937 строк, 34221 / 294457 / 1944535;
          -- расхождение 0 по всем регионам. Совпадение больше не держится на данных.
          AND {_CITY_KEY_SQL} IS NOT NULL
          AND region_code IS NOT NULL
          -- #3051: исключение ЕКБ судится ТЕМ ЖЕ выражением ключа, по которому идёт
          -- GROUP BY ниже (было: голая колонка city). Разъехавшиеся предикат и ключ
          -- держались на данных: сегодня у региона 66 src_city пуст у всех 108 623
          -- сделок, ключ == city, и обе формы дают одни и те же 55 749 исключённых
          -- строк (прод 2026-09-11: by_city=55749, by_key=55749, расхождение 0).
          -- Появись источник с src_city='Екатеринбург' у сделки с другим city —
          -- старая форма пропустила бы её в derivation и завела строку полосы
          -- 'Екатеринбург', которую ступень 1 нашла бы для настоящих ЕКБ-сделок,
          -- сломав намеренное исключение. Теперь исключение и ключ — одно выражение.
          AND NOT (region_code = 66 AND {_CITY_KEY_SQL} = 'Екатеринбург')
        GROUP BY region_code, {_CITY_KEY_SQL}
    ),
    tiered AS (
        SELECT c.region_code, c.city, c.ppm2_p1 AS ppm2_min,
               LEAST(c.ppm2_p99, r.region_ppm2_max) AS ppm2_max, c.n_deals,
               'full'::text AS tier
        FROM city_stats c
        JOIN region_stats r ON r.region_code = c.region_code
        WHERE c.n_deals >= 30
          AND c.ppm2_p99 >= 8000

        UNION ALL

        SELECT c.region_code, c.city,
               LEAST(c.ppm2_p1, r.region_ppm2_max - 100000) AS ppm2_min,
               r.region_ppm2_max AS ppm2_max,
               c.n_deals, 'rough'::text AS tier
        FROM city_stats c
        JOIN region_stats r ON r.region_code = c.region_code
        WHERE c.n_deals BETWEEN 10 AND 29

        UNION ALL

        SELECT c.region_code, c.city, r.region_ppm2_min AS ppm2_min,
               r.region_ppm2_max AS ppm2_max,
               c.n_deals, 'region_fallback'::text AS tier
        FROM city_stats c
        JOIN region_stats r ON r.region_code = c.region_code
        WHERE c.n_deals < 10
    )
    INSERT INTO deal_city_price_bands
        (region_code, city, ppm2_min, ppm2_max, n_deals, tier, refreshed_at)
    SELECT region_code, city, ppm2_min, ppm2_max, n_deals, tier, now()
    FROM tiered
    ON CONFLICT (region_code, city) DO UPDATE
        SET ppm2_min = EXCLUDED.ppm2_min,
            ppm2_max = EXCLUDED.ppm2_max,
            n_deals = EXCLUDED.n_deals,
            tier = EXCLUDED.tier,
            refreshed_at = EXCLUDED.refreshed_at
    """
)

# ── Post-insert counters ──────────────────────────────────────────────────────
# #3051 (298): regions — число различных region_code в таблице после re-derive
# (Свердловская обл. + Москва после включения региона 77). Добавлено в конец
# SELECT-списка, прежние 4 счётчика на тех же местах — контракт _COUNTERS_SQL
# (rows_written/full_rows/rough_rows/region_fallback_rows) не ломается.
_COUNTERS_SQL = text(
    """
    SELECT
        COUNT(*)                                          AS rows_written,
        COUNT(*) FILTER (WHERE tier = 'full')              AS full_rows,
        COUNT(*) FILTER (WHERE tier = 'rough')             AS rough_rows,
        COUNT(*) FILTER (WHERE tier = 'region_fallback')   AS region_fallback_rows,
        COUNT(DISTINCT region_code)                        AS regions
    FROM deal_city_price_bands
    """
)


def refresh_deal_city_price_bands(db: Session, run_id: int) -> dict[str, int]:
    """Пересчитать deal_city_price_bands (#2576 Stage B — tier-aware refresh).

    Sync (вызывается scheduler-триггером в executor, как recompute_asking_to_sold_ratios).
    Одна транзакция: re-derive INSERT ... ON CONFLICT DO UPDATE (нет DELETE — см.
    module docstring), затем counters из таблицы, commit, mark_done.

    Финализирует scrape_runs (mark_done / mark_failed) и пишет counters.

    Returns {"rows_written": N, "full_rows": .., "rough_rows": .., "region_fallback_rows": ..,
    "regions": ..}.
    """
    counters: dict[str, int] = {
        "rows_written": 0,
        "full_rows": 0,
        "rough_rows": 0,
        "region_fallback_rows": 0,
        "regions": 0,
    }
    try:
        db.execute(_REDERIVE_SQL)

        row = db.execute(_COUNTERS_SQL).mappings().first()
        if row is not None:
            counters["rows_written"] = int(row["rows_written"] or 0)
            counters["full_rows"] = int(row["full_rows"] or 0)
            counters["rough_rows"] = int(row["rough_rows"] or 0)
            counters["region_fallback_rows"] = int(row["region_fallback_rows"] or 0)
            counters["regions"] = int(row["regions"] or 0)

        db.commit()
        runs_mod.mark_done(db, run_id, counters)
        logger.info(
            "refresh_deal_city_price_bands run_id=%d done: "
            "rows_written=%d full=%d rough=%d region_fallback=%d regions=%d",
            run_id,
            counters["rows_written"],
            counters["full_rows"],
            counters["rough_rows"],
            counters["region_fallback_rows"],
            counters["regions"],
        )
        return counters
    except Exception as exc:
        logger.exception("refresh_deal_city_price_bands run_id=%d failed", run_id)
        db.rollback()
        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
        raise