"""Daily recompute of the asking→sold correction ratios (#648 Stage 4).

ПРОБЛЕМА: asking_to_sold_ratios (migration 080) засеяна один раз derivation-CTE
(медиана SOLD ДКП за 12 мес vs медиана ASKING активных listings, per-rooms + global -1
fallback). Оценщик (estimator.py, Stage 3) читает её per-estimate (cache 300s) и домножает
asking-median прогноз на sold/asking. По мере ночного импорта новых ДКП-сделок
(rosreestr_dkp_import) ratio устаревает — нужен периодический пересчёт по тому же derivation.

TRUE-MIRROR REFRESH (флаг database-expert): 080-seed использовал ON CONFLICT DO UPDATE,
который при повторном прогоне ОБНОВЛЯЕТ существующие строки, но НЕ удаляет per-rooms строки
бакетов, упавших ниже порога 30/30 на более позднем прогоне (StaLE rows). Поэтому refresh
сначала DELETE FROM asking_to_sold_ratios WHERE district = '' (все строки #648), затем
заново гоняет ту же 080-derivation INSERT...SELECT. DELETE+INSERT в ОДНОЙ транзакции
(атомарно — таблица никогда не пуста mid-refresh). После commit таблица == свежий re-seed.

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

Окно расписания 06:00-07:00 UTC — ПОСЛЕ rosreestr_dkp_import (04:00-06:00 UTC), чтобы
refresh потреблял свежие ДКП-сделки того же дня.

SQL derivation ниже повторяет seed в data/sql/080_asking_to_sold_ratios.sql (deal_side /
ask_side / per_bucket + deal_global / ask_global / global_row: трейлинг-12мес окно, ppm²-полоса
[_PPM2_MIN, settings.asking_ratio_ppm2_max] (default [30000,1200000]), порог n_deals>=30 AND
n_listings>=30 для per_rooms, global -1 строка всегда). ON CONFLICT убран — DELETE идёт первым,
конфликтов нет (повторный прогон в одной tx невозможен, refresh = re-seed по семантике).

#2620 — ОДНО ПРЕДНАМЕРЕННОЕ РАСХОЖДЕНИЕ с 080: deal_side бакетится по
LEAST(GREATEST(rooms,0),4), а ask_side — по _AREA_ROOMS_BUCKET_SQL (площадь, та же формула,
что deals.rooms получает при импорте). Причина — deals.rooms НЕ настоящая комнатность
(Росреестр её не отдаёт), это синтетика из площади; сравнивать её с РЕАЛЬНЫМИ комнатами
listings значило сравнивать разные классификации. Замер на проде (2026-08, #2620) показал
миграцию 23-55% объявлений между бакетами при таком сравнении — не только в бакете «4+»
(который к тому же обрезан обрезкой ELSE 4, тогда как listings.rooms доходит до 10) — и
это и была причина ratio>1 в бакете 4+ (см. _AREA_ROOMS_BUCKET_SQL ниже).
"""

from __future__ import annotations

import logging

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.services import scrape_runs as runs_mod

# Нижняя граница ppm² — отсекает нежилые/технические сделки; не меняется.
_PPM2_MIN: int = 30_000

# #C2 — исторически asking-сторона (listings) была покрыта скрейпом ТОЛЬКО по ЕКБ, а
# миграция 177 залила ДКП-сделки по всей обл.66 (368 городов) → sold-медиана смешивала
# дешёвую область с ЕКБ-asking и обваливала ratio (0.877→0.62, «выкупная» −29% системно).
# Скоупили SOLD-сторону (deal_side/deal_global) на ЕКБ, чтобы sold и asking считались по
# ОДНОМУ рынку.
#
# #2583 H2 (аудит, 2026-08): oblast-развёртки заработали 12 июля — областные объявления
# попали в знаменатель (ask_side/ask_global) без городского скоупа, а sold-сторона
# осталась скоуплена на ЕКБ → асимметрия вернулась с другой стороны (дешёвая область
# занижает ask-медиану → ratio завышен на 2.5-5.3% по всем бакетам, выкупные цены
# системно переплачены). Теперь ask_side/ask_global ТОЖЕ скоупятся этим паттерном
# (предикат `city IS NULL OR city ILIKE :asking_city` — см. комментарий на месте в CTE
# ниже) — симметрично deal-стороне. Когда появится per-city ratio через зарезервированный
# столбец `district` (#647), эта константа станет per-city параметром для обеих сторон.
_ASKING_CITY_PATTERN: str = "%Екатеринбург%"
# Верхняя граница берётся из settings.asking_ratio_ppm2_max (default 1_200_000).
# QA-note: точное значение сверить с `SELECT max(price_per_m2) FROM deals
# WHERE source='rosreestr'` на проде — ceiling должен быть > max(ppm²) premium-сделок.

# #2620 — синтетический "бакет комнат по площади", ИСТОЧНИК ИСТИНЫ:
# tradein-mvp/deploy/import-rosreestr.sh (Росреестр не отдаёт комнатность — deals.rooms
# синтезируется из area_m2 при импорте ровно этим CASE). Три представления ОДНОЙ формулы —
# держи границы (30/44/62/85) в синхроне при правке: shell (import-rosreestr.sh) → SQL
# (эта константа, ask_side ниже) → Python (area_bucket() ниже, estimator.py rekey #2620-2).
_AREA_ROOMS_BUCKET_SQL = (
    "CASE WHEN area_m2 < 30 THEN 0 WHEN area_m2 < 44 THEN 1 "
    "WHEN area_m2 < 62 THEN 2 WHEN area_m2 < 85 THEN 3 ELSE 4 END"
)


def area_bucket(area_m2: float) -> int:
    """Python-двойник _AREA_ROOMS_BUCKET_SQL (границы ИДЕНТИЧНЫ, #2620).

    Используется estimator.py при ПРИМЕНЕНИИ ratio (не только при расчёте здесь) —
    ratio_resolver должен ключевать по ТОМУ ЖЕ area-бакету, что и ask_side при
    деривации, иначе mismatch просто переезжает из расчёта в применение (прод-замер
    ревьюера #2620: 310/1038 = 29.9% исторических запросов легли бы в другой бакет
    при rooms-ключе vs area-ключе).
    """
    if area_m2 < 30:
        return 0
    if area_m2 < 44:
        return 1
    if area_m2 < 62:
        return 2
    if area_m2 < 85:
        return 3
    return 4


logger = logging.getLogger(__name__)

# ── True-mirror cleanup: drop all #648 rows before re-derivation ──────────────
# district = '' — все строки #648 (district зарезервирован под #647, пока всегда '').
# Удаляем ПЕРЕД re-derive, чтобы бакеты, упавшие ниже порога 30/30, не оставались
# stale (ON CONFLICT DO UPDATE такие строки бы не тронул). В одной транзакции с INSERT.
_DELETE_SQL = text(
    """
    DELETE FROM asking_to_sold_ratios
    WHERE district = ''
    """
)

# ── Derivation + re-seed (БАЙТ-В-БАЙТ из 080, ON CONFLICT убран — DELETE идёт первым) ──
# deal_side / ask_side / per_bucket + deal_global / ask_global / global_row:
#   sold_median = percentile_cont(0.5) по deals.price_per_m2 (source='rosreestr',
#     ppm² ∈ [_PPM2_MIN, settings.asking_ratio_ppm2_max], deal_date >= CURRENT_DATE − 12 months),
#     бакет LEAST(GREATEST(rooms,0),4) (rooms уже синтетика-из-площади при импорте, см. #2620
#     комментарий у _AREA_ROOMS_BUCKET_SQL выше).
#   ask_median  = percentile_cont(0.5) по listings.price_per_m2
#     (is_active, та же ppm²-полоса [_PPM2_MIN, asking_ratio_ppm2_max], тот же город что
#     SOLD-сторона — city IS NULL OR city ILIKE :asking_city, #2583 H2). Бакет —
#     _AREA_ROOMS_BUCKET_SQL (площадь, #2620), НЕ listings.rooms — см. комментарий там.
#   per_rooms строки — только при n_deals>=30 AND n_listings>=30 AND ask>0 AND sold>0.
#   global -1 строка (basis='global_fallback') — всегда (если ask>0 AND sold>0). window_months=12.
# Порог/окно — литералы; ppm²-полоса передаётся bind-параметрами :ppm2_min/:ppm2_max
# (безопасно от SQL-инъекций; CAST не нужен — psycopg v3 передаёт int напрямую).
_REDERIVE_SQL = text(
    f"""
    WITH
    -- SOLD медианы по бакетам комнат за трейлинг-12мес (ДКП Росреестра).
    deal_side AS (
        SELECT
            LEAST(GREATEST(rooms, 0), 4)                                  AS rooms_bucket,
            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)     AS sold_median,
            COUNT(*)                                                      AS n_deals
        FROM deals
        WHERE source = 'rosreestr'
          AND rooms IS NOT NULL
          AND city ILIKE :asking_city  -- #C2 SOLD-сторона на ЕКБ (match asking-рынок)
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
        GROUP BY LEAST(GREATEST(rooms, 0), 4)
    ),
    -- ASKING медианы по ТОМУ ЖЕ area-бакету, что deal_side (#2620) — НЕ по listings.rooms.
    -- deals.rooms — синтетика из площади (Росреестр её не отдаёт), listings.rooms — реальная
    -- комнатность; сравнение area-бакета с area-бакетом (не area-бакета с real-rooms-бакетом)
    -- убирает миграцию объявлений между бакетами (23-55% строк на проде, 2026-08, #2620) —
    -- включая инверсию ratio>1 в бакете «4+» (deals.rooms обрезан ELSE 4, а listings.rooms
    -- нет: 110/782 пяти- и более комнатных объявлений раньше схлопывались в бакет 4).
    ask_side AS (
        SELECT
            {_AREA_ROOMS_BUCKET_SQL}                                      AS rooms_bucket,
            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)     AS ask_median,
            COUNT(*)                                                      AS n_listings
        FROM listings
        WHERE is_active
          AND rooms IS NOT NULL
          -- #2620 hardening: area_m2 IS NULL falls into the CASE ELSE branch (bucket 4)
          -- of _AREA_ROOMS_BUCKET_SQL — a latent "everything unmeasured looks like a big
          -- flat" trap. Excluded explicitly instead of relying on ELSE-as-junk-drawer.
          AND area_m2 IS NOT NULL
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          -- novostroyki guard (#1186): NULL = legacy вторичка до м.011
          AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
          -- #2583 H2: скоупим ASKING-сторону на тот же город, что и SOLD-сторона
          -- (симметрично deal_side выше) — иначе дешёвые oblast-объявления (развёртки
          -- с 12 июля) занижают ask-медиану и завышают ratio. city IS NULL считается
          -- "своим" (не отбрасывается) НАМЕРЕННО: listings.city заполнена пока только у
          -- Авито (Циан/Домклик/Яндекс — NULL, #2598/#2606), симметричный
          -- `city ILIKE :asking_city` без IS NULL выбросил бы ~70% выборки. По мере
          -- роста покрытия колонки этот предикат сам ужесточается без правок кода; когда
          -- покрытие станет полным — заменить на строго симметричный `city ILIKE :asking_city`.
          AND (city IS NULL OR city ILIKE :asking_city)
        GROUP BY {_AREA_ROOMS_BUCKET_SQL}
    ),
    -- Per-rooms строки: только бакеты с обеими сторонами, прошедшие порог 30/30 и ask>0.
    -- Тонкие бакеты (n<30) сюда НЕ попадают → estimator делает fallback на -1.
    per_bucket AS (
        SELECT
            d.rooms_bucket,
            ''::text                                  AS district,
            (d.sold_median / a.ask_median)::numeric   AS ratio,
            round(d.sold_median)::bigint              AS sold_median,
            round(a.ask_median)::bigint               AS ask_median,
            d.n_deals::int                            AS n_deals,
            a.n_listings::int                         AS n_listings,
            12                                        AS window_months,
            'per_rooms'::text                         AS basis
        FROM deal_side d
        JOIN ask_side a USING (rooms_bucket)
        WHERE d.n_deals >= 30
          AND a.n_listings >= 30
          AND a.ask_median IS NOT NULL
          AND a.ask_median > 0
          AND d.sold_median IS NOT NULL          -- divide-safety (порог n_deals>=30 гарантирует)
          AND d.sold_median > 0
    ),
    -- SOLD медиана по ВСЕМ комнатам (без бакет-фильтра) за трейлинг-12мес — для global row.
    deal_global AS (
        SELECT
            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median,
            COUNT(*)                                                  AS n_deals
        FROM deals
        WHERE source = 'rosreestr'
          AND rooms IS NOT NULL
          AND city ILIKE :asking_city  -- #C2 SOLD-сторона на ЕКБ (match asking-рынок)
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
    ),
    -- ASKING медиана по ВСЕМ активным listings (без бакет-фильтра) — для global row.
    ask_global AS (
        SELECT
            percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median,
            COUNT(*)                                                  AS n_listings
        FROM listings
        WHERE is_active
          AND rooms IS NOT NULL
          -- #2620 hardening: same area_m2 IS NOT NULL as ask_side — keeps the global-row
          -- population consistent with the per-bucket rows it's a fallback for.
          AND area_m2 IS NOT NULL
          AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
          -- novostroyki guard (#1186): NULL = legacy вторичка до м.011
          AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
          -- #2583 H2: тот же городской скоуп, что и ask_side выше (см. комментарий там
          -- про причину city IS NULL == "свой" и #2598/#2606).
          AND (city IS NULL OR city ILIKE :asking_city)
    ),
    -- Global fallback строка rooms_bucket=-1 (пишется всегда, если ask>0).
    global_row AS (
        SELECT
            -1                                        AS rooms_bucket,
            ''::text                                  AS district,
            (d.sold_median / a.ask_median)::numeric   AS ratio,
            round(d.sold_median)::bigint              AS sold_median,
            round(a.ask_median)::bigint               AS ask_median,
            d.n_deals::int                            AS n_deals,
            a.n_listings::int                         AS n_listings,
            12                                        AS window_months,
            'global_fallback'::text                   AS basis
        FROM deal_global d
        CROSS JOIN ask_global a
        WHERE a.ask_median IS NOT NULL
          AND a.ask_median > 0
          AND d.sold_median IS NOT NULL          -- защита от пустого окна сделок (иначе ratio=NULL)
          AND d.sold_median > 0
    )
    INSERT INTO asking_to_sold_ratios (
        rooms_bucket, district, ratio, sold_median, ask_median,
        n_deals, n_listings, window_months, basis
    )
    SELECT rooms_bucket, district, ratio, sold_median, ask_median,
           n_deals, n_listings, window_months, basis FROM global_row
    UNION ALL
    SELECT rooms_bucket, district, ratio, sold_median, ask_median,
           n_deals, n_listings, window_months, basis FROM per_bucket
    """
)

# ── Post-insert counters ──────────────────────────────────────────────────────
# Считываем итог из таблицы (всё ещё в той же транзакции — до commit): сколько строк
# записано всего, сколько per_rooms, был ли использован global -1 fallback.
_COUNTERS_SQL = text(
    """
    SELECT
        COUNT(*)                                              AS rows_written,
        COUNT(*) FILTER (WHERE basis = 'per_rooms')           AS per_rooms_rows,
        COUNT(*) FILTER (WHERE rooms_bucket = -1)             AS used_global_fallback
    FROM asking_to_sold_ratios
    WHERE district = ''
    """
)


def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
    """Пересчитать asking_to_sold_ratios (TRUE-MIRROR refresh, #648 Stage 4).

    Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
    В ОДНОЙ транзакции (атомарно — таблица никогда не пуста mid-refresh):
      1. DELETE FROM asking_to_sold_ratios WHERE district = '' — снести stale-строки.
      2. Заново прогнать 080-derivation INSERT...SELECT (per_rooms при 30/30 + global -1).
    Затем counters из таблицы, commit, mark_done. Семантика == re-seed миграции 080.

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

    LIMITATION (#2620, честно задокументировано — не гард, а факт данных): sold-сторона
    (deals) НЕ имеет маркера новостройка/вторичка — Росреестр таким свойством ДКП не
    делится, а listing_segment (гард #1186) существует только у listings. ask_side/ask_global
    отфильтрованы на вторичку, deal_side/deal_global — нет. Замер на проде (2026-08, #2620):
    доля сделок с year_built >= 2020 (грубый прокси новостройки) — 44.1% в бакете «4+» против
    23.1% в бакетах 1-3 — заметный перекос, но year_built НЕ идентифицирует первичку/вторичку
    (продажа квартиры 2021 года постройки в 2026м — легитимная вторичка), поэтому фильтр по
    году НЕ добавлен (создал бы новую, столь же спекулятивную асимметрию). Area-бакет-фикс
    ниже (см. _AREA_ROOMS_BUCKET_SQL) сам по себе убрал инверсию ratio>1 в бакете «4+»
    (0.8315 на замере прод-данных 2026-08, было 1.0257) — снятие миграции между бакетами было
    root cause, а не новостройки.

    Returns {"rows_written": N, "per_rooms_rows": M, "used_global_fallback": 0|1}.
    """
    counters: dict[str, int] = {
        "rows_written": 0,
        "per_rooms_rows": 0,
        "used_global_fallback": 0,
    }
    try:
        # DELETE + re-derive INSERT в одной транзакции (НЕ коммитим между ними —
        # таблица не должна остаться пустой, если INSERT упадёт).
        db.execute(_DELETE_SQL)
        db.execute(
            _REDERIVE_SQL,
            {
                "ppm2_min": _PPM2_MIN,
                "ppm2_max": settings.asking_ratio_ppm2_max,
                "asking_city": _ASKING_CITY_PATTERN,
            },
        )

        row = db.execute(_COUNTERS_SQL).mappings().first()
        if row is not None:
            counters["rows_written"] = int(row["rows_written"] or 0)
            counters["per_rooms_rows"] = int(row["per_rooms_rows"] or 0)
            counters["used_global_fallback"] = int(row["used_global_fallback"] or 0)

        db.commit()
        runs_mod.mark_done(db, run_id, counters)
        logger.info(
            "recompute_asking_to_sold_ratios run_id=%d done: "
            "rows_written=%d per_rooms_rows=%d used_global_fallback=%d",
            run_id,
            counters["rows_written"],
            counters["per_rooms_rows"],
            counters["used_global_fallback"],
        )
        return counters
    except Exception as exc:
        logger.exception("recompute_asking_to_sold_ratios run_id=%d failed", run_id)
        db.rollback()
        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
        raise