gendesign/tradein-mvp/backend/app/api/v1/trade_in.py

"""Trade-In Estimator — endpoints (TI-2 PDF, photos, history).

Реальная оценка делается через app.services.estimator.estimate_quality().
"""

from __future__ import annotations

import asyncio
import calendar
import json
import logging
import math
from datetime import UTC, date, datetime, timedelta
from typing import Annotated, Any, Literal
from uuid import UUID

from fastapi import APIRouter, Depends, File, Header, HTTPException, Request, Response, UploadFile
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.anon_session import get_or_create_anon_session_id
from app.core.config import settings
from app.core.db import get_db
from app.core.ratelimit import SlidingWindowLimiter, _client_ip
from app.schemas.trade_in import (
    AggregatedEstimate,
    AnalogLot,
    AvitoImvSummary,
    CianPriceChangeStats,
    CoverageProbeInput,
    CoverageProbeResponse,
    DkpCorridor,
    HouseAnalyticsKpi,
    HouseAnalyticsResponse,
    HouseInfoForEstimate,
    IMVBenchmarkResponse,
    LocationIndexResponse,
    NearbyPoiOut,
    PhotoMeta,
    PlacementHistoryEntry,
    PriceHistoryYearPoint,
    PriceTrendPoint,
    QuotaStatus,
    RecentSoldEntry,
    SalesListingPair,
    SalesVsListingsResponse,
    SellTimeBucket,
    SellTimeSensitivityResponse,
    StreetDealsResponse,
    TradeInEstimateInput,
)
from app.services import account_quota
from app.services import regions as regions_mod
from app.services.exporters.trade_in_pdf import generate_trade_in_pdf
from app.services.image_sanitizer import ImageSanitizationError, sanitize_image
from app.services.user_events import schedule_event

logger = logging.getLogger(__name__)

router = APIRouter()

# ── B2C anti-abuse этап 2 (#b2c-antiabuse-2) ────────────────────────────────
# Отдельный, куда более строгий лимит частоты specifically на POST /estimate —
# см. settings.estimate_rate_limit/_window_s (app/core/config.py) для обоснования
# значений. Тот же паттерн, что _send_limiter в app/api/v1/support.py: singleton
# SlidingWindowLimiter поверх общего RateLimitMiddleware (app/main.py), который
# уже применяется КО ВСЕМ /api/* путям, но с щедрым порогом, рассчитанным на
# дешёвые запросы — один /estimate запускает цепочку внешних вызовов, суммарно
# занимающую десятки секунд (см. estimator._with_budget budgets).
_estimate_limiter = SlidingWindowLimiter(
    limit=settings.estimate_rate_limit, window_s=settings.estimate_rate_limit_window_s
)

# ── Потолок одновременных оценок (#3082) ─────────────────────────────────────
#
# Рейт-лимит выше меряет ЧАСТОТУ (300/60с вправе стартовать в одну секунду), а
# квота — счётная и помесячная: ни один из них не ограничивает ПАРАЛЛЕЛИЗМ.
# Оценка 0.8–2.4с держит соединение общего пула SQLAlchemy (дефолт 5+10) и
# внешние тиры; пила одновременных оценок выедает пул и тормозит весь /api/v1/*.
# Образец — public/mera.py::_suggest_slots (4 слота на секундное автодополнение).
#
# 4 слота: вместе с 4 слотами suggest — 8 одновременно удерживаемых соединений.
# Пул под это заведомо шире: 5+15=20 на процесс, и потолок пула держится не
# меньше СУММЫ объявленных потолков одновременности, включая 8 фоновых догрузок
# (core/db.py + tests/test_3408_pool_ceiling.py, #3408). Ожидание слота 5с ≈ две
# длительности оценки: если за это время слот не освободился, очередь глубока и
# честный ответ — быстрый 429 с Retry-After, а не растущая очередь (очередь под
# нагрузкой — те же занятые соединения плюс таймаут у клиента; mera.py:117-127).
#
# Семафор живёт в памяти процесса — при переходе на несколько воркеров uvicorn
# (#3083) фактический лимит умножится на число воркеров; задачи согласовывать.
_ESTIMATE_CONCURRENCY = 4
_ESTIMATE_SLOT_WAIT_S = 5.0
_estimate_slots = asyncio.Semaphore(_ESTIMATE_CONCURRENCY)


def _resolve_quota_identity(
    request: Request,
    response: Response,
    x_authenticated_user: str | None,
) -> tuple[str | None, int]:
    """Резолвит (quota_key, default_limit) для account_quota.* — #b2c-antiabuse-2.

    - X-Authenticated-User присутствует → (username, MONTHLY_LIMIT) — существующий
      pilot/admin-флоу БЕЗ изменений (персональные override в
      account_quota_overrides применяются как раньше через account_quota.user_limit).
    - Заголовка нет И settings.quota_dev_fail_open=True (явный dev-флаг локальной
      разработки без Caddy) → (None, MONTHLY_LIMIT) — account_quota трактует None
      как unlimited. Флаг по умолчанию ВЫКЛЮЧЕН — это НЕ дефолтный прод-путь.
    - Заголовка нет И флаг не задан (default, прод-путь для анонимов — продукт
      открывается наружу) → анонимный ключ на основе подписанной session-cookie
      (app.core.anon_session) + client IP, default_limit =
      settings.anon_estimate_quota_limit (гораздо строже пилот-лимита). Честно:
      смена IP или чистка cookie обходит этот лимит — цель поднять стоимость
      злоупотребления, а не сделать его невозможным (тот же принцип, что и во
      всей account_quota-схеме, #747).
    """
    if x_authenticated_user:
        return x_authenticated_user, account_quota.MONTHLY_LIMIT
    if settings.quota_dev_fail_open:
        return None, account_quota.MONTHLY_LIMIT
    session_id = get_or_create_anon_session_id(request, response)
    anon_key = f"anon:{session_id}:{_client_ip(request)}"
    return anon_key, settings.anon_estimate_quota_limit


# PR-D1: единственное определение «оценка читаема» — раньше SQL-фильтр (404,
# ниже в get_estimate) и Python-проверка (410, в estimate_pdf) уже разошлись
# по коду ответа; третий потребитель (`/r/<token>`, PR-9) разошёлся бы
# неизбежно без унификации. `retain_until > NOW()` при NULL даёт NULL → false
# в SQL — для всех существующих строк (retain_until IS NULL) поведение не
# меняется вообще. Не копировать это выражение по месту — только через
# константу/хелпер ниже. Payments retention, PR #2754.
ESTIMATE_READABLE_SQL = "(expires_at > NOW() OR retain_until > NOW())"


def estimate_readable(expires_at: datetime, retain_until: datetime | None) -> bool:
    """Python-зеркало ESTIMATE_READABLE_SQL — та же дизъюнкция, без похода в БД.

    tzinfo-нормализация повторяет прежнюю Python-проверку (estimate_pdf) —
    `.replace(tzinfo=UTC)`, не переизобретается.
    """
    now = datetime.now(tz=UTC)
    if expires_at.replace(tzinfo=UTC) > now:
        return True
    return retain_until is not None and retain_until.replace(tzinfo=UTC) > now


def _assert_estimate_access(created_by: str | None, x_authenticated_user: str | None) -> None:
    """IDOR guard (#690): только владелец оценки или admin могут её читать.

    Зеркало скоупинга /history (#656). 401 — нет заголовка X-Authenticated-User
    (Caddy basic_auth обязателен); 403 — юзер отсутствует в roles.yaml; 404 —
    pilot читает чужую оценку (скрываем существование, не подтверждаем чужой id).
    """
    if not x_authenticated_user:
        raise HTTPException(
            status_code=401,
            detail="no authenticated user (valid session required)",
        )

    from app.core.auth import get_role

    try:
        role = get_role(x_authenticated_user)
    except KeyError:
        logger.warning(
            "user %r authenticated via Caddy but missing from roles.yaml",
            x_authenticated_user,
        )
        raise HTTPException(status_code=403, detail="user not in roles config") from None

    if role == "admin":
        return
    if created_by is None or created_by != x_authenticated_user:
        raise HTTPException(status_code=404, detail="estimate not found")


def _assert_estimate_access_by_id(
    db: Session, estimate_id: UUID, x_authenticated_user: str | None
) -> None:
    """IDOR guard для derived-роутов, не читающих саму оценку (#690).

    Тянет created_by оценки и применяет _assert_estimate_access. 404 если оценки
    нет (как и owner-mismatch — существование не подтверждаем).
    """
    row = db.execute(
        text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
        {"id": str(estimate_id)},
    ).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="estimate not found")
    _assert_estimate_access(row.created_by, x_authenticated_user)


def _resolve_target_house_id(
    db: Session, address: str | None, lat: float | None, lon: float | None
) -> int | None:
    """Резолвит house_id целевого дома по адресу (geo fallback) — для GET-rehydrate.

    Тот же паттерн, что в placement-history / house-analytics: нормализованный
    short_address → houses, иначе tight 100м geo-fallback. Берём первый матч
    (детерминированно — для price_trend/avito_imv достаточно одного дома).
    Best-effort: None при отсутствии адреса/координат/матча (graceful).
    """
    if address:
        row = db.execute(
            text(
                """
                SELECT id FROM houses
                WHERE short_address = tradein_normalize_short_addr(:addr)
                   OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)
                LIMIT 1
                """
            ),
            {"addr": address},
        ).fetchone()
        if row is not None:
            return row.id

    if lat is not None and lon is not None:
        row = db.execute(
            text(
                """
                SELECT id FROM houses
                WHERE geom IS NOT NULL
                  AND ST_DWithin(
                          geom::geography, ST_MakePoint(:lon, :lat)::geography, 100
                      )
                ORDER BY ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
                LIMIT 1
                """
            ),
            {"lat": lat, "lon": lon},
        ).fetchone()
        if row is not None:
            return row.id

    return None


# ── Revival на GET /estimate/{id} (incident 2026-08-10) ─────────────────────
# Заказчик открыл сохранённую ссылку (?id=...) и увидел «НЕДОСТАТОЧНО ДАННЫХ»:
# запись создана ДО фикса оценщика (#oblast-E/#oblast-F, PR #2823/#2825) и
# лежит в БД мёртвой (median_price<=0/NULL), хотя тот же адрес/параметры
# сейчас честно считаются. get_estimate() ниже пытается пересчитать такую
# строку ОДИН раз (throttled) через тот же estimate_quality(), что и POST
# /estimate, и пишет результат В ТУ ЖЕ строку (id/ссылка не меняются). Живую
# строку (median_price>0) этот путь не трогает вообще — сохранённая клиенту
# цена неприкосновенна.
def _precision_to_qc_geo(precision: str | None) -> int | None:
    """Best-effort обратное отображение к estimator._qc_geo_to_precision.

    AggregatedEstimate наружу отдаёт только бакетированный address_precision
    (house/street/approximate), не сырой dadata.qc_geo (0..5) — тот остаётся
    приватным для estimate_quality(). При revival нам нужно записать ЧТО-ТО в
    колонку dadata_qc_geo, чтобы будущие (уже НЕ revival, обычные) GET той же
    теперь-живой строки не откатили address_precision в None. Бакеты 2..5
    (settlement/city/region/unknown) неразличимы ПОСЛЕ _qc_geo_to_precision —
    2 репрезентативно для всех: тот же helper на чтении схлопывает их обратно
    в тот же "approximate", наблюдаемое поведение не меняется.
    """
    if precision == "house":
        return 0
    if precision == "street":
        return 1
    if precision == "approximate":
        return 2
    return None


def _payload_from_dead_row(row: Any) -> TradeInEstimateInput:
    """Восстанавливает вход оценки из мёртвой сохранённой строки для revival.

    Только поля, реально персистящиеся в trade_in_estimates при создании
    (address/lat/lon/area_m2/rooms/floor/total_floors/year_built/house_type/
    repair_state/has_balcony) — CRM-only поля (ownership_type/has_mortgage)
    на расчёт не влияют и не нужны здесь. radius_m НИКОГДА не персистится
    (payload.radius_m живёт только в рамках одного POST-запроса, ни главный
    INSERT, ни _empty_estimate его не пишут) — None здесь даёт тот же
    default-каскад (DEFAULT_RADIUS_M/FALLBACK_RADIUS_M), что у подавляющего
    большинства сохранённых строк (явный радиус выбирает меньшинство).
    consent=None + require_consent=False у вызывающего — revival не новое
    согласие физлица, а служебный recompute уже существующей записи.
    """
    return TradeInEstimateInput(
        address=row.address,
        area_m2=float(row.area_m2),
        rooms=row.rooms,
        floor=row.floor,
        total_floors=row.total_floors,
        year_built=row.year_built,
        house_type=row.house_type,
        repair_state=row.repair_state,
        has_balcony=row.has_balcony,
        lat=row.lat,
        lon=row.lon,
        radius_m=None,
        consent=None,
    )


async def _try_revive_dead_estimate(
    db: Session, estimate_id: UUID, row: Any
) -> AggregatedEstimate | None:
    """Пытается пересчитать «мёртвую» (median_price<=0/NULL) строку на месте.

    Возвращает свежий AggregatedEstimate (estimate_id ПОДМЕНЁН на исходный —
    id/ссылка не меняются) при успехе; None если: (а) throttle ещё не истёк /
    заявку уже забрал параллельный запрос — anti-storm через атомарный
    conditional `UPDATE ... RETURNING` ниже (тот же паттерн, что
    account_quota.increment, #747): WHERE перепроверяет и «мертва ли строка
    сейчас», и «давно ли последняя попытка» НЕПОСРЕДСТВЕННО в БД, а не по
    значению, прочитанному раньше в Python — TOCTOU-гонка между двумя
    параллельными GET невозможна, проигравший просто не дублирует работу;
    (б) пересчёт сам дал 0 (по-прежнему недостаточно данных); (в) пересчёт
    упал с исключением (сеть/геокод/что угодно). Во всех трёх случаях caller
    обязан отдать сохранённую (по-прежнему мёртвую) строку как раньше — НЕ 500.
    """
    claim = db.execute(
        text(
            """
            UPDATE trade_in_estimates
            SET revival_attempted_at = NOW()
            WHERE id = CAST(:id AS uuid)
              AND (median_price <= 0 OR median_price IS NULL)
              AND (
                  revival_attempted_at IS NULL
                  OR revival_attempted_at
                     < NOW() - make_interval(mins => CAST(:throttle AS integer))
              )
            RETURNING id
            """
        ),
        {"id": str(estimate_id), "throttle": settings.trade_in_revival_throttle_minutes},
    ).fetchone()
    db.commit()
    if claim is None:
        logger.info("estimate revival throttled/lost race: id=%s", estimate_id)
        return None

    from app.services.estimator import estimate_quality

    try:
        payload = _payload_from_dead_row(row)
        result = await estimate_quality(
            payload,
            db,
            created_by=row.created_by,
            client_ip=None,
            require_consent=False,
        )
    except Exception:
        logger.exception("estimate revival failed: id=%s address=%r", estimate_id, row.address)
        return None

    temp_id = result.estimate_id
    if result.median_price_rub <= 0:
        logger.info("estimate revival still insufficient data: id=%s", estimate_id)
        db.execute(
            text("DELETE FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
            {"id": str(temp_id)},
        )
        db.commit()
        return None

    # estimate_quality() persists under a BRAND NEW uuid (temp_id) — it has no
    # notion of "recompute this existing row". Copy the computed OUTPUT fields
    # into the ORIGINAL row (id/link contract), then drop the throwaway one.
    # INPUT snapshot (address/area/rooms/...) is untouched — it did not change,
    # only the outputs were recomputed.
    # #incident-2026-08-11: created_at is DELIBERATELY excluded from this SET —
    # it is the client's original request date (printed in /history and in
    # AggregatedEstimate.created_at, see app/schemas/trade_in.py:317-318), NOT
    # a recompute output. It previously got clobbered with the throwaway temp
    # row's created_at (=NOW() at recompute time), which also silently
    # re-sorted the row to the top of `GET /history ORDER BY created_at DESC`.
    # revival_completed_at (migration 256) is the audit trail for "when did a
    # revival LAST successfully rewrite this row" — distinct from
    # revival_attempted_at (255), which is stamped on every claim regardless
    # of outcome (throttle loss / recompute failure included).
    db.execute(
        text(
            """
            UPDATE avito_imv_evaluations
            SET estimate_id = CAST(:orig AS uuid)
            WHERE estimate_id = CAST(:temp AS uuid)
            """
        ),
        {"orig": str(estimate_id), "temp": str(temp_id)},
    )
    db.execute(
        text(
            """
            UPDATE trade_in_estimates SET
                median_price = :median_price,
                range_low = :range_low,
                range_high = :range_high,
                median_price_per_m2 = :median_ppm2,
                confidence = :confidence,
                confidence_explanation = :explanation,
                n_analogs = :n_analogs,
                analogs = CAST(:analogs_json AS jsonb),
                actual_deals = CAST(:deals_json AS jsonb),
                sources_used = CAST(:sources_json AS jsonb),
                data_freshness_minutes = :freshness,
                canonical_address = :canonical_address,
                house_cadnum = :house_cadnum,
                house_fias_id = :house_fias_id,
                dadata_qc_geo = :dadata_qc_geo,
                dadata_metro = CAST(:dadata_metro_json AS jsonb),
                expected_sold_price = :expected_sold_price,
                expected_sold_range_low = :expected_sold_range_low,
                expected_sold_range_high = :expected_sold_range_high,
                expected_sold_per_m2 = :expected_sold_per_m2,
                asking_to_sold_ratio = :asking_to_sold_ratio,
                ratio_basis = :ratio_basis,
                relaxations = CAST(:relaxations_json AS jsonb),
                reliability = :reliability,
                revival_completed_at = NOW()
            WHERE id = CAST(:id AS uuid)
            """
        ),
        {
            "id": str(estimate_id),
            "median_price": result.median_price_rub,
            "range_low": result.range_low_rub,
            "range_high": result.range_high_rub,
            "median_ppm2": result.median_price_per_m2,
            "confidence": result.confidence,
            "explanation": result.confidence_explanation,
            "n_analogs": result.n_analogs,
            "analogs_json": json.dumps(
                [a.model_dump(mode="json") for a in result.analogs], ensure_ascii=False
            ),
            "deals_json": json.dumps(
                [a.model_dump(mode="json") for a in result.actual_deals], ensure_ascii=False
            ),
            "sources_json": json.dumps(result.sources_used, ensure_ascii=False),
            "freshness": result.data_freshness_minutes,
            "canonical_address": result.canonical_address,
            "house_cadnum": result.house_cadnum,
            "house_fias_id": result.house_fias_id,
            "dadata_qc_geo": _precision_to_qc_geo(result.address_precision),
            "dadata_metro_json": json.dumps(result.metro_nearest, ensure_ascii=False),
            "expected_sold_price": result.expected_sold_price_rub,
            "expected_sold_range_low": result.expected_sold_range_low_rub,
            "expected_sold_range_high": result.expected_sold_range_high_rub,
            "expected_sold_per_m2": result.expected_sold_per_m2,
            "asking_to_sold_ratio": result.asking_to_sold_ratio,
            "ratio_basis": result.ratio_basis,
            "relaxations_json": json.dumps(result.relaxations, ensure_ascii=False),
            "reliability": result.reliability,
        },
    )
    db.execute(
        text("DELETE FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
        {"id": str(temp_id)},
    )
    db.commit()

    logger.info(
        "estimate revived: id=%s median=%d n=%d confidence=%s reliability=%s",
        estimate_id,
        result.median_price_rub,
        result.n_analogs,
        result.confidence,
        result.reliability,
    )
    # created_at on the returned object must mirror the DB row (untouched
    # original request date, NOT the temp row's NOW()) — see UPDATE above.
    return result.model_copy(update={"estimate_id": estimate_id, "created_at": row.created_at})


@router.post("/estimate", response_model=AggregatedEstimate)
async def estimate(
    payload: TradeInEstimateInput,
    request: Request,
    response: Response,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> AggregatedEstimate:
    """Реальная оценка через SQL aggregation поверх listings + deals.

    1. Geocode address → lat/lon
    2. PostGIS ST_DWithin радиус 800м (или 2км fallback)
    3. Tukey IQR outlier filter
    4. Median + Q1 + Q3 + confidence с explanation

    Применяется лимит 15 успешных оценок в месяц на аккаунт (кроме admin/kopylov);
    анонимные запросы (без X-Authenticated-User) — свой, гораздо более строгий
    лимит на anon-сессию+IP (#b2c-antiabuse-2).
    """
    # #b2c-antiabuse-2 п.4: отдельный жёсткий лимит частоты на дорогой публичный
    # путь — самая дешёвая проверка первой, до квоты и до дорогой цепочки внешних
    # вызовов. Ключ user:/ip: — тот же принцип, что общий RateLimitMiddleware
    # (app/main.py); НЕ anon-сессия квоты (rate limit — про network-identity и
    # burst-защиту capacity сервера, а не про месячный business-лимит).
    _rl_key = (
        f"user:{x_authenticated_user}" if x_authenticated_user else f"ip:{_client_ip(request)}"
    )
    _retry_after = _estimate_limiter.check(_rl_key)
    if _retry_after is not None:
        raise HTTPException(
            status_code=429,
            detail="Слишком много запросов на оценку. Попробуйте через несколько минут.",
            headers={"Retry-After": str(int(_retry_after) + 1)},
        )

    quota_key, quota_default_limit = _resolve_quota_identity(
        request, response, x_authenticated_user
    )
    account_quota.check_and_raise(db, quota_key, default_limit=quota_default_limit)
    from app.services.estimator import estimate_quality

    # #3082: слот одновременности берём ПОСЛЕ дешёвых отказов (рейт-лимит, квота)
    # и ДО дорогой цепочки внешних вызовов; release — в finally ниже.
    try:
        await asyncio.wait_for(_estimate_slots.acquire(), timeout=_ESTIMATE_SLOT_WAIT_S)
    except TimeoutError:
        raise HTTPException(
            status_code=429,
            detail="Сервис оценки сейчас занят. Попробуйте ещё раз через несколько секунд.",
            headers={"Retry-After": "5"},
        ) from None

    # #654: ранее любое исключение estimate_quality всплывало необработанным и
    # маскировалось апстрим-прокси (Caddy) как непрозрачный 502. Ловим, логируем
    # через logger.exception (→ GlitchTip/Sentry получает stack trace) и отдаём
    # явный 503 — так любая БУДУЩАЯ реальная ошибка становится видимой, а не
    # «глотается» шлюзом. HTTPException пробрасываем как есть (это не сбой).
    # created_by (#656) прокидываем в estimate_quality для скоупа /history — ТОЛЬКО
    # реальный account username (НЕ anon-ключ квоты): у анонимов нет "аккаунта",
    # по которому имеет смысл скоупить /history.
    # ЭТАП 4 B2C (152-ФЗ): require_consent=True только когда нет
    # X-Authenticated-User — сегодня rbac_guard (app/core/rbac.py) уже требует
    # этот заголовок на любом non-public пути, так что эта ветка пока
    # недостижима в проде (анонимный /estimate ещё не открыт другими частями
    # ЭТАП 4/B2C работ) — гейт готов ЗАРАНЕЕ, на момент открытия анонимного
    # доступа. client_ip — proof-of-consent (estimate_quality персистит его
    # на trade_in_estimates только когда require_consent=True; B2B-пилоты
    # остаются NULL, см. estimator.py::_estimate_consent_persist_fields).
    try:
        result = await estimate_quality(
            payload,
            db,
            created_by=x_authenticated_user,
            client_ip=_client_ip(request),
            require_consent=x_authenticated_user is None,
        )
    except HTTPException:
        raise
    except Exception:
        logger.exception("estimate failed for address=%r", payload.address)
        raise HTTPException(
            status_code=503,
            detail="estimate temporarily unavailable — try again shortly",
        ) from None
    finally:
        # #3082: слот возвращаем сразу после дорогой части — инкремент квоты и
        # сериализация ответа ниже дёшевы и слот держать не должны.
        _estimate_slots.release()
    # #747: атомарно-условный инкремент — источник истины по лимиту. check_and_raise
    # выше остаётся быстрым pre-check (429 до дорогой оценки), но финальное решение
    # тут: при гонке двух /estimate на used=lim-1 второй получит False.
    # Не списываем квоту за пустой результат (нерезолвящийся адрес и т.п.) — иначе
    # платный слот сгорает за insufficient_data=True (median=0, n_analogs=0) с HTTP 200.
    if not result.insufficient_data and not account_quota.increment(
        db, quota_key, default_limit=quota_default_limit
    ):
        # Аутентифицированный путь — байт-в-байт прежнее сообщение (может не
        # отражать персональный override, это pre-existing поведение, вне
        # scope этого фикса). Анонимный путь — динамический текст с ПРАВИЛЬНЫМ
        # anon-лимитом (#b2c-antiabuse-2), а не захардкоженным MONTHLY_LIMIT.
        detail = (
            account_quota.LIMIT_EXHAUSTED_MESSAGE
            if x_authenticated_user
            else account_quota.limit_exhausted_message(quota_default_limit)
        )
        raise HTTPException(status_code=429, detail=detail)

    # Feature 2/3 foundation: "что искали" — обогащённая estimate_request-запись
    # в user_events (адрес/площадь/комнаты + estimate_id для join с trade_in_estimates).
    # schedule_event сам никогда не raises — сбой аудита не должен ронять ответ.
    schedule_event(
        event_type="estimate_request",
        username=x_authenticated_user or "",
        ip=_client_ip(request),
        user_agent=request.headers.get("user-agent"),
        path=str(request.url.path),
        method="POST",
        estimate_id=str(result.estimate_id),
        payload={
            "address": payload.address,
            "area_m2": payload.area_m2,
            "rooms": payload.rooms,
        },
    )
    return result


@router.get("/quota", response_model=QuotaStatus)
def get_quota(
    request: Request,
    response: Response,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> QuotaStatus:
    """Статус квоты оценок для текущего аккаунта (или анонимной сессии).

    Возвращает limit / used / remaining / unlimited для X-Authenticated-User.
    Без заголовка (прод, публичный путь) — статус СВОЕЙ анонимной квоты
    (anon-session cookie + IP), а НЕ безлимитный, если явно не включён
    settings.quota_dev_fail_open (#b2c-antiabuse-2, dev без Caddy).
    """
    quota_key, quota_default_limit = _resolve_quota_identity(
        request, response, x_authenticated_user
    )
    status = account_quota.get_status(db, quota_key, default_limit=quota_default_limit)
    return QuotaStatus(**status)


@router.get("/estimate/{estimate_id}", response_model=AggregatedEstimate)
def get_estimate(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> AggregatedEstimate:
    """Получить сохранённую оценку по UUID (для генерации PDF).

    Возвращает 404 если оценка не найдена или TTL истёк.
    """
    return load_estimate(db, estimate_id, x_authenticated_user=x_authenticated_user)


def load_estimate(
    db: Session,
    estimate_id: UUID,
    *,
    x_authenticated_user: str | None,
    capability_granted: bool = False,
) -> AggregatedEstimate:
    """Тело GET /estimate/{id} без FastAPI-обвязки — чтобы у ВТОРОГО права
    доступа был тот же самый загрузчик, а не его копия.

    `capability_granted=True` — вызывающая сторона уже доказала право доступа
    ДРУГИМ способом, чем `X-Authenticated-User` + roles.yaml: capability-ссылка
    `/r/<token>` (app/api/v1/payments.py) отдаёт оплаченный отчёт анониму,
    у которого идентичности нет и не будет — там правом является сам
    непредсказуемый токен, сверенный по `payment_entitlements`.

    Флаг — ИМЕННО параметр обычной функции, а не поле запроса: у route-хендлера
    `get_estimate` выше его нет, поэтому подобрать его снаружи (query/заголовком)
    невозможно — включить его может только код в этом процессе. Обратное
    (добавить параметр в сам хендлер с default=False) сделало бы обход IDOR-
    гварда #690 доступным любому клиенту через `?capability_granted=true`.
    """
    row = db.execute(
        text(
            f"""
            SELECT id, median_price, range_low, range_high, median_price_per_m2,
                   confidence, confidence_explanation, n_analogs,
                   market_percentile,
                   analogs, actual_deals, sources_used, data_freshness_minutes,
                   expires_at, retain_until, address, lat, lon,
                   area_m2, rooms, floor, total_floors,
                   year_built, house_type, repair_state, has_balcony,
                   canonical_address, house_cadnum, house_fias_id,
                   dadata_qc_geo, dadata_metro,
                   expected_sold_price, expected_sold_range_low,
                   expected_sold_range_high, expected_sold_per_m2,
                   asking_to_sold_ratio, ratio_basis, created_by, created_at,
                   relaxations, reliability
            FROM trade_in_estimates
            WHERE id = CAST(:id AS uuid)
              AND {ESTIMATE_READABLE_SQL}
            """
        ),
        {"id": str(estimate_id)},
    ).fetchone()

    if row is None:
        raise HTTPException(status_code=404, detail="estimate not found or expired")

    if not capability_granted:
        _assert_estimate_access(row.created_by, x_authenticated_user)

    # #incident-2026-08-10: строка «мертва» (median_price<=0/NULL) — посчитана
    # ДО фикса оценщика (#oblast-E/#oblast-F, PR #2823/#2825). Пробуем
    # пересчитать её на месте (throttled, race-safe — см. докстринг
    # _try_revive_dead_estimate) через тот же путь, что и POST /estimate.
    # Живую строку (median_price>0) не трогаем вообще. asyncio.run() — sync↔
    # async мост (тот же паттерн, что app/scheduler_main.py): get_estimate
    # остаётся `def` (Starlette гоняет его в threadpool, как сейчас), поэтому
    # ОСТАЛЬНЫЕ синхронные db.execute() ниже по функции не переезжают на event
    # loop — только сама попытка revival временно занимает свой поток на время
    # await estimate_quality(). Любая ошибка расчёта — не 500: revived is None,
    # и функция просто продолжает как раньше, отдавая сохранённую строку.
    if row.median_price is None or row.median_price <= 0:
        revived = asyncio.run(_try_revive_dead_estimate(db, estimate_id, row))
        if revived is not None:
            return revived

    from app.services.estimator import (
        _canonical_sources,
        _cv_from_ppm2,
        _fetch_dkp_corridor,
        _fetch_house_imv_anchor,
        _fetch_price_trend,
        _qc_geo_to_precision,
        _resolve_target_city,
        _source_counts,
        rehydrate_search_radius_m,
    )

    analogs = [AnalogLot(**a) for a in (row.analogs or [])]
    actual_deals = [AnalogLot(**a) for a in (row.actual_deals or [])]

    # #2632: search_radius_m колонкой не персистится — восстанавливаем его из
    # того, что персистится (подпись каскада «радиус расширен до N м», иначе
    # размах сохранённых аналогов). Без этого GET отдавал null, фронт падал на
    # превью-радиус 1 км и рисовал круг, за которым лежат его же пины (прод
    # 2026-08-11: 10 из 10 аналогов вне круга, самый дальний — 4381 м).
    persisted_relaxations = list(getattr(row, "relaxations", None) or [])
    search_radius_m = rehydrate_search_radius_m(
        persisted_relaxations, [a.distance_m for a in analogs]
    )

    # #2043 (BE-1): CV / счётчики источников на rehydrate — best-effort из
    # сохранённых analogs (top-N, усечённо: полная выборка не персистится). На
    # свежей оценке (POST) считаются по полной выборке; здесь — по тому, что есть
    # в строке. created_at берём из колонки (persisted).
    cv = _cv_from_ppm2([a.price_per_m2 for a in analogs])
    source_counts = _source_counts([a.source for a in analogs])
    # #2087 (M1): sources_used рехайдрейтим тем же helper'ом, что и POST — из ТЕХ ЖЕ
    # persisted analogs (листинговая часть) ∪ оценочных флагов, вычитанных из
    # persisted-колонки sources_used (avito_imv/yandex_valuation/cian_valuation).
    # Так один estimate_id → идентичный sources_used на POST/GET/reload, а
    # source_counts.keys() ⊆ sources_used (обе части — из одного набора analogs).
    # Чинит рассинхром и на СТАРЫХ строках (где колонка хранит радиусный набор +
    # quarter_index): листинговая часть пересобирается из analogs, служебный шум
    # отбрасывается фильтром helper'а.
    sources_used = _canonical_sources((a.source for a in analogs), (row.sources_used or []))

    # #696: POST-only производные поля (price_trend / avito_imv / dkp_corridor /
    # last_scraped_at) на GET-rehydrate ранее были null → на shared-link / PDF /
    # ?id= restore пропадали графики тренда, IMV-якорь, коридор ДКП и метка
    # свежести. Пересчитываем их здесь из персистированного состояния (house_id
    # ререзолвится по адресу/гео; last_scraped_at реконструируется из
    # created_at − data_freshness_minutes — оба поля персистятся). Всё best-effort:
    # None при отсутствии данных (без регрессий по сравнению с прежним поведением).
    area_f = float(row.area_m2) if row.area_m2 is not None else None
    target_house_id = _resolve_target_house_id(db, row.address, row.lat, row.lon)

    price_trend_raw = _fetch_price_trend(db, target_house_id=target_house_id)
    price_trend = (
        [PriceTrendPoint(month=p["month"], ppm2=p["ppm2"]) for p in price_trend_raw]
        if price_trend_raw
        else None
    )

    # (oblast C2): city-scope корридора на GET-rehydrate. NB: row.address здесь =
    # geo.full_address (персистится на estimate-time), который РОНЯЕТ город для
    # не-ЕКБ («Ленина, 1») → target_city=None → corridor unscoped/None для не-ЕКБ
    # (KNOWN LIMITATION: POST-path чинит через payload.address; GET получит паритет
    # когда raw payload.address начнёт персиститься — follow-up). Для ЕКБ ок; reorder
    # ниже безвреден (оба source city-stripped для не-ЕКБ, оба екб для ЕКБ).
    target_city = _resolve_target_city(row.address) or _resolve_target_city(row.canonical_address)
    # #3051 PR-A: тот же region_code-скоуп, что и POST /estimate — гарантирует
    # «регион не резолвится → DEFAULT_REGION_CODE (66)», не NULL (NULL в SQL
    # обнулил бы фильтр). row.lat/row.lon персистятся с estimate-time.
    target_region = (
        regions_mod.region_for_point(row.lat, row.lon)
        if row.lat is not None and row.lon is not None
        else None
    )
    target_region_code = target_region.code if target_region else regions_mod.DEFAULT_REGION_CODE
    dkp_raw = _fetch_dkp_corridor(
        db,
        address=row.address,
        rooms=row.rooms,
        area=area_f,
        city=target_city,
        region_code=target_region_code,
    )
    dkp_corridor = DkpCorridor(**dkp_raw) if dkp_raw else None

    imv_raw = _fetch_house_imv_anchor(
        db, target_house_id=target_house_id, rooms=row.rooms, area=area_f
    )
    avito_imv = (
        AvitoImvSummary(
            recommended_price=int(imv_raw["recommended_price"]),
            lower_price=int(imv_raw["lower_price"]) if imv_raw.get("lower_price") else None,
            higher_price=int(imv_raw["higher_price"]) if imv_raw.get("higher_price") else None,
            # #3323: `is not None` (0 — самый тонкий рынок, не «неизвестно») + thin_market
            # считаем тем же порогом, что POST-путь в estimator, иначе одна и та же
            # оценка при переоткрытии по ссылке / в PDF теряла флаг тонкого рынка.
            market_count=(
                int(imv_raw["market_count"]) if imv_raw.get("market_count") is not None else None
            ),
            thin_market=(
                imv_raw.get("market_count") is not None
                and int(imv_raw["market_count"]) < settings.avito_imv_thin_market_threshold
            ),
        )
        if imv_raw is not None and imv_raw.get("recommended_price")
        else None
    )

    last_scraped_at = (
        row.created_at - timedelta(minutes=row.data_freshness_minutes)
        if row.created_at is not None and row.data_freshness_minutes is not None
        else None
    )

    # ВАЖНО: возвращаем ПОЛНЫЙ набор полей. Раньше эндпоинт отдавал огрызок
    # без sources_used / confidence_explanation / координат — и при открытии
    # оценки по ссылке (?id=) карточка источников пустела до «0/7».
    # DaData-обогащёнка (canonical/cadnum/fias/precision/metro) тоже
    # rehydrate'ится из row — иначе бейдж точности + метро не показывались
    # на shared-link reopen / в PDF.
    return AggregatedEstimate(
        estimate_id=row.id,
        median_price_rub=row.median_price,
        range_low_rub=row.range_low,
        range_high_rub=row.range_high,
        median_price_per_m2=row.median_price_per_m2,
        confidence=row.confidence,
        confidence_explanation=row.confidence_explanation,
        n_analogs=row.n_analogs,
        # #2899: getattr — тот же defensive-идиом, что у relaxations/reliability
        # ниже: строка без колонки (старый in-memory double, любая выборка до
        # миграции 267) деградирует в None — «позиции не знаем», — а не роняет
        # ответ AttributeError'ом.
        market_percentile=getattr(row, "market_percentile", None),
        period_months=12,
        analogs=analogs,
        actual_deals=actual_deals,
        expires_at=row.expires_at,
        retain_until=row.retain_until,
        target_address=row.address,
        target_lat=row.lat,
        target_lon=row.lon,
        sources_used=sources_used,
        data_freshness_minutes=row.data_freshness_minutes,
        # #696 — POST-only производные поля, пересчитанные на rehydrate (см. выше).
        last_scraped_at=last_scraped_at,
        price_trend=price_trend,
        avito_imv=avito_imv,
        dkp_corridor=dkp_corridor,
        # #648 Stage 3 — sold-correction columns rehydrated for shared-link reopen.
        # (numeric ratio → float for the Pydantic field.)
        expected_sold_price_rub=row.expected_sold_price,
        expected_sold_range_low_rub=row.expected_sold_range_low,
        expected_sold_range_high_rub=row.expected_sold_range_high,
        expected_sold_per_m2=row.expected_sold_per_m2,
        asking_to_sold_ratio=(
            float(row.asking_to_sold_ratio) if row.asking_to_sold_ratio is not None else None
        ),
        ratio_basis=row.ratio_basis,
        area_m2=row.area_m2,
        rooms=row.rooms,
        floor=row.floor,
        total_floors=row.total_floors,
        year_built=row.year_built,
        house_type=row.house_type,
        repair_state=row.repair_state,
        has_balcony=row.has_balcony,
        canonical_address=row.canonical_address,
        house_cadnum=row.house_cadnum,
        house_fias_id=row.house_fias_id,
        address_precision=_qc_geo_to_precision(row.dadata_qc_geo),
        metro_nearest=(row.dadata_metro or []),
        # #2043 (BE-1): достоверность выборки — CV, счётчики источников, дата создания.
        cv=cv,
        source_counts=source_counts,
        created_at=row.created_at,
        # PR #2823 open follow-up (fixed incident 2026-08-10, migration 255):
        # relaxations/reliability теперь персистятся — GET-rehydrate больше не
        # теряет красный баннер «точность снижена» при открытии по ссылке.
        # getattr defensive: старые in-memory test doubles / любая строка без
        # этих колонок (не должно случаться после миграции) деградируют в
        # дефолт схемы (ok / []), а не падают AttributeError.
        relaxations=persisted_relaxations,
        reliability=getattr(row, "reliability", None) or "ok",
        # #2632: фактический радиус подбора (реконструкция выше). requested_radius_m
        # осознанно НЕ заполняем — payload.radius_m не персистится, и подставить
        # сюда дефолт значило бы выдать догадку за то, что просил пользователь.
        search_radius_m=search_radius_m,
    )


@router.get("/estimate/{estimate_id}/pdf")
def estimate_pdf(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> Response:
    """Скачать 4-страничный PDF-отчёт для оценки trade-in.

    Бренд PDF определяется по владельцу оценки (estimate.created_by → brand slug),
    а НЕ из ?brand= query param (#7 brand-spoofing fix).

    Возвращает application/pdf attachment.
    404 — оценка не найдена.
    410 — оценка просрочена (TTL 24ч).
    """
    row = db.execute(
        text(
            """
            SELECT id, median_price, range_low, range_high, median_price_per_m2,
                   confidence, confidence_explanation, n_analogs,
                   market_percentile,
                   analogs, actual_deals, sources_used, data_freshness_minutes,
                   expires_at, retain_until,
                   address, lat, lon, area_m2, rooms, floor, total_floors,
                   year_built, house_type, repair_state, has_balcony,
                   canonical_address, house_cadnum, house_fias_id,
                   dadata_qc_geo, dadata_metro,
                   expected_sold_price, expected_sold_range_low,
                   expected_sold_range_high, expected_sold_per_m2,
                   asking_to_sold_ratio, ratio_basis, created_by,
                   relaxations, reliability
            FROM trade_in_estimates
            WHERE id = CAST(:id AS uuid)
            """
        ),
        {"id": str(estimate_id)},
    ).fetchone()

    if row is None:
        raise HTTPException(status_code=404, detail="estimate not found")

    _assert_estimate_access(row.created_by, x_authenticated_user)

    # PR-D1: тот же гейт, что в get_estimate (см. ESTIMATE_READABLE_SQL) — раньше
    # здесь была независимая Python-проверка expires_at, разошедшаяся с SQL-
    # фильтром GET-ручки. "estimate expired (24h TTL)" убрано из текста: при
    # годовом retain_until упоминание 24ч в ответе API стало бы ложью.
    if not estimate_readable(row.expires_at, row.retain_until):
        raise HTTPException(status_code=410, detail="estimate expired")

    from app.services.estimator import _qc_geo_to_precision

    analogs = [AnalogLot(**a) for a in (row.analogs or [])]
    actual_deals = [AnalogLot(**a) for a in (row.actual_deals or [])]

    estimate = AggregatedEstimate(
        estimate_id=row.id,
        median_price_rub=row.median_price,
        range_low_rub=row.range_low,
        range_high_rub=row.range_high,
        median_price_per_m2=row.median_price_per_m2,
        confidence=row.confidence,
        confidence_explanation=row.confidence_explanation,
        n_analogs=row.n_analogs,
        # #2899: getattr — тот же defensive-идиом, что у relaxations/reliability
        # ниже: строка без колонки (старый in-memory double, любая выборка до
        # миграции 267) деградирует в None — «позиции не знаем», — а не роняет
        # ответ AttributeError'ом.
        market_percentile=getattr(row, "market_percentile", None),
        # #1351: окно сделок — 12 мес (estimator.DEALS_PERIOD_MONTHS), как в POST
        # /estimate и GET /estimate/{id}. Раньше PDF-ветка хардкодила 24 →
        # экспортёр рисовал ложный ~2-летний диапазон в клиентском документе.
        period_months=12,
        analogs=analogs,
        actual_deals=actual_deals,
        expires_at=row.expires_at,
        retain_until=row.retain_until,
        target_address=row.address,
        target_lat=row.lat,
        target_lon=row.lon,
        sources_used=row.sources_used or [],
        data_freshness_minutes=row.data_freshness_minutes,
        # #648 Stage 3 — sold-correction columns rehydrated so the PDF carries them.
        expected_sold_price_rub=row.expected_sold_price,
        expected_sold_range_low_rub=row.expected_sold_range_low,
        expected_sold_range_high_rub=row.expected_sold_range_high,
        expected_sold_per_m2=row.expected_sold_per_m2,
        asking_to_sold_ratio=(
            float(row.asking_to_sold_ratio) if row.asking_to_sold_ratio is not None else None
        ),
        ratio_basis=row.ratio_basis,
        canonical_address=row.canonical_address,
        house_cadnum=row.house_cadnum,
        house_fias_id=row.house_fias_id,
        address_precision=_qc_geo_to_precision(row.dadata_qc_geo),
        metro_nearest=(row.dadata_metro or []),
        # migration 255 — та же сноска «точность снижена», что и на JSON GET,
        # теперь и в PDF-регенерации сохранённой оценки (см. get_estimate).
        relaxations=list(getattr(row, "relaxations", None) or []),
        reliability=getattr(row, "reliability", None) or "ok",
    )
    input_snapshot = {
        "address": row.address,
        "area_m2": row.area_m2,
        "rooms": row.rooms,
        "floor": row.floor,
        "total_floors": row.total_floors,
        "year_built": row.year_built,
        "house_type": row.house_type,
        "repair_state": row.repair_state,
        "has_balcony": row.has_balcony,
    }

    from app.core.auth import get_brand_for_user as _brand_for_user
    from app.services.brand import get_brand as _resolve_brand

    # #7 brand-spoofing fix: бренд определяется по владельцу оценки,
    # а не по ?brand= query param (который был удалён из сигнатуры).
    owner_brand_slug = _brand_for_user(row.created_by) if row.created_by else None
    brand_obj = _resolve_brand(owner_brand_slug, db)
    pdf_bytes = generate_trade_in_pdf(estimate, input_snapshot, brand=brand_obj)
    filename = f"trade-in-{brand_obj.slug}-{estimate_id}.pdf"
    logger.info(
        "PDF generated estimate_id=%s brand=%s size=%d",
        estimate_id,
        brand_obj.slug,
        len(pdf_bytes),
    )
    return Response(
        content=pdf_bytes,
        media_type="application/pdf",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


# ── Фото квартиры (#394) ─────────────────────────────────────────────────────
_MAX_PHOTO_BYTES = 10 * 1024 * 1024  # 10 МБ на фото
_MAX_PHOTOS_PER_ESTIMATE = 12
_ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/heic"}


@router.post("/estimate/{estimate_id}/photos", response_model=PhotoMeta)
async def upload_photo(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    file: Annotated[UploadFile, File()],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> PhotoMeta:
    """Загрузить фото квартиры к оценке (#394). Хранение в estimate_photos (bytea)."""
    estimate = db.execute(
        text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
        {"id": str(estimate_id)},
    ).fetchone()
    if estimate is None:
        raise HTTPException(status_code=404, detail="estimate not found")
    _assert_estimate_access(estimate.created_by, x_authenticated_user)

    ctype = (file.content_type or "").lower()
    if ctype not in _ALLOWED_IMAGE_TYPES:
        raise HTTPException(
            status_code=415, detail=f"unsupported content-type: {ctype or 'unknown'}"
        )

    count = db.execute(
        text("SELECT count(*) FROM estimate_photos WHERE estimate_id = CAST(:id AS uuid)"),
        {"id": str(estimate_id)},
    ).scalar_one()
    if count >= _MAX_PHOTOS_PER_ESTIMATE:
        raise HTTPException(
            status_code=409, detail=f"photo limit reached ({_MAX_PHOTOS_PER_ESTIMATE})"
        )

    # #2233: читаем тело чанками с жёстким капом, а не await file.read() целиком —
    # иначе multi-GB аплоад буферизуется в RAM и OOM-killed backend (mem_limit 768m, #2214).
    # 413 бросается СРАЗУ при превышении, остаток тела запроса НЕ читается.
    chunks: list[bytes] = []
    total = 0
    while chunk := await file.read(64 * 1024):
        total += len(chunk)
        if total > _MAX_PHOTO_BYTES:
            raise HTTPException(status_code=413, detail="file too large (max 10 MB)")
        chunks.append(chunk)
    content = b"".join(chunks)
    if not content:
        raise HTTPException(status_code=400, detail="empty file")

    # Sanitize: re-encode through Pillow to drop EXIF, kill polyglot payloads,
    # cap dimensions. Closes finding #6 from 2026-05-24 audit.
    try:
        sanitized_bytes, sanitized_ctype = sanitize_image(content)
    except ImageSanitizationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e

    row = (
        db.execute(
            text(
                """
            INSERT INTO estimate_photos
                (estimate_id, filename, content_type, content, size_bytes)
            VALUES (CAST(:eid AS uuid), :fn, :ct, :content, :sz)
            RETURNING id, filename, content_type, size_bytes, uploaded_at
            """
            ),
            {
                "eid": str(estimate_id),
                "fn": file.filename,
                "ct": sanitized_ctype,
                "content": sanitized_bytes,
                "sz": len(sanitized_bytes),
            },
        )
        .mappings()
        .fetchone()
    )
    db.commit()
    logger.info(
        "photo uploaded: estimate=%s photo=%s orig_size=%d sanitized_size=%d ctype=%s",
        estimate_id,
        row["id"],
        len(content),
        len(sanitized_bytes),
        sanitized_ctype,
    )
    return PhotoMeta(**row)


@router.get("/estimate/{estimate_id}/photos", response_model=list[PhotoMeta])
def list_photos(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[PhotoMeta]:
    """Список фото оценки — метаданные, без содержимого (#394)."""
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
    rows = (
        db.execute(
            text(
                """
            SELECT id, filename, content_type, size_bytes, uploaded_at
            FROM estimate_photos
            WHERE estimate_id = CAST(:id AS uuid)
            ORDER BY uploaded_at
            """
            ),
            {"id": str(estimate_id)},
        )
        .mappings()
        .all()
    )
    return [PhotoMeta(**r) for r in rows]


@router.get("/estimate/{estimate_id}/photos/{photo_id}")
def get_photo(
    estimate_id: UUID,
    photo_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> Response:
    """Отдать содержимое фото — image bytes (#394)."""
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
    row = db.execute(
        text(
            """
            SELECT content, content_type
            FROM estimate_photos
            WHERE id = CAST(:pid AS uuid) AND estimate_id = CAST(:eid AS uuid)
            """
        ),
        {"pid": str(photo_id), "eid": str(estimate_id)},
    ).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="photo not found")
    return Response(
        content=bytes(row.content),
        media_type=row.content_type,
        headers={"Cache-Control": "private, max-age=3600"},
    )


# ── История и кэш (#399) ─────────────────────────────────────────────────────
@router.get("/history")
def estimate_history(
    db: Annotated[Session, Depends(get_db)],
    limit: int = 50,
    account: str | None = None,
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[dict[str, object]]:
    """История оценок (#399) — последние N записей trade_in_estimates.

    Скоупинг (#656 — закрывает cross-pilot data-leak): non-admin видит ТОЛЬКО
    свои оценки (created_by = X-Authenticated-User); legacy NULL-строки без
    владельца не попадают. Admin видит все строки, либо фильтрует по ?account=<user>.
    401 если заголовок отсутствует (mirror /me — Caddy basic_auth обязателен).

    #2043 (BE-1): в проекцию добавлен sources_used (jsonb-массив источников) —
    разлочивает колонку «ИСТОЧНИКОВ N/7» в CacheView (len(sources_used) на строку).

    #2417: добавлены floor/total_floors/year_built/house_type/repair_state/
    has_balcony — уже персистятся в trade_in_estimates при создании оценки
    (см. TradeInEstimateInput / AggregatedEstimate), но раньше не выбирались
    здесь. Nullable — legacy строки / незаполненные поля отдаются как None.
    """
    if not x_authenticated_user:
        raise HTTPException(
            status_code=401,
            detail="no authenticated user (valid session required)",
        )

    from app.core.auth import get_role

    try:
        role = get_role(x_authenticated_user)
    except KeyError:
        logger.warning(
            "user %r authenticated via Caddy but missing from roles.yaml",
            x_authenticated_user,
        )
        raise HTTPException(status_code=403, detail="user not in roles config") from None

    params: dict[str, object] = {"limit": min(max(limit, 1), 200)}
    where = ""
    if role == "admin":
        # Admin: все строки, либо фильтр по конкретному аккаунту.
        if account:
            where = "WHERE created_by = :account"
            params["account"] = account
    else:
        # Non-admin: жёстко скоупим на свои строки; ?account игнорируется.
        where = "WHERE created_by = :owner"
        params["owner"] = x_authenticated_user

    rows = (
        db.execute(
            text(
                f"""
            SELECT id, address, rooms, area_m2, median_price,
                   confidence, n_analogs, created_at,
                   COALESCE(sources_used, '[]'::jsonb) AS sources_used,
                   floor, total_floors, year_built,
                   house_type, repair_state, has_balcony
            FROM trade_in_estimates
            {where}
            ORDER BY created_at DESC
            LIMIT :limit
            """
            ),
            params,
        )
        .mappings()
        .all()
    )
    return [dict(r) for r in rows]


@router.get("/cache-stats")
def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]:
    """Состояние данных и кэшей (#399) — для страницы «Кэш».

    #2043 (BE-1) KPI-расширение для CacheView:
      - avg_median_price — средняя headline-медиана по НЕпустым оценкам
        (median_price > 0, чтобы insufficient-data нули не занижали среднее).
        NULL если непустых оценок нет.
      - repeat_address_pct — доля строк-оценок, чей address встречается ≥2 раз
        (индикатор потенциала кэш-хитов «повторный адрес»). Считается по
        trade_in_estimates с непустым address; NULL при отсутствии адресов.
        NB: это честный best-effort по persisted оценкам, а не hit-rate реального
        кэша (отдельного счётчика попаданий не ведём).

    #2660: listings_active сам по себе врал — «активно» на проде не означает
    «живо» (деактиватор протухших покрывает не все источники). Рядом отдаём
    listings_active_stale — сколько из них не виделись listings_stale_days
    (= LISTINGS_FRESH_DAYS эстиматора; прод 2026-08-05: 37 900 активных при
    20 935 не виденных 14+ дней). Счётчик не прячем, а разделяем.
    """
    from app.services.estimator import LISTINGS_FRESH_DAYS

    row = (
        db.execute(
            text(
                """
            SELECT
              (SELECT count(*) FROM geocode_cache) AS geocode_cache,
              (SELECT count(*) FROM geocode_cache WHERE expires_at > NOW())
                  AS geocode_cache_fresh,
              (SELECT count(*) FROM listings WHERE is_active) AS listings_active,
              (SELECT count(*) FROM listings
                WHERE is_active
                  AND last_seen_at <= NOW() - (:fresh_days || ' days')::interval)
                  AS listings_active_stale,
              (SELECT max(scraped_at) FROM listings) AS listings_last_scraped,
              (SELECT count(*) FROM deals) AS deals,
              (SELECT count(*) FROM gendesign_cad_buildings) AS cad_buildings,
              (SELECT count(*) FROM house_metadata) AS house_metadata,
              (SELECT count(*) FROM trade_in_estimates) AS estimates_total,
              (SELECT round(avg(median_price))
                 FROM trade_in_estimates WHERE median_price > 0) AS avg_median_price,
              (SELECT round(
                        100.0 * count(*) FILTER (WHERE addr_count > 1)
                        / NULLIF(count(*), 0), 1)
                 FROM (
                   SELECT count(*) OVER (PARTITION BY address) AS addr_count
                   FROM trade_in_estimates
                   WHERE address IS NOT NULL AND address <> ''
                 ) t) AS repeat_address_pct
            """
            ),
            {"fresh_days": LISTINGS_FRESH_DAYS},
        )
        .mappings()
        .fetchone()
    )
    # Порог отдаём рядом с числом — чтобы UI подписывал «не виделись N дней»,
    # а не заводил второе определение свежести у себя.
    return (dict(row) | {"listings_stale_days": LISTINGS_FRESH_DAYS}) if row else {}


# ── Stage 4a: house info + IMV benchmark для UI ───────────────────────────────


_HOUSE_SELECT_COLS = """
    h.id AS house_id, h.source, h.ext_house_id, h.address, h.short_address,
    h.lat, h.lon, h.year_built, h.total_floors, h.house_type,
    h.passenger_elevators, h.cargo_elevators,
    h.has_concierge, h.closed_yard, h.has_playground, h.parking_type,
    h.developer_name, h.rating, h.reviews_count,
    COALESCE(h.raw_characteristics, '[]'::jsonb) AS raw_characteristics
"""


@router.get("/estimate/{estimate_id}/houses", response_model=list[HouseInfoForEstimate])
def get_estimate_houses(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[HouseInfoForEstimate]:
    """House(s) информация для estimate.

    Логика (двойной поиск, union-deduplicate):
      1. Прямое совпадение по нормализованному адресу (tradein_normalize_short_addr).
      2. Geo-nearby — ST_DWithin 500м, любой source (avito/derived/etc.).

    Возвращаем прямой матч + nearby (dedup by id), up to ~6 домов.
    Пустой список если нет matches.
    """
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
    target = db.execute(
        text(
            """
            SELECT lat, lon, address FROM trade_in_estimates
            WHERE id = CAST(:id AS uuid)
            """
        ),
        {"id": str(estimate_id)},
    ).fetchone()
    if target is None:
        raise HTTPException(status_code=404, detail="estimate not found")

    # Path 1: прямой матч по нормализованному адресу
    direct: list[Any] = []
    if target.address:
        direct = list(
            db.execute(
                text(
                    f"""
                    SELECT DISTINCT {_HOUSE_SELECT_COLS},
                           0 AS distance_m
                    FROM houses h
                    WHERE h.short_address = tradein_normalize_short_addr(:addr)
                       OR tradein_normalize_short_addr(h.address)
                              = tradein_normalize_short_addr(:addr)
                    LIMIT 1
                    """
                ),
                {"addr": target.address},
            )
            .mappings()
            .all()
        )

    # Path 2: geo-nearby (any source, 500м radius)
    nearby: list[Any] = []
    if target.lat is not None and target.lon is not None:
        nearby = list(
            db.execute(
                text(
                    f"""
                    SELECT DISTINCT {_HOUSE_SELECT_COLS},
                           ST_Distance(
                               h.geom::geography,
                               ST_MakePoint(:lon, :lat)::geography
                           )::int AS distance_m
                    FROM houses h
                    WHERE h.geom IS NOT NULL
                      AND ST_DWithin(
                              h.geom::geography,
                              ST_MakePoint(:lon, :lat)::geography,
                              500
                          )
                    ORDER BY distance_m
                    LIMIT 5
                    """
                ),
                {"lat": target.lat, "lon": target.lon},
            )
            .mappings()
            .all()
        )

    # Merge: direct первым, затем nearby (dedup by house_id)
    seen_ids = {r["house_id"] for r in direct}
    merged = list(direct) + [r for r in nearby if r["house_id"] not in seen_ids]

    return [
        HouseInfoForEstimate(**{k: v for k, v in row.items() if k != "distance_m"})
        for row in merged
    ]


@router.get("/estimate/{estimate_id}/placement-history", response_model=list[PlacementHistoryEntry])
def get_estimate_placement_history(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[PlacementHistoryEntry]:
    """Историческая продажная активность по дому(ам) target estimate.

    Возвращает rows из house_placement_history для всех houses связанных с
    target адресом. Сортировано по last_price_date DESC.
    """
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
    target = db.execute(
        text(
            """
            SELECT lat, lon, address FROM trade_in_estimates
            WHERE id = CAST(:id AS uuid)
            """
        ),
        {"id": str(estimate_id)},
    ).fetchone()
    if target is None:
        raise HTTPException(status_code=404, detail="estimate not found")

    # Поиск house_ids по нормализованному адресу
    house_ids: list[int] = []
    if target.address:
        rows = db.execute(
            text(
                """
                SELECT id FROM houses
                WHERE short_address = tradein_normalize_short_addr(:addr)
                   OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)
                """
            ),
            {"addr": target.address},
        ).all()
        house_ids = [r.id for r in rows]

    if not house_ids and target.lat is not None and target.lon is not None:
        # Geo fallback: 100м (tight radius чтобы не смешать соседние дома)
        rows = db.execute(
            text(
                """
                SELECT id FROM houses
                WHERE geom IS NOT NULL
                  AND ST_DWithin(
                          geom::geography,
                          ST_MakePoint(:lon, :lat)::geography,
                          100
                      )
                LIMIT 3
                """
            ),
            {"lat": target.lat, "lon": target.lon},
        ).all()
        house_ids = [r.id for r in rows]

    if not house_ids:
        return []

    history = (
        db.execute(
            text(
                """
            SELECT id, source, house_id, ext_item_id, title, rooms, area_m2,
                   floor, total_floors, start_price, start_price_date,
                   last_price, last_price_date, removed_date, exposure_days,
                   notes
            FROM house_placement_history
            WHERE house_id = ANY(:house_ids)
            ORDER BY COALESCE(last_price_date, start_price_date) DESC NULLS LAST
            LIMIT 50
            """
            ),
            {"house_ids": house_ids},
        )
        .mappings()
        .all()
    )

    return [PlacementHistoryEntry(**dict(r)) for r in history]


@router.get("/estimate/{estimate_id}/house-analytics", response_model=HouseAnalyticsResponse)
def get_estimate_house_analytics(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
    radius_m: int | None = None,
) -> HouseAnalyticsResponse:
    """House-level analytics from house_placement_history backfill.

    Resolves target house(s) — если в самом доме <8 hist rows — расширяем поиск до 300м.
    Возвращает: price-history by year (median ₽/м²), recent sold (12mo), KPI.

    #2044 (BE-2): optional query-param radius_m (100–5000) явно задаёт радиус
    расширения выборки домов. None → текущая авто-логика (расширяем до 300 м
    только если в самом доме <8 записей). radius_m возвращается в ответе.
    """
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
    target = db.execute(
        text("SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
        {"id": str(estimate_id)},
    ).fetchone()
    if target is None:
        raise HTTPException(status_code=404, detail="estimate not found")

    # 1. Resolve target house_ids (same as placement-history endpoint)
    house_ids: list[int] = []
    if target.address:
        rows = db.execute(
            text(
                "SELECT id FROM houses WHERE short_address = tradein_normalize_short_addr(:addr) "
                "OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)"
            ),
            {"addr": target.address},
        ).all()
        house_ids = [r.id for r in rows]
    if not house_ids and target.lat is not None and target.lon is not None:
        rows = db.execute(
            text(
                "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
                "geom::geography, ST_MakePoint(:lon, :lat)::geography, 100) LIMIT 3"
            ),
            {"lat": target.lat, "lon": target.lon},
        ).all()
        house_ids = [r.id for r in rows]

    # 2. Expand search radius. #2044 (BE-2): explicit radius_m query-param
    #    overrides the auto 0→300 heuristic. None → byte-identical current
    #    behaviour (расширяем до 300 м только при <8 in-house rows).
    radius_used = 0
    if radius_m is not None:
        expand_radius = max(100, min(radius_m, 5000))
        if target.lat is not None and target.lon is not None:
            rows = db.execute(
                text(
                    "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
                    "geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) LIMIT 30"
                ),
                {"lat": target.lat, "lon": target.lon, "radius": expand_radius},
            ).all()
            house_ids = sorted(set(house_ids) | {r.id for r in rows})
        radius_used = expand_radius
    else:
        n_in_house = 0
        if house_ids:
            n_in_house = (
                db.execute(
                    text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"),
                    {"ids": house_ids},
                ).scalar()
                or 0
            )
        if n_in_house < 8 and target.lat is not None and target.lon is not None:
            rows = db.execute(
                text(
                    "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
                    "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30"
                ),
                {"lat": target.lat, "lon": target.lon},
            ).all()
            house_ids = sorted(set(house_ids) | {r.id for r in rows})
            radius_used = 300

    if not house_ids:
        return HouseAnalyticsResponse(
            house_ids=[],
            radius_m=0,
            price_history=[],
            recent_sold=[],
            kpi=HouseAnalyticsKpi(
                total_lots=0,
                sold_count=0,
                sold_rate_pct=0.0,
                median_exposure_days=None,
                median_bargain_pct=None,
            ),
        )

    # 3. Price history by year × source (median ₽/м²)
    price_history_rows = (
        db.execute(
            text(
                """
            SELECT
                EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year,
                source,
                COUNT(*) AS n_lots,
                percentile_cont(0.5) WITHIN GROUP (ORDER BY last_price / NULLIF(area_m2, 0))::int
                    AS median_price_per_m2,
                percentile_cont(0.5) WITHIN GROUP (ORDER BY last_price)::int AS median_price_rub
            FROM house_placement_history
            WHERE house_id = ANY(:ids)
              AND last_price IS NOT NULL AND last_price > 100000
              AND area_m2 IS NOT NULL AND area_m2 > 10
              AND COALESCE(last_price_date, start_price_date) IS NOT NULL
            GROUP BY year, source
            ORDER BY year ASC, source ASC
            """
            ),
            {"ids": house_ids},
        )
        .mappings()
        .all()
    )

    # 4. Recent sold (12 months, with removed_date)
    recent_sold_rows = (
        db.execute(
            text(
                """
            SELECT id, source, rooms, area_m2, floor, start_price, last_price,
                   removed_date, exposure_days,
                   CASE WHEN start_price > 0 AND last_price IS NOT NULL
                        THEN ROUND((start_price - last_price)::numeric / start_price * 100, 1)
                        ELSE NULL END AS discount_pct
            FROM house_placement_history
            WHERE house_id = ANY(:ids)
              AND removed_date IS NOT NULL
              AND removed_date > (NOW() - INTERVAL '12 months')::date
            ORDER BY removed_date DESC
            LIMIT 20
            """
            ),
            {"ids": house_ids},
        )
        .mappings()
        .all()
    )

    # 5. KPI aggregate
    kpi_row = (
        db.execute(
            text(
                """
            SELECT
                COUNT(*) AS total_lots,
                COUNT(*) FILTER (WHERE removed_date IS NOT NULL) AS sold_count,
                CASE WHEN COUNT(*) > 0
                     THEN ROUND(
                              COUNT(*) FILTER (WHERE removed_date IS NOT NULL)::numeric
                              / COUNT(*) * 100,
                              1
                          )
                     ELSE 0 END AS sold_rate_pct,
                percentile_cont(0.5) WITHIN GROUP (ORDER BY exposure_days)
                    FILTER (WHERE exposure_days IS NOT NULL) AS median_exposure_days,
                ROUND(
                    percentile_cont(0.5) WITHIN GROUP (
                        ORDER BY (start_price - last_price)::numeric / NULLIF(start_price, 0) * 100
                    ) FILTER (
                        WHERE start_price > 0
                          AND last_price IS NOT NULL
                          AND last_price != start_price
                    )::numeric,
                    1
                ) AS median_bargain_pct
            FROM house_placement_history
            WHERE house_id = ANY(:ids)
            """
            ),
            {"ids": house_ids},
        )
        .mappings()
        .first()
    )

    return HouseAnalyticsResponse(
        house_ids=house_ids,
        radius_m=radius_used,
        price_history=[PriceHistoryYearPoint(**dict(r)) for r in price_history_rows],
        recent_sold=[RecentSoldEntry(**dict(r)) for r in recent_sold_rows],
        kpi=HouseAnalyticsKpi(
            total_lots=kpi_row["total_lots"] or 0 if kpi_row else 0,
            sold_count=kpi_row["sold_count"] or 0 if kpi_row else 0,
            sold_rate_pct=float(kpi_row["sold_rate_pct"] or 0) if kpi_row else 0.0,
            median_exposure_days=(
                int(kpi_row["median_exposure_days"])
                if kpi_row and kpi_row["median_exposure_days"] is not None
                else None
            ),
            median_bargain_pct=(
                float(kpi_row["median_bargain_pct"])
                if kpi_row and kpi_row["median_bargain_pct"] is not None
                else None
            ),
        ),
    )


@router.get(
    "/estimate/{estimate_id}/cian-price-changes",
    response_model=list[CianPriceChangeStats],
)
def get_estimate_cian_price_changes(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> list[CianPriceChangeStats]:
    """История изменений цены для Cian-аналогов из estimate.

    Для каждого cian-аналога в estimate.analogs:
      - extract cian_id из URL (/sale/flat/<id>/)
      - JOIN listings (source='cian', source_id IN cian_ids)
      - JOIN offer_price_history per listing_id
      - Aggregate: n_changes, last_change_time, last_diff_percent, total_change_pct

    Возвращает только аналоги с хотя бы одним изменением цены.
    Пустой список если нет cian-аналогов или нет истории.
    """
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
    rows = (
        db.execute(
            text(
                """
            WITH cian_analogs AS (
                SELECT DISTINCT
                    substring(a->>'source_url' from '/sale/flat/(\\d+)/') AS cian_id
                FROM trade_in_estimates, jsonb_array_elements(analogs) a
                WHERE id = CAST(:eid AS uuid)
                  AND a->>'source' = 'cian'
                  AND a->>'source_url' IS NOT NULL
                  AND substring(a->>'source_url' from '/sale/flat/(\\d+)/') IS NOT NULL
            ),
            listings_resolved AS (
                SELECT l.id AS listing_id, l.source_id AS cian_id,
                       l.price_rub::int AS current_price
                FROM listings l
                JOIN cian_analogs c ON l.source_id = c.cian_id
                WHERE l.source = 'cian'
            ),
            changes_agg AS (
                SELECT
                    oph.listing_id,
                    COUNT(*) AS n_changes,
                    MAX(oph.change_time) AS last_change_time,
                    (array_agg(oph.diff_percent ORDER BY oph.change_time DESC))[1]
                        AS last_diff_percent,
                    (
                        SELECT oph2.price_rub::int
                        FROM offer_price_history oph2
                        WHERE oph2.listing_id = oph.listing_id
                        ORDER BY oph2.change_time ASC
                        LIMIT 1
                    ) AS first_seen_price
                FROM offer_price_history oph
                WHERE oph.listing_id IN (SELECT listing_id FROM listings_resolved)
                GROUP BY oph.listing_id
            )
            SELECT
                lr.cian_id,
                lr.listing_id,
                ca.n_changes::int,
                ca.last_change_time,
                ca.last_diff_percent::float AS last_diff_percent,
                ca.first_seen_price,
                lr.current_price,
                CASE
                    WHEN ca.first_seen_price IS NOT NULL AND ca.first_seen_price > 0
                    THEN ROUND(
                        (lr.current_price - ca.first_seen_price)::numeric
                        / ca.first_seen_price * 100,
                        1
                    )::float
                    ELSE NULL
                END AS total_change_pct
            FROM listings_resolved lr
            JOIN changes_agg ca ON ca.listing_id = lr.listing_id
            WHERE ca.n_changes > 0
            ORDER BY ca.last_change_time DESC
            """
            ),
            {"eid": str(estimate_id)},
        )
        .mappings()
        .all()
    )

    return [CianPriceChangeStats(**dict(r)) for r in rows]


@router.get(
    "/estimate/{estimate_id}/sell-time-sensitivity",
    response_model=SellTimeSensitivityResponse,
)
def get_estimate_sell_time_sensitivity(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
    radius_m: int | None = None,
) -> SellTimeSensitivityResponse:
    """Срок продажи в зависимости от цены к медиане дома/района.

    4 бакета: -5% / медиана (±3%) / +5% / +10%. Median exposure_days + p25/p75.
    Filter last_price > start_price * 0.7 — отбрасываем подозрительно
    заниженные лоты (выбросы, ошибки парсинга).

    #2044 (BE-2): optional query-param radius_m (100–5000) явно задаёт радиус
    расширения выборки домов. None → текущая авто-логика (300 м при <8 записях).
    """
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
    # 1. Resolve house_ids (same logic as house-analytics)
    target = db.execute(
        text("SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
        {"id": str(estimate_id)},
    ).fetchone()
    if target is None:
        raise HTTPException(status_code=404, detail="estimate not found")

    house_ids: list[int] = []
    if target.address:
        rows = db.execute(
            text(
                "SELECT id FROM houses WHERE short_address = tradein_normalize_short_addr(:addr) "
                "OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)"
            ),
            {"addr": target.address},
        ).all()
        house_ids = [r.id for r in rows]
    if not house_ids and target.lat is not None and target.lon is not None:
        rows = db.execute(
            text(
                "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
                "geom::geography, ST_MakePoint(:lon, :lat)::geography, 100) LIMIT 3"
            ),
            {"lat": target.lat, "lon": target.lon},
        ).all()
        house_ids = [r.id for r in rows]

    # Expand search radius (same threshold as house-analytics). #2044 (BE-2):
    # explicit radius_m overrides the auto 0→300 heuristic; None → byte-identical.
    radius_used = 0
    if radius_m is not None:
        expand_radius = max(100, min(radius_m, 5000))
        if target.lat is not None and target.lon is not None:
            rows = db.execute(
                text(
                    "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
                    "geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) LIMIT 30"
                ),
                {"lat": target.lat, "lon": target.lon, "radius": expand_radius},
            ).all()
            house_ids = sorted(set(house_ids) | {r.id for r in rows})
        radius_used = expand_radius
    else:
        n_in_house = 0
        if house_ids:
            n_in_house = (
                db.execute(
                    text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"),
                    {"ids": house_ids},
                ).scalar()
                or 0
            )
        if n_in_house < 8 and target.lat is not None and target.lon is not None:
            rows = db.execute(
                text(
                    "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
                    "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30"
                ),
                {"lat": target.lat, "lon": target.lon},
            ).all()
            house_ids = sorted(set(house_ids) | {r.id for r in rows})
            radius_used = 300

    if not house_ids:
        return SellTimeSensitivityResponse(
            house_ids=[],
            radius_m=0,
            target_median_price_per_m2=None,
            buckets=[],
        )

    # 2. Compute benchmark median ₽/м² for last 2 years
    target_median = db.execute(
        text(
            """
            SELECT percentile_cont(0.5) WITHIN GROUP (
                ORDER BY last_price / NULLIF(area_m2, 0)
            )::int AS median_ppm2
            FROM house_placement_history
            WHERE house_id = ANY(:ids)
              AND last_price IS NOT NULL AND last_price > 100000
              AND area_m2 > 10
              AND COALESCE(last_price_date, start_price_date) > (NOW() - INTERVAL '2 years')::date
              AND (start_price = 0 OR last_price > start_price * 0.7)
            """
        ),
        {"ids": house_ids},
    ).scalar()

    # 3. Per-year median (для расчёта premium per lot); используем CTE для bucket-расчёта
    bucket_rows = (
        db.execute(
            text(
                """
            WITH year_medians AS (
                SELECT
                    EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year,
                    percentile_cont(0.5) WITHIN GROUP (
                        ORDER BY last_price / NULLIF(area_m2, 0)
                    ) AS median_ppm2
                FROM house_placement_history
                WHERE house_id = ANY(:ids)
                  AND last_price IS NOT NULL AND area_m2 > 10
                  AND (start_price = 0 OR last_price > start_price * 0.7)
                GROUP BY year
            ),
            lots_with_premium AS (
                SELECT
                    hph.exposure_days,
                    CASE
                        WHEN ym.median_ppm2 IS NULL OR ym.median_ppm2 = 0 THEN NULL
                        ELSE ((hph.last_price / NULLIF(hph.area_m2, 0)) - ym.median_ppm2)
                             / ym.median_ppm2 * 100
                    END AS premium_pct
                FROM house_placement_history hph
                JOIN year_medians ym ON ym.year = EXTRACT(YEAR FROM
                    COALESCE(hph.last_price_date, hph.start_price_date))::int
                WHERE hph.house_id = ANY(:ids)
                  AND hph.removed_date IS NOT NULL
                  AND hph.exposure_days IS NOT NULL
                  AND hph.area_m2 > 10
                  AND (hph.start_price = 0 OR hph.last_price > hph.start_price * 0.7)
            ),
            bucketed AS (
                SELECT
                    CASE
                        WHEN premium_pct BETWEEN -10 AND -3 THEN 'cheap'
                        WHEN premium_pct BETWEEN -3 AND 3 THEN 'median'
                        WHEN premium_pct BETWEEN 3 AND 8 THEN 'plus5'
                        WHEN premium_pct BETWEEN 8 AND 15 THEN 'plus10'
                        ELSE NULL
                    END AS bucket,
                    exposure_days
                FROM lots_with_premium
                WHERE premium_pct IS NOT NULL
            )
            SELECT
                bucket,
                COUNT(*) AS n_lots,
                percentile_cont(0.5) WITHIN GROUP (ORDER BY exposure_days)::int
                    AS median_exposure_days,
                percentile_cont(0.25) WITHIN GROUP (ORDER BY exposure_days)::int AS p25_days,
                percentile_cont(0.75) WITHIN GROUP (ORDER BY exposure_days)::int AS p75_days
            FROM bucketed
            WHERE bucket IS NOT NULL
            GROUP BY bucket
            """
            ),
            {"ids": house_ids},
        )
        .mappings()
        .all()
    )

    # 4. Build buckets — гарантируем все 4 даже если данных нет в bucket
    bucket_map = {r["bucket"]: dict(r) for r in bucket_rows}
    bucket_definitions = [
        ("cheap", -5.0),
        ("median", 0.0),
        ("plus5", 5.0),
        ("plus10", 10.0),
    ]
    buckets: list[SellTimeBucket] = []
    for label, pct in bucket_definitions:
        r = bucket_map.get(label)
        n_lots = r["n_lots"] if r else 0
        buckets.append(
            SellTimeBucket(
                price_premium_label=label,
                price_premium_pct=pct,
                median_exposure_days=r["median_exposure_days"] if r else None,
                p25_days=r["p25_days"] if r else None,
                p75_days=r["p75_days"] if r else None,
                n_lots=n_lots,
                # #1995: малая выборка → median/p25/p75 шумные (наблюдалась
                # немонотонность между бакетами, напр. +10% быстрее +5%). Честный
                # флаг вместо тихого шума — фронт решает, как показать.
                insufficient_data=n_lots < settings.sell_time_sensitivity_min_n_lots,
            )
        )

    return SellTimeSensitivityResponse(
        house_ids=house_ids,
        radius_m=radius_used,
        target_median_price_per_m2=int(target_median) if target_median else None,
        buckets=buckets,
    )


@router.get("/estimate/{estimate_id}/imv-benchmark", response_model=IMVBenchmarkResponse)
def get_estimate_imv_benchmark(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> IMVBenchmarkResponse:
    """Avito IMV benchmark для estimate (для UI badge «наша 6.4М · Avito 6.29М»).

    Источники lookup:
      1. avito_imv_evaluations WHERE estimate_id = :id (если linked в estimator)
      2. Если не linked — fallback: most recent IMV для same address (TTL 24h)
    """
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
    # Сначала пытаемся найти directly linked
    row = db.execute(
        text(
            """
            SELECT cache_key, recommended_price, lower_price, higher_price,
                   market_count, fetched_at
            FROM avito_imv_evaluations
            WHERE estimate_id = CAST(:id AS uuid)
            ORDER BY fetched_at DESC
            LIMIT 1
            """
        ),
        {"id": str(estimate_id)},
    ).fetchone()

    if row is None:
        # Fallback: same address за 24h (на случай если link не успел)
        est = db.execute(
            text(
                """
                SELECT address FROM trade_in_estimates
                WHERE id = CAST(:id AS uuid)
                """
            ),
            {"id": str(estimate_id)},
        ).fetchone()
        if est is None:
            raise HTTPException(status_code=404, detail="estimate not found")
        if est.address:
            row = db.execute(
                text(
                    """
                    SELECT cache_key, recommended_price, lower_price, higher_price,
                           market_count, fetched_at
                    FROM avito_imv_evaluations
                    WHERE address = :address
                      AND fetched_at > NOW() - INTERVAL '24 hours'
                    ORDER BY fetched_at DESC
                    LIMIT 1
                    """
                ),
                {"address": est.address},
            ).fetchone()

    if row is None:
        return IMVBenchmarkResponse(available=False)

    # Get our_median_price для compare
    our = db.execute(
        text(
            """
            SELECT median_price FROM trade_in_estimates
            WHERE id = CAST(:id AS uuid)
            """
        ),
        {"id": str(estimate_id)},
    ).fetchone()
    our_median = our.median_price if our else None
    diff_pct = None
    if our_median and row.recommended_price:
        diff_pct = round((our_median - row.recommended_price) / row.recommended_price * 100, 1)

    return IMVBenchmarkResponse(
        available=True,
        cache_key=row.cache_key,
        recommended_price=row.recommended_price,
        lower_price=row.lower_price,
        higher_price=row.higher_price,
        market_count=row.market_count,
        fetched_at=row.fetched_at,
        our_median_price=our_median,
        diff_pct=diff_pct,
    )


# ── Location index (issue TBD, замена сломанного location-coef #2045) ────────


@router.get("/location-index", response_model=LocationIndexResponse)
def get_location_index(
    estimate_id: UUID,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
    radius_m: int | None = None,
) -> LocationIndexResponse:
    """Location index для оценки (замена сломанного location-coef, LocationDrawer).

    Резолвит lat/lon оценки, считает индекс через
    app.services.location_index.compute_location_index: % отклонения медианы ₽/м²
    сопоставимых активных листингов в радиусе точки от медианы ₽/м² по всему Екатеринбургу
    (percentile_cont(0.5) — устойчиво к выбросам). НЕ участвует в цене — estimator.py про
    этот показатель не знает (аналоги уже несут локацию в базовой цене).

    404 — оценки нет / IDOR (тот же _assert_estimate_access_by_id, что и у соседних
    derived-роутов). radius_m опционален (None → адаптивная лестница радиусов
    RADIUS_LADDER_M, расширяется пока выборка не наберёт MIN_SAMPLE_SIZE); явное значение
    клэмпится в [500, 3000] и используется РОВНО как задано (без расширения).

    Честная деградация (НЕ 500, НЕ сфабрикованные значения) — см. LocationIndexResponse:
      - status="out_of_coverage" — у оценки нет lat/lon, ИЛИ точка вне гео-охвата продукта
        (Екатеринбург).
      - status="insufficient_data" — даже на максимальном радиусе сопоставимых активных
        листингов меньше порога.
      - poi_status="unavailable" — osm_poi_ekb_local пуста/не отрефрешена (независимо от
        status выше — «что рядом» и числовой индекс деградируют раздельно).
    """
    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)

    row = db.execute(
        text(
            """
            SELECT lat, lon
            FROM trade_in_estimates
            WHERE id = CAST(:id AS uuid)
            """
        ),
        {"id": str(estimate_id)},
    ).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="estimate not found")

    if row.lat is None or row.lon is None:
        logger.info(
            "location_index: estimate=%s has no lat/lon — out_of_coverage fallback", estimate_id
        )
        return LocationIndexResponse(
            status="out_of_coverage",
            location_index_pct=None,
            local_median_price_per_m2=None,
            city_median_price_per_m2=None,
            sample_size=0,
            radius_m=radius_m or 0,
            nearby_poi=[],
            poi_status="unavailable",
        )

    from app.services.location_index import compute_location_index

    resolved_radius = None if radius_m is None else max(500, min(radius_m, 3000))
    result = compute_location_index(db, float(row.lat), float(row.lon), radius_m=resolved_radius)

    return LocationIndexResponse(
        status=result.status,
        location_index_pct=result.location_index_pct,
        local_median_price_per_m2=result.local_median_price_per_m2,
        city_median_price_per_m2=result.city_median_price_per_m2,
        sample_size=result.sample_size,
        radius_m=result.radius_m,
        nearby_poi=[
            NearbyPoiOut(poi_type=p.poi_type, name=p.name, distance_m=p.distance_m)
            for p in result.nearby_poi
        ],
        poi_status=result.poi_status,
    )


# ── Street-level deals (rosreestr open dataset) ───────────────────────────────


@router.get("/street-deals", response_model=StreetDealsResponse)
def get_street_deals(
    address: str,
    area_m2: float,
    rooms: int,
    db: Annotated[Session, Depends(get_db)] = None,  # type: ignore[assignment]
    period_months: int = 12,
    area_tolerance: float = 0.15,
) -> StreetDealsResponse:
    """ДКП-сделки Росреестра по улице целевого адреса.

    Open dataset Росреестра агрегирует адреса до улицы (без номера дома).
    Поэтому это per-street view, не per-house. До квартир-аналогов выборку
    сужает полоса площади ±area_tolerance; комнатность клиента в фильтр НЕ
    входит — `deals.rooms` не комнатность, а синтетика из той же площади
    (#3256, см. estimator._fetch_dkp_corridor). `rooms` остаётся параметром
    ручки: он описывает запрос и попадает в лог, но не в WHERE.

    После PR-A (#549) таблица deals содержит только ДКП (ДДУ-первичка отфильтрована
    в import-rosreestr.sh).
    """
    from app.services.estimator import (
        _deal_to_analog,
        _percentile,
        _resolve_target_city,
        extract_street_name,
    )

    now = datetime.now(tz=UTC)
    # #1381: отображаемое окно должно совпадать с SQL-фильтром ниже, который
    # использует календарный interval PostgreSQL (NOW() - N months), а не
    # фиксированные 30-дневные месяцы. Считаем cutoff календарной арифметикой.
    _from_month0 = (now.year * 12 + (now.month - 1)) - period_months
    _from_year, _from_month_idx = divmod(_from_month0, 12)
    _from_month = _from_month_idx + 1
    # Клампим день для коротких месяцев (как делает PostgreSQL interval).
    _from_day = min(now.day, calendar.monthrange(_from_year, _from_month)[1])
    period_from: date = date(_from_year, _from_month, _from_day)
    period_to: date = now.date()

    def _empty() -> StreetDealsResponse:
        return StreetDealsResponse(
            street=None,
            period_from=period_from,
            period_to=period_to,
            count=0,
            median_price_rub=0,
            median_price_per_m2=0,
            range_low_rub=0,
            range_high_rub=0,
            deals=[],
        )

    street_name = extract_street_name(address)
    if not street_name:
        logger.warning("street-deals: could not extract street from address=%r", address)
        return _empty()

    area_min = area_m2 * (1.0 - area_tolerance)
    area_max = area_m2 * (1.0 + area_tolerance)

    # #C1 city-scope (п.3, консистентно с estimator._fetch_dkp_corridor): резолвим
    # город целевого адреса через _resolve_target_city (словарь ~30 городов обл.66
    # вкл. ЕКБ + sweep-города) и фильтруем сделки по этому городу — одноимённые улицы
    # др. городов не контаминируют витрину. None (адрес вне словаря) → фильтр не
    # применяется. `city_filter` — литерал (не user-input), значение идёт bind-параметром.
    target_city = _resolve_target_city(address)
    city_filter = "AND LOWER(city) = CAST(:target_city AS text)" if target_city else ""

    rows = (
        db.execute(
            text(
                f"""
            SELECT address, area_m2, rooms, floor, total_floors,
                   price_rub, price_per_m2, deal_date, source
            FROM deals
            WHERE source = 'rosreestr'
              AND address ILIKE :street_pattern
              AND address ~* :street_regex
              {city_filter}
              -- #3256: фильтра по rooms нет — deals.rooms синтезирована из площади
              -- (тот же CASE 30/44/62/85, что area_bucket), т.е. это был второй
              -- ступенчатый фильтр по площади поверх полосы ±15% ниже. Развёрнуто —
              -- в комментарии estimator._fetch_dkp_corridor.
              AND area_m2 BETWEEN :area_min AND :area_max
              AND deal_date > NOW() - (CAST(:period_months AS integer) || ' months')::interval
              AND price_rub > 0
            ORDER BY deal_date DESC
            """
            ),
            {
                "street_pattern": "%" + street_name + "%",
                "street_regex": r"\m" + street_name + r"\M",
                "target_city": target_city.lower() if target_city else None,
                "area_min": area_min,
                "area_max": area_max,
                "period_months": period_months,
            },
        )
        .mappings()
        .all()
    )

    if not rows:
        # #3256: лог называет ТОТ ключ, которым искали. Комнатность клиента в
        # выборку не входит (deals.rooms — синтетика из площади), поэтому она
        # печатается как контекст запроса, а не как параметр фильтра.
        logger.info(
            "street-deals: no rows found street=%r area=%.1f±%.0f%% "
            "(ключ по комнатам не применяется, #3256; комнатность клиента=%d)",
            street_name,
            area_m2,
            area_tolerance * 100,
            rooms,
        )
        return StreetDealsResponse(
            street=street_name,
            period_from=period_from,
            period_to=period_to,
            count=0,
            median_price_rub=0,
            median_price_per_m2=0,
            range_low_rub=0,
            range_high_rub=0,
            deals=[],
        )

    count = len(rows)
    prices_rub = sorted(float(r["price_rub"]) for r in rows)
    prices_ppm2 = sorted(float(r["price_per_m2"]) for r in rows if r["price_per_m2"])

    median_ppm2 = _percentile(prices_ppm2, 0.5) if prices_ppm2 else 0.0
    median_price_rub = (
        int(median_ppm2 * area_m2) if median_ppm2 else int(_percentile(prices_rub, 0.5))
    )
    range_low_rub = int(prices_rub[0])
    range_high_rub = int(prices_rub[-1])

    top10 = [_deal_to_analog(dict(r)) for r in rows[:10]]

    logger.info(
        "street-deals: street=%r area=%.1f±%.0f%% count=%d median_ppm2=%.0f "
        "(ключ по комнатам не применяется, #3256; комнатность клиента=%d)",
        street_name,
        area_m2,
        area_tolerance * 100,
        count,
        median_ppm2,
        rooms,
    )

    return StreetDealsResponse(
        street=street_name,
        period_from=period_from,
        period_to=period_to,
        count=count,
        median_price_rub=median_price_rub,
        median_price_per_m2=int(median_ppm2),
        range_low_rub=range_low_rub,
        range_high_rub=range_high_rub,
        deals=top10,
    )


# ── Sales vs Listings (PR K — Foundation Phase 1 of issue #564) ──────────────

# #2666 гейт правдоподобия на «медианный торг». Пейринг ДКП↔объявление идёт по
# УЛИЦЕ без номера дома (data_quality="street_only", ADR #721): на длинной улице
# сделка и объявление могут стоять в разных домах и разных ценовых классах, и
# тогда discount_pct — не торг, а разница между двумя чужими друг другу лотами.
# Гард #2660 (миграция 211) убрал предвзятые пары «вторичка ↔ новостройка» и тем
# самым сделал остаток артефактов ВИДНЫМ: по `%Космонавтов%` 2-комн. медиана
# уехала с −11.9% на +36.4%, т.е. пользователю написали бы «продали на 36%
# дороже, чем просили». Здесь не чиним пейринг (это ADR-уровень), а перестаём
# показывать число, которому нельзя верить.
#
# Пороги подобраны по проду 2026-08-05 (симуляция эндпоинта на 238 РЕАЛЬНЫХ
# пользовательских запросах из trade_in_estimates — тот же address/area/rooms,
# что уходил в виджет; 128 из них дали хотя бы одну пару):
#
# MIN_PAIRS = 10 — бутстрап по 12 «плотным» группам (n ≥ 60 пар): из полной
#   выборки берём подвыборку размера k и смотрим, насколько медиана подвыборки
#   отклоняется от полной. p90 |отклонения|: k=5 → 18.8 п.п., k=10 → 12.0,
#   k=15 → 9.9, k=20 → 8.2. Кривая ломается ровно на 10 (5→10 даёт −6.8 п.п.
#   шума, 10→15 уже только −2.1, а каждые +5 к порогу стоят ещё ~8-10% улиц).
#   Совпадает с уже принятым в продукте порогом малой выборки
#   settings.sell_time_sensitivity_min_n_lots = 10.
#
# SANE_MIN/MAX = [−60%, +20%] — асимметричны намеренно, у сторон разная природа:
#   ВЕРХ. В наблюдаемом распределении 128 групп положительный хвост РАЗОРВАН:
#   +11.1, +10.8, +16.9 — и дальше пусто до +33.7, +34.2, +34.6, +39.0, +52.5,
#   +70.2, +81.5, +103.1. Отсечка +20% попадает в пустой промежуток, т.е. режет
#   отдельный кластер, а не край континуума. Сверху её подпирает рынок: ни один
#   городской бакет asking_to_sold_ratios не даёт плюса вообще (max ratio 0.9132
#   = −8.7% торга), так что «продали на +20% дороже ask» уже вдвое дальше любого
#   рыночно объяснимого плюса.
#   НИЗ. Разрыва нет — минус идёт сплошняком от −5% до −87%, и это ожидаемо:
#   у большого отрицательного торга есть механизм (занижение цены в ДКП), в
#   отличие от большого плюса. Поэтому граница грубая, «заведомо не рынок»:
#   худший городской бакет (студии, ratio 0.7623) = −23.8%, −60% в 2.5 раза
#   глубже. Режет 6 групп из 128 (−87 … −64).
#
# Цена гейта на проде: из 128 групп с парами число сохраняют 64 (50%), 59 (46%)
# теряют его по «мало пар» и ещё 5 (4%) — по диапазону. Виджет при этом остаётся:
# сделки, медиана ₽/м², диапазон и сами пары считаются мимо гейта, гаснет ровно
# строка «медианный торг», и вместо неё уходит median_discount_explanation.
#
# MIN_DISTINCT_LISTINGS = 2 (#2672) — ПАРЫ НЕ ЯВЛЯЮТСЯ НЕЗАВИСИМЫМИ НАБЛЮДЕНИЯМИ,
#   и MIN_PAIRS этого не видит. DISTINCT ON подбирает по объявлению на сделку, но
#   ОДНО объявление переиспользуется на многих сделках улицы: у показываемых групп
#   медиана — 18 сделок на одно различное объявление. До этого порога из 64
#   показываемых чисел 22 (34%) стояли на ОДНОМ объявлении (худший живой кейс —
#   `Белинского` 1-комн.: 50 пар, 1 объявление, −50.6%), 50 (78%) — меньше чем на
#   трёх. «50 пар» там означало не 50 наблюдений рынка, а 50 сделок, поделённых на
#   ОДНУ цену предложения: число говорило о том, чем эта конкретная квартира
#   отличалась от типичной сделки, а не о торге на улице.
#
#   Почему именно 2, и почему порог здесь обоснован ИНАЧЕ, чем MIN_PAIRS. Разброс
#   со стороны объявлений мерили джекнайфом (выкинуть одно объявление, 45 групп,
#   118 повторов): p50 3.6, p90 18.8, max 80.4 п.п. — тот же порядок, что и шум
#   при 5 парах, который при выборе MIN_PAIRS сочли неприемлемым. Но на группах с
#   ОДНИМ объявлением ни джекнайф, ни кластерный бутстрап не дают числа вообще:
#   выкидывать нечего, пересэмплировать нечего, отклонение тождественно 0.
#   Их «нулевая ошибка» — не малая ошибка, а отсутствие измерения, и агрегат по
#   всем 64 группам от их добавления УЛУЧШАЛСЯ (кластер-бутстрап p90 16.0 → 11.5),
#   т.е. метрика становилась тем зеленее, чем больше в ней неизмеримого. Поэтому
#   2 — не статистический выбор, а граница выразимости: ниже неё нет выборки, о
#   разбросе которой можно спрашивать, и показывать число = фабриковать точность.
#   Выше 2 порог уже статистический, и данные (прод 2026-08-06, те же 128 групп)
#   говорят, что он должен быть выше — но ценой почти всей витрины:
#     объявлений ≥ 2 → 42 группы (33%), джекнайф p90 17.4;
#     объявлений ≥ 3 → 14 групп (11%), p90 10.9 (планка MIN_PAIRS — 12.0);
#     объявлений ≥ 4 →  7 групп ( 5%), p90  5.5.
#   Порог 3 попадал бы в принятую планку шума, но оставляет 11% витрины и всё
#   равно не делает число защищаемым (ошибка со стороны СДЕЛОК никуда не делась и
#   складывается с ней). Выбирать между «9% покрытия» и «выключить строку» —
#   решение владельца, не гейта; здесь снимается ровно то, что не является
#   наблюдением рынка в принципе. Понижать MIN_PAIRS в компенсацию нельзя:
#   вернувшиеся группы стоят на тех же одном-двух объявлениях (ложная точность).
#
# SANE_MIN ужесточён −60% → −35% (#2672). Исходное подозрение «−60% режет живой
#   рынок» проверено и ОПРОВЕРГНУТО: до −60% проходило всё, законный механизм
#   большого минуса (занижение цены в ДКП) сохранён целиком. Ошибка была в другую
#   сторону — граница пропускала неправдоподобный отрицательный хвост: 26 из 64
#   показываемых чисел (41%) лежали ниже −23.7%, худшего объяснимого рынком
#   бакета (asking_to_sold_ratios: студии, ratio 0.7634, 1 519 сделок; ни один
#   бакет не глубже), самое глубокое показываемое — −58.5%. Мы гасили «+34%» и
#   показывали «−58.5%», полученный из ТОГО ЖЕ артефакта пейринга. Асимметрия
#   работала против пользователя: абсурдный плюс сам себя опровергает («продали
#   дороже, чем просили» — виджету просто не поверят), абсурдный минус выглядит
#   правдоподобно и подталкивает продавца к выводу, что его улица торгуется за
#   полцены. −35% ≈ в 1.5 раза глубже худшего рыночного бакета (запас на занижение
#   в ДКП сохранён) и попадает в разрыв наблюдаемого распределения −37.6 → −33.9.
#   Живой кейс из ревью: Серов, Ленина 163, 2-комн., 21 пара → −46.5% показывался.
#
# ПОШТУЧНЫЙ discount_pct В СТРОКАХ ТАБЛИЦЫ (#2672 п.3). Гейт гасил сводное число,
#   а таблица под ним продолжала показывать проценты, посчитанные из ТЕХ ЖЕ пар:
#   на живом Космонавтове (2-комн., медиана −37.6% погашена) шесть из первых
#   двенадцати строк — от +42% до +77%, и все против одной и той же цены
#   предложения. Масштаб на проде 2026-08-07 (427 реальных запросов из
#   trade_in_estimates, 135 групп с парами): медиана погашена у 102 групп, и в
#   них видно 3 678 строк с процентом — 72.8% всех показываемых процентов.
#
#   Гасим строку там, и только там, где причина — свойство САМОЙ ПАРЫ:
#     а) объявлений < MIN_DISTINCT_LISTINGS — тогда столбец «разница» это
#        столбец цены сделки, поделённый на одну и ту же константу: он не даёт
#        ни одного наблюдения сверх уже показанных цен, но выглядит как N торгов;
#     б) медиана вне санитарного диапазона — по определению медианы это
#        утверждение О СТРОКАХ: половина из них ещё дальше от рынка, чем она.
#   «Мало пар» строку НЕ гасит: это свойство ВЫБОРКИ, про отдельную пару оно
#   ничего не говорит, а микрокопия «пар всего 4, поэтому процент в строке не
#   показываем» была бы ложной причиной. Цена этого исключения — 3 группы / 20
#   строк на проде, где медианы нет, а проценты в строках есть.
#   Флаги (а)/(б) считаются НЕЗАВИСИМО от порядка веток гейта: порядок «мало пар
#   → одно объявление → диапазон» прячет вторую причину за первой, и на проде 55
#   групп гаснут как «мало пар», хотя стоят ещё и на ОДНОМ объявлении. По ветке
#   гейта строки гасились бы не там, где надо.
#   Цена на проде: из 5 054 строк с процентом гаснет 3 658 (72.4%), остаётся
#   1 396. Само число «медианный торг» этой правкой НЕ меняется — 33 группы из
#   135 и до, и после (замер обеих версий модуля в одном процессе на ОДНИХ И ТЕХ
#   ЖЕ живых парах). Обе цены — сделки и объявления — в строке остаются:
#   убирается не данные, а наша подпись «торг» под их разностью.
#
# ПОТОЛОК ГЕЙТА (знать до следующей правки — здесь НЕ чинится):
#   1. Пейринг по УЛИЦЕ, а не по дому — корень всего перечисленного (ADR #721).
#      Гейт по различным объявлениям честный промежуточный шаг, а не решение:
#      он убирает числа, которые не являются наблюдением, но оставшиеся всё ещё
#      сравнивают сделку в одном доме с объявлением в другом.
#   2. В группах, ПРОШЕДШИХ гейт, поштучные проценты остаются как есть — включая
#      426 строк из 1 396 (31%), лежащих вне того же диапазона [−35%, +20%], по
#      которому мы гасим медиану. Отдельного порога для ОДНОЙ пары у нас нет:
#      диапазон калиброван на медианах групп, а у одной сделки законный разброс
#      шире (занижение цены в ДКП — механизм поштучный, не медианный). Считать
#      его = вводить некалиброванный порог, чего #2672 прямо предостерегает.
SALES_VS_LISTINGS_MIN_PAIRS = 10
SALES_VS_LISTINGS_MIN_DISTINCT_LISTINGS = 2
SALES_VS_LISTINGS_SANE_DISCOUNT_MIN_PCT = -35.0
SALES_VS_LISTINGS_SANE_DISCOUNT_MAX_PCT = 20.0


@router.get("/sales-vs-listings", response_model=SalesVsListingsResponse)
def get_sales_vs_listings(
    address: str,
    area_m2: float,
    rooms: int,
    db: Annotated[Session, Depends(get_db)] = None,  # type: ignore[assignment]
    window_days: int = 180,
    area_tolerance: float = 0.15,
    period_months: int = 24,
) -> SalesVsListingsResponse:
    """Pairs (ДКП-сделка, listing) для улицы целевого адреса (PR K / #564).

    Для каждой ДКП-сделки Росреестра в окне `period_months` пытаемся найти
    matching listing на той же улице с такими же rooms / близкой area_m2 /
    listing_date в окне [deal_date - window_days, deal_date + 30d grace].

    Возвращаем LEFT JOIN: сделки без listing match сохраняются (listing_* = None),
    чтобы вычислить linkage_rate.

    discount_pct = (deal_price - listing_price) / listing_price * 100.
    Отрицательный = продали дешевле asking → reasoned discount от торга.

    Per-street view: Росреестр open dataset агрегирует адреса до улицы.
    """
    from app.services.estimator import _percentile, _resolve_target_city, extract_street_name

    def _empty(reason_street: str | None = None) -> SalesVsListingsResponse:
        return SalesVsListingsResponse(
            street=reason_street,
            period_months=period_months,
            window_days=window_days,
            area_tolerance=area_tolerance,
            total_deals=0,
            deals_with_listings=0,
            linkage_rate_pct=0.0,
            median_discount_pct=None,
            data_quality="no_data",  # #721 ADR v3: нет сделок / улица не извлеклась
            pairs=[],
        )

    street_name = extract_street_name(address)
    if not street_name:
        logger.warning("sales-vs-listings: cannot extract street from %r", address)
        return _empty()

    # #2583 H4 city-scope (зеркало /street-deals #C1, trade_in.py:1717): без него
    # street_pattern матчит одноимённые улицы ЛЮБОГО города обл.66 на ОБЕИХ сторонах
    # JOIN (deals.address / listings.address хранят "<Город>, <Улица>") — прод-аудит
    # показал 49% явно чужого города + 50% NULL-city listings для проверенных стритов,
    # медианный discount_pct уезжал в -60%+ на смеси рынков. target_city резолвится тем
    # же словарём (~30 городов обл.66), что и street-deals; None (адрес вне словаря,
    # известная H1) → фильтр не применяется на TVF-стороне (см. миграцию 205).
    target_city = _resolve_target_city(address)

    rows = (
        db.execute(
            text(
                """
            SELECT
                deal_id, deal_date, deal_price_rub, deal_price_per_m2,
                deal_area_m2, deal_rooms, deal_floor, deal_address,
                listing_id, listing_source, listing_source_url,
                listing_date, listing_price_rub, listing_price_per_m2,
                listing_area_m2, days_listing_to_deal, discount_pct
            FROM street_sales_vs_listings(
                CAST(:street_pattern AS text),
                CAST(:area_m2 AS numeric),
                CAST(:rooms AS integer),
                CAST(:window_days AS integer),
                CAST(:area_tolerance AS numeric),
                CAST(:period_months AS integer),
                CAST(:target_city AS text)
            )
            """
            ),
            {
                "street_pattern": "%" + street_name + "%",
                "area_m2": area_m2,
                "rooms": rooms,
                "window_days": window_days,
                "area_tolerance": area_tolerance,
                "period_months": period_months,
                "target_city": target_city,
            },
        )
        .mappings()
        .all()
    )

    if not rows:
        logger.info(
            "sales-vs-listings: no deals street=%r rooms=%d area=%.1f period_months=%d",
            street_name,
            rooms,
            area_m2,
            period_months,
        )
        return _empty(reason_street=street_name)

    pairs = [
        SalesListingPair(
            deal_id=r["deal_id"],
            deal_date=r["deal_date"],
            deal_price_rub=int(r["deal_price_rub"]),
            deal_price_per_m2=int(r["deal_price_per_m2"] or 0),
            deal_area_m2=float(r["deal_area_m2"]),
            deal_rooms=int(r["deal_rooms"]),
            deal_floor=r["deal_floor"],
            deal_address=r["deal_address"],
            listing_id=r["listing_id"],
            listing_source=r["listing_source"],
            listing_source_url=r["listing_source_url"],
            listing_date=r["listing_date"],
            listing_price_rub=(
                int(r["listing_price_rub"]) if r["listing_price_rub"] is not None else None
            ),
            listing_price_per_m2=(
                int(r["listing_price_per_m2"]) if r["listing_price_per_m2"] is not None else None
            ),
            listing_area_m2=(
                float(r["listing_area_m2"]) if r["listing_area_m2"] is not None else None
            ),
            days_listing_to_deal=r["days_listing_to_deal"],
            discount_pct=(float(r["discount_pct"]) if r["discount_pct"] is not None else None),
            # #1995: street_sales_vs_listings() фильтрует ТОЛЬКО source='rosreestr'
            # (067_v_street_sales_vs_listings.sql) → все pairs сейчас квартальной
            # precision (deal_date = period_start_date). Честная маркировка, не баг.
            deal_date_precision="quarter",
        )
        for r in rows
    ]

    total_deals = len(pairs)
    deals_with_listings = sum(1 for p in pairs if p.listing_id is not None)
    linkage_rate_pct = round(deals_with_listings / total_deals * 100, 1) if total_deals else 0.0

    discounts = sorted(p.discount_pct for p in pairs if p.discount_pct is not None)
    median_discount = round(_percentile(discounts, 0.5), 2) if discounts else None
    # #2672: сколько РАЗЛИЧНЫХ объявлений стоит за этими парами. len(discounts)
    # считает сделки, а не наблюдения рынка — одно объявление попадает в пару
    # к десяткам сделок улицы (см. шапку секции).
    n_distinct_listings = len(
        {p.listing_id for p in pairs if p.discount_pct is not None and p.listing_id is not None}
    )

    # #2672 п.3: те же две проверки, но применённые к КАЖДОЙ СТРОКЕ таблицы, а не
    # к сводному числу (обоснование — в шапке секции, блок «ПОШТУЧНЫЙ ПРОЦЕНТ»).
    # Считаются ДО гейта, потому что гейт обнуляет median_discount, и порядок его
    # веток (мало пар → одно объявление → диапазон) прячет вторую причину за
    # первой: на проде 55 групп гасятся как «мало пар», хотя стоят ещё и на ОДНОМ
    # объявлении. Для строк важна причина, а не то, какая ветка сработала раньше.
    pairs_stand_on_one_listing = n_distinct_listings < SALES_VS_LISTINGS_MIN_DISTINCT_LISTINGS
    median_is_implausible = median_discount is not None and not (
        SALES_VS_LISTINGS_SANE_DISCOUNT_MIN_PCT
        <= median_discount
        <= SALES_VS_LISTINGS_SANE_DISCOUNT_MAX_PCT
    )

    # #2666 гейт правдоподобия (обоснование порогов — в шапке секции). Число либо
    # отдаётся, либо гасится с объяснением ПОЧЕМУ — молча пустое поле пользователь
    # прочитает как поломку, а не как честность.
    median_discount_explanation: str | None = None
    if median_discount is not None:
        if len(discounts) < SALES_VS_LISTINGS_MIN_PAIRS:
            # Формулировка — ФАКТ про выборку, а не обещание надёжности выше
            # порога: 10 пар тоже не гарантия (см. «ПОТОЛОК ГЕЙТА» выше —
            # пары псевдореплики), обещать «от 10 надёжно» мы не вправе.
            median_discount_explanation = (
                f"Медианный торг не показываем: пар «сделка ↔ объявление» всего "
                f"{len(discounts)} — на такой выборке медиана гуляет на десятки "
                f"процентных пунктов."
            )
        elif pairs_stand_on_one_listing:
            # Числа стоят В КОНЦЕ клауз намеренно: «различных объявлений всего 1»
            # грамматично при любом значении, «на 1 различных объявлений» — нет.
            median_discount_explanation = (
                f"Медианный торг не показываем: сделок {len(discounts)}, а разных "
                f"объявлений для сравнения всего {n_distinct_listings} — такой процент "
                f"говорит о цене одной конкретной квартиры, а не о торге на улице."
            )
        elif median_is_implausible:
            # Типографский минус (U+2212) — как в fmtDiscount на фронте.
            shown = f"{median_discount:+.1f}".replace("-", "−")
            # Про «пары строятся по улице, а не по дому» здесь НЕ пишем: ровно
            # следующим блоком это говорит street_only-дисклеймер (карточка) /
            # хвост note (v2-mappers). Проверено скриншотом — две формулировки
            # подряд читались как стена текста.
            median_discount_explanation = (
                f"Медианный торг не показываем: расчёт дал неправдоподобное значение "
                f"({shown}%) — такого торга на рынке не бывает."
            )
        if median_discount_explanation is not None:
            logger.info(
                "sales-vs-listings: median_discount gated street=%r rooms=%d "
                "n_pairs=%d distinct_listings=%d value=%+.2f%%",
                street_name,
                rooms,
                len(discounts),
                n_distinct_listings,
                median_discount,
            )
            median_discount = None

    # #2672 п.3: под погашенной медианой строки таблицы продолжали показывать
    # проценты из ТЕХ ЖЕ пар (живой кейс — Космонавтов: +76%, +73%, +63% против
    # одной и той же цены предложения). Гасим их там, и только там, где причина —
    # свойство самой пары; «мало пар» свойство ВЫБОРКИ, про отдельную строку оно
    # ничего не говорит, поэтому одну строку не трогает (обоснование и цена —
    # в шапке секции). Обе цены остаются в строке: мы убираем не данные, а нашу
    # подпись «торг» под разностью, которой не можем ручаться.
    if discounts and (pairs_stand_on_one_listing or median_is_implausible):
        if pairs_stand_on_one_listing:
            # Оба числа названы совместно с фразой медианы: там «сделок N», здесь
            # «одна и та же цена» — читателю видно и сколько строк, и на скольких
            # объявлениях они стоят.
            row_explanation = (
                "Проценты по каждой сделке тоже не показываем: все они считаются "
                "против одной и той же цены объявления."
            )
        else:
            # Медиана вне диапазона — это утверждение О СТРОКАХ: по определению
            # медианы половина из них лежит по дальнюю сторону от неё, т.е. тоже
            # вне рыночного диапазона. Значение здесь НЕ повторяем: в ветке
            # диапазона оно уже названо предыдущим предложением (вышло бы дважды
            # в одном абзаце), а в ветке «мало пар» мы его намеренно не
            # показываем — и печатать его в пояснении было бы отказом на словах.
            row_explanation = (
                "Проценты по каждой сделке тоже не показываем: половина из них — "
                "за пределами того, как торгуется рынок."
            )
        for pair in pairs:
            pair.discount_pct = None
        median_discount_explanation = (
            f"{median_discount_explanation} {row_explanation}"
            if median_discount_explanation
            else row_explanation
        )
        logger.info(
            "sales-vs-listings: per-row discount_pct gated street=%r rooms=%d rows=%d "
            "distinct_listings=%d reason=%s",
            street_name,
            rooms,
            len(discounts),
            n_distinct_listings,
            "one_listing" if pairs_stand_on_one_listing else "implausible_median",
        )

    logger.info(
        "sales-vs-listings: street=%r deals=%d with_listings=%d distinct_listings=%d "
        "linkage=%.1f%% median_disc=%s",
        street_name,
        total_deals,
        deals_with_listings,
        n_distinct_listings,
        linkage_rate_pct,
        f"{median_discount:+.2f}%" if median_discount is not None else "n/a",
    )

    return SalesVsListingsResponse(
        street=street_name,
        period_months=period_months,
        window_days=window_days,
        area_tolerance=area_tolerance,
        total_deals=total_deals,
        deals_with_listings=deals_with_listings,
        linkage_rate_pct=linkage_rate_pct,
        median_discount_pct=median_discount,
        median_discount_explanation=median_discount_explanation,
        # street_sales_vs_listings матчит по УЛИЦЕ (не по дому, #721 ADR) →
        # даже при deals_with_listings>0 это street-level, не house. house_linked НЕ emit'им.
        data_quality="street_only" if total_deals > 0 else "no_data",
        pairs=pairs,
    )


# ── Coverage probe (#2894) — бесплатный шаг лэндинга, ЦЕНЫ НЕТ ─────────────────
# До оплаты человек видит, СКОЛЬКО похожих квартир продаётся рядом и КАК БЫСТРО
# они уходят — ни одной рублёвой цифры (см. CoverageProbeResponse docstring).
# Один SQL, ноль внешних вызовов, ноль записей — ручка дешёвая специально: её
# планируется открыть анонимам отдельной задачей (#2895, со своим consent-
# гейтом). RBAC здесь НЕ трогаем — путь остаётся закрытым (не в _PUBLIC_PATHS).
# строго 1000м по ТЗ #2894 (НЕ DEFAULT_RADIUS_M эстиматора — тот допускает fallback до 2000)
COVERAGE_RADIUS_M = 1000
COVERAGE_AREA_TOLERANCE = 0.15  # ±15% площади
COVERAGE_FRESH_DAYS = 14  # объявления не старше 14 дней (тот же канон, что LISTINGS_FRESH_DAYS)

# MAJOR-2 (независимый ревью #2894): days_on_market на проде заполнена практически
# только у yandex (avito/cian/domklik — 0 заполнено) — возраст известен у меньшинства
# когорты, и на тонких когортах "медиана" считалась по 1-2 объявлениям. Ниже порога
# n_with_age медиану не отдаём (null) — не продуктовое решение, а честность при
# заведомо шумной статистике по единичным точкам.
COVERAGE_MIN_AGE_SAMPLES = 5
# 15% свежих yandex-строк имеют days_on_market > 365 (максимум 4261) — это почти
# наверняка мёртвое/забытое объявление, которое никто не снял с публикации, а не
# сигнал о реальном времени экспозиции рынка. Отбрасываем как выброс из медианы.
COVERAGE_MAX_AGE_DAYS = 365

# Списки городов и пороги — константа РЯДОМ С РУЧКОЙ (issue #2894 требование), не в БД.
#
# ⚠️ Эти списки обязаны совпадать с `OBLAST_CITIES`
# (frontend/src/lib/city-registry.ts) — тем, что человек видит в дропдауне.
# Расхождение поймано на проде 16.08.2026: Серов предлагался к выбору, но
# отсутствовал здесь, и житель Серова получал «этот адрес вне области, по
# которой мы собираем данные» — про город В ТОЙ ЖЕ области, который мы ему сами
# и предложили. Сверка теперь автоматическая, см.
# tests/test_public_mera_api.py::test_offered_cities_match_coverage_cities.
COVERAGE_GREEN_CITIES = ("Екатеринбург", "Верхняя Пышма", "Берёзовский", "Среднеуральск")
# Серов добавлен 16.08.2026: в жёлтый тир, а не в зелёный — в радиусе 15 км от
# центра 363 активных объявления (все свежие), это на порядок меньше городов
# вокруг Екатеринбурга, но заведомо не ноль.
COVERAGE_YELLOW_CITIES = ("Нижний Тагил", "Каменск-Уральский", "Первоуральск", "Ревда", "Серов")
COVERAGE_GREEN_MIN_N = 8
COVERAGE_YELLOW_MIN_N = 12

# Москва добавлена 10.09.2026 — ОТДЕЛЬНОЙ константой, а не в
# COVERAGE_YELLOW_CITIES. Причина структурная: пара GREEN/YELLOW_CITIES выше —
# это контракт со свердловским дропдауном на сайте (city-registry.ts, сверяется
# тестом test_offered_cities_match_coverage_cities), а Москвы в том дропдауне
# нет и быть не должно — там выбирают город Свердловской области. Порог при
# этом нужен, и живёт он здесь, в том же _COVERAGE_CITY_THRESHOLDS.
#
# Тир — ЖЁЛТЫЙ (12), не зелёный. Замер на проде 10.09.2026: 200 случайных
# московских объявлений, когорта считалась тем же предикатом, что сама проба
# (см. SQL в coverage_probe ниже) — медиана когорты 14, p25 = 8, доля точек с
# когортой >= 12 равна 0.57, с когортой >= 8 равна 0.77. Контроли той же
# метрикой: Екатеринбург (зелёный, порог 8) — медиана 37, доля >= 12 равна
# 0.865; Нижний Тагил (жёлтый, порог 12) — медиана 11, доля >= 12 равна 0.473.
# То есть по плотности когорты Москва втрое реже зелёного эталона и стоит
# вплотную к жёлтому: с порогом 8 «ok» получали бы 77% адресов, но за этим «ok»
# стояла бы заметно менее надёжная оценка. Плюс модельная сторона для Москвы
# ещё не откалибрована (время правится свердловским рядом СберИндекса, полоса
# цен одна на весь город), так что 43% адресов честнее показать как thin.
COVERAGE_MOSCOW_MIN_N = COVERAGE_YELLOW_MIN_N

# Ключ центроида и display-имя города — РАЗНЫЕ вещи, и здесь это видно яснее
# всего: у Москвы 67 центроидов (сетка, см. `_MOSCOW_GRID_DEG` ниже), а наружу,
# в поле `city` ответа, обязано уходить одно имя «Москва» и один порог. Иначе
# повторится ровно тот баг, что был с Берёзовским: человек получает в ответе
# название района вместо своего города.
COVERAGE_MOSCOW_DISPLAY = "Москва"

# СЕТКА МОСКОВСКИХ ЦЕНТРОИДОВ (11.09.2026).
#
# Откуда координаты. Это не справочник районов и не ручной подбор: точки
# получены кластеризацией НАШЕГО ЖЕ корпуса объявлений средствами PostGIS —
# `ST_ClusterKMeans(ST_Transform(geom, 32637), K) OVER ()` по 35 552 активным
# вторичным объявлениям region_code=77 (geom заполнена у 100%), центроид
# кластера = `ST_Centroid(ST_Collect(...))` в UTM 37N, округление до 4 знаков.
# Поэтому точки стоят там, где реально живут объявления, а не там, где на карте
# нарисован центр района. Подписи в ключах — человеческие ярлыки для читаемости
# (ближайший район/поселение), на резолв они не влияют никак.
#
# Почему точек 67. Кластеризация дала K=56: порог качества был «меньше 2%
# московских объявлений уходит к подмосковному центроиду», свип по K на тех же
# данных — 24 → 6.06%, 30 → 4.02%, 36 → 3.67%, 42 → 3.46%, 48 → 2.23%,
# 54 → 2.12%, 56 → 1.79%, 60 → 1.64%, 75 → 0.84%, 90 → 0.45%, и K=56 — первое
# значение, берущее 2%. Из этих 56 один кластер выброшен как артефакт геокода,
# и 12 точек добавлены вручную по замеру 11.09.2026 (жилые районы, целиком
# проигрывавшие конкурс подмосковной точке): 56 − 1 + 12 = 67.
#
# Выброшенный призрак. «Москва (Восточный, эксклав)» 56.0087/37.7960: n=1,
# отрыв от остальной сетки 17.9 км. Адрес объявления — «ВАО, р-н Восточный, 4»,
# а настоящий район Восточный лежит на 55.81/37.85: геокодер промахнулся
# на 22 км. Точка стояла в 3.06 км от Пушкино и раздавала ему имя «Москва».
# Взамен добавлена «Москва (Восточный)» по фактическому центру девяти
# объявлений района. Проверены ВСЕ кластеры с n<60 либо изоляцией >12 км
# (9 штук, адреса подняты с прода) — артефакт ровно один. Второй кандидат,
# «Москва (юг ТАО)» 55.2643/37.1033 (n=1, адрес без улицы и дома =
# settlement-fallback геокодера), ОСТАВЛЕН намеренно: точка стоит внутри ТАО,
# при радиусе 8 км не достаёт ни до одного города области, а снос стоил бы
# покрытия анклава.
#
# Где проходит граница с областью. Двумя механизмами сразу, и нужны оба.
# (1) Конкурс ближайшего центроида: 32 подмосковные точки
# (`_COVERAGE_NEGATIVE_CENTROIDS_DEG` ниже) участвуют в поиске ближайшего, но
# порога не имеют, поэтому адрес, для которого выиграл подмосковный центр,
# получает «город не определён». (2) Радиус, который теперь СВОЙСТВО ЦЕНТРОИДА
# (`_CENTROID_RADIUS_KM` ниже), а не одна константа на всю страну: у московских
# точек он 8 км, у свердловских остались прежние 25.
#
# Почему радиус пришлось сделать свойством точки. 25 км были рассчитаны на ОДИН
# центроид в центре города — круг вокруг Кремля. У плотной сетки круги
# СКЛАДЫВАЮТСЯ, и объединение 67 кругов по 25 км — это уже не «Москва с
# запасом», а пятно, накрывающее половину области. Наро-Фоминск, Кубинка и Чехов
# резолвились как «Москва» именно так: до ближайшей точки сетки им 9.83, 17.18
# и 22.25 км (расстояния посчитаны _haversine_km этого же модуля, сфера
# R=6371 — замер на эллипсоиде PostGIS даёт на 0.1-0.5 км больше), то есть
# внутрь общего круга они попадали, а собственной
# отрицательной точки у них нет. Прежнее утверждение в этом месте — будто
# радиус 25 км «больше не ограничение, запас четырёхкратный» — было ложным:
# ограничение не исчезло, оно поменяло знак. Раньше радиус резал покрытие
# изнутри (адрес дальше 25 км от Кремля терял город), теперь протекал наружу.
#
# Откуда 8 км. Свип по радиусу на итоговой сетке, тот же корпус (35 552
# активных вторичных объявления, region_code=77): R = 5, 6, 7 и 8 дают
# ОДИНАКОВЫЙ результат — 0.00% объявлений вне радиуса, 29/29 московских
# контрольных адресов резолвятся в «Москву», 32/32 областных дают «не
# определён». Верхний край плато жёсткий: при R=10 внутрь входит Наро-Фоминск
# (9.83 км — город БЕЗ отрицательной точки), при R=12 — ещё Голицыно и
# Лыткарино. Берём верхний край: 8 км — это 3.5 км запаса на адреса вне
# корпуса (худший московский контроль — Рублёво, 4.44 км до сетки) и всё ещё
# на 1.83 км ниже потолка 9.83. Брать меньше нечем оправдать: по корпусу
# выигрыша нет, а запас на дырки в сетке тает.
#
# Остаточная цена. 48 московских объявлений из 35 552 = 0.14% (на прежней
# конфигурации — 636 = 1.79%) всё ещё выигрываются подмосковным центром:
# Молжаниновский 15, Реутов 8, Подольск 7, Долгопрудный 7, Мытищи 7,
# Красногорск 3, Королёв 1. Крупнейший кусок снимается 13-й точкой
# «Молжаниновский» 55.9480/37.3478 (0.14% → 0.09%) — она в замере посчитана,
# но в список не внесена, чтобы конфигурация здесь совпадала с той, на которой
# прогнаны оба контрольных набора. Одно объявление не берётся ничем:
# 56.0087/37.7960, тот самый кривой геокод, — лечится перегеокодированием
# строки адреса, а не сеткой.
#
# Не перегенерируйте кластеризацию перед правкой: ST_ClusterKMeans без сида
# недетерминирован, разброс между прогонами ±0.2 п.п. Список ниже перепроверен
# отдельным прогоном именно в том виде, в каком он здесь записан.
_MOSCOW_GRID_DEG: dict[str, tuple[float, float]] = {
    "Москва (Рогово, ТАО)": (55.2146, 37.0695),
    "Москва (юг ТАО)": (55.2643, 37.1033),
    "Москва (Вороновское, ТАО)": (55.3159, 37.1759),
    "Москва (Кленовское, ТАО)": (55.3343, 37.3299),
    "Москва (ТАО у Подольска)": (55.3569, 37.4010),
    "Москва (Шишкин Лес, ТАО)": (55.4202, 37.1687),
    "Москва (Щапово, ТАО)": (55.4212, 37.3979),
    "Москва (Новофёдоровское, ТАО)": (55.4307, 36.8675),
    "Москва (Троицк, юг)": (55.4568, 37.2867),
    "Москва (Остафьево, НАО)": (55.4760, 37.5249),
    "Москва (Киевский, ТАО)": (55.4884, 36.9199),
    "Москва (Троицк)": (55.4953, 37.3150),
    "Москва (Щербинка, НАО)": (55.5046, 37.5578),
    "Москва (Ватутинки, НАО)": (55.5187, 37.3605),
    "Москва (Первомайское, ТАО)": (55.5405, 37.1599),
    "Москва (Коммунарка, НАО)": (55.5474, 37.4927),
    "Москва (Филимонковское, НАО)": (55.5568, 37.3188),
    "Москва (Северное Бутово)": (55.5615, 37.5697),
    "Москва (Саларьево, НАО)": (55.5862, 37.4568),
    "Москва (Марушкино, НАО)": (55.5933, 37.1658),
    "Москва (Бирюлёво Западное)": (55.5945, 37.6447),
    "Москва (Внуковское, НАО)": (55.6030, 37.3796),
    "Москва (Ясенево)": (55.6138, 37.5535),
    "Москва (Зябликово)": (55.6225, 37.7303),
    "Москва (Ново-Переделкино)": (55.6317, 37.3215),
    "Москва (Нагорный)": (55.6425, 37.6245),
    "Москва (Солнцево)": (55.6455, 37.3966),
    "Москва (Тропарёво-Никулино)": (55.6576, 37.4684),
    "Москва (Люблино)": (55.6696, 37.7461),
    "Москва (Обручевский)": (55.6711, 37.5432),
    "Москва (Даниловский)": (55.6921, 37.6397),
    "Москва (Некрасовка)": (55.7033, 37.9093),
    "Москва (Раменки)": (55.7048, 37.4870),
    "Москва (Кузьминки)": (55.7143, 37.7723),
    "Москва (Кунцево)": (55.7281, 37.4239),
    "Москва (Якиманка)": (55.7290, 37.6055),
    "Москва (Таганский)": (55.7370, 37.6845),
    "Москва (Пресненский)": (55.7486, 37.5282),
    "Москва (Новогиреево)": (55.7545, 37.8107),
    "Москва (Тверской)": (55.7655, 37.6038),
    "Москва (Хорошёво-Мнёвники)": (55.7657, 37.4668),
    "Москва (Преображенское)": (55.7886, 37.7045),
    "Москва (Хорошёвский)": (55.7951, 37.5373),
    "Москва (Строгино)": (55.7982, 37.3859),
    "Москва (Северное Измайлово)": (55.8071, 37.7881),
    "Москва (Останкинский)": (55.8086, 37.6174),
    "Москва (Покровское-Стрешнево)": (55.8203, 37.4470),
    "Москва (Митино)": (55.8467, 37.3877),
    "Москва (Отрадное)": (55.8558, 37.5796),
    "Москва (Ховрино)": (55.8581, 37.4924),
    "Москва (Бабушкинский)": (55.8633, 37.6740),
    "Москва (Лианозово)": (55.8944, 37.5622),
    "Москва (Куркино — Молжаниново)": (55.9169, 37.3837),
    "Москва (Зеленоград, юг)": (55.9793, 37.1620),
    "Москва (Зеленоград, север)": (55.9958, 37.2147),
    # Кластер «Москва (Восточный, эксклав)» (56.0087, 37.7960) ВЫБРОШЕН
    # 11.09.2026 как артефакт геокода — см. «Выброшенный призрак» выше.
    #
    # Ниже — 12 точек, добавленных тем же замером вручную: жилые районы
    # Москвы, которые целиком проигрывали конкурс ближайшего подмосковной
    # точке. В комментарии — сколько объявлений возвращает точка и кому они
    # уходили (прод-корпус, is_active AND region_code=77).
    "Москва (Восточный)": (55.8194, 37.8740),  # +9, уходили к Балашихе
    "Москва (Левобережный)": (55.8713, 37.4592),  # +91, к Химкам
    "Москва (Орехово-Борисово Южное)": (55.5986, 37.7233),  # +80, к Развилке
    "Москва (Северный)": (55.9278, 37.5411),  # +79, к Долгопрудному
    "Москва (Митино-запад)": (55.8410, 37.3507),  # +78, к Красногорску
    "Москва (Ивановское)": (55.7717, 37.8326),  # +65, к Реутову
    "Москва (Жулебино)": (55.6885, 37.8521),  # +62, к Люберцам
    "Москва (Можайский-запад)": (55.7008, 37.3936),  # +50, к Немчиновке
    "Москва (Новокосино)": (55.7385, 37.8576),  # +44, к Реутову
    "Москва (Капотня)": (55.6341, 37.7981),  # +29, к Дзержинскому
    "Москва (Куркино)": (55.8895, 37.3980),  # +20, к Химкам
    "Москва (Некрасовка-восток)": (55.6838, 37.9216),  # +13, к Люберцам
}

# Все ключи сетки — один display «Москва» и один порог. Кортеж выводится из
# самого словаря, чтобы список ключей физически не мог разойтись с сеткой.
COVERAGE_MOSCOW_CENTROID_KEYS = tuple(_MOSCOW_GRID_DEG)


def _fold_city(name: str) -> str:
    """ёЁ→еЕ + casefold — та же normalization-идиома, что для адресов (см. #1774)."""
    return name.strip().translate(str.maketrans("ёЁ", "ее")).casefold()


# Ключ — folded ИМЯ ЦЕНТРОИДА, значение — (display-имя города, порог). Для
# свердловских городов ключ и display совпадают, для московских центроидов —
# нет (все 67 ключей сетки дают «Москва»).
_COVERAGE_CITY_THRESHOLDS: dict[str, tuple[str, int]] = {
    **{_fold_city(c): (c, COVERAGE_GREEN_MIN_N) for c in COVERAGE_GREEN_CITIES},
    **{_fold_city(c): (c, COVERAGE_YELLOW_MIN_N) for c in COVERAGE_YELLOW_CITIES},
    **{
        _fold_city(k): (COVERAGE_MOSCOW_DISPLAY, COVERAGE_MOSCOW_MIN_N)
        for k in COVERAGE_MOSCOW_CENTROID_KEYS
    },
}

# Повторная проверка ручки #2894 (2026-08): город раньше резолвился модой
# `listings.city` найденной когорты — оказалось, что `listings.city` это город
# СВИП-контекста скрейпера (миграция 196 — колонка заполняется тем городом,
# который скрейпер обходил, не геокодом самого объявления). Замер на проде:
# в радиусе 1000 м вокруг Берёзовского 90/90 строк имеют city='Екатеринбург';
# вокруг Ревды 74/74 — city='Первоуральск'. Следствие: продавец в Берёзовском
# видел на лэндинге «Екатеринбург», а сами COVERAGE_GREEN/YELLOW_CITIES для
# городов-спутников были НЕДОСТИЖИМЫ (в БД нет ни одной строки с их city).
# Фикс — детерминированный резолв по координатам ЗАПРОСА (никакого участия
# клиента, никакой моды когорты): ближайший центроид города из списка ниже,
# если он в пределах COVERAGE_CITY_MATCH_RADIUS_KM.
#
# Координаты — константа РЯДОМ С РУЧКОЙ, не таблица в БД: единственный
# существующий кандидат на "готовый реестр городов" — это
# frontend/src/lib/city-registry.ts (OBLAST_CITIES) и backend
# geocoder.py::SVERDLOVSK_OBLAST_CITIES — оба хранят ТОЛЬКО текстовые лейблы
# (city_hint для геокодера), без координат. Заводить миграцию + таблицу ради
# статичного справочника из 8 географических центров населённых пунктов —
# оверинжиниринг; координаты (WGS84, общедоступные центры НП) живут здесь же,
# рядом с порогами, которые они резолвят.
# Радиусы матчинга. Радиус — СВОЙСТВО ЦЕНТРОИДА (таблица `_CENTROID_RADIUS_KM`
# ниже, применяется в `_resolve_coverage_city`), потому что 25 км осмысленны
# ровно для конфигурации «один центроид на город», а у Москвы центроидов 67 и
# их круги складываются — разбор в комментарии над `_MOSCOW_GRID_DEG`.
# Свердловские точки радиуса не меняли: у каждой из них один центр на город,
# складываться нечему, а замер, которым выбраны эти 25 км, к сетке отношения
# не имеет.
COVERAGE_CITY_MATCH_RADIUS_KM = 25.0  # дефолт: один центроид на город, дальше — not_covered
COVERAGE_MOSCOW_MATCH_RADIUS_KM = 8.0  # сетка из 67 точек; верхний край плато 5..8, потолок 9.83

_CITY_CENTROIDS_DEG: dict[str, tuple[float, float]] = {
    "Екатеринбург": (56.8389, 60.6057),
    "Верхняя Пышма": (56.9789, 60.5636),
    "Берёзовский": (56.9096, 60.8034),
    "Среднеуральск": (56.9848, 60.4759),
    "Нижний Тагил": (57.9099, 59.9819),
    "Каменск-Уральский": (56.4110, 61.9243),
    "Первоуральск": (56.9083, 59.9483),
    "Ревда": (56.7986, 59.9298),
    "Серов": (59.6047, 60.5772),
    # Москва — сетка из 67 центроидов, одно display-имя и один порог на все.
    # Как сетка получена, почему точек именно столько, где проходит граница
    # с областью и какова остаточная цена — см. большой комментарий над
    # `_MOSCOW_GRID_DEG`. Здесь только подстановка: координаты и ключи живут
    # в одном месте, чтобы их нельзя было рассинхронизировать.
    **_MOSCOW_GRID_DEG,
}

# ОТРИЦАТЕЛЬНЫЕ центроиды: подмосковные города и посёлки, попадающие внутрь
# радиуса московских точек. Список 11.09.2026 перепроверен на итоговой
# конфигурации (67 точек сетки, радиус 8 км): все 32 точки живы — ни одну
# не перехватывает московский центроид, в самой точке выигрывает она сама,
# и ответ остаётся «город не определён». Внутрь 8 км от какой-нибудь точки
# сетки попадают 13 из них (Химки, Реутов, Люберцы, Котельники, Дзержинский,
# Красногорск, Долгопрудный, Мытищи, Видное, Одинцово, Подольск, Апрелевка,
# Селятино) — каждая выигрывает собственным центром. Ближайшая пара после
# добавления новых точек: «Митино-запад» в 1.6 км от Красногорска. Участвуют
# в конкурсе ближайшего центроида наравне с городами из
# `_CITY_CENTROIDS_DEG`, но НАМЕРЕННО отсутствуют в `_COVERAGE_CITY_THRESHOLDS`:
# центроид без порога = «город не определён», тот же ответ, что для точки
# в чистом поле. Обоснование и цена — в комментарии «ГРАНИЦА С ОБЛАСТЬЮ» выше.
#
# Координаты — WGS84, общедоступные центры НП; список закрытый и расширяется
# только по замеру (точка вне радиуса всех московских бесполезна, точка внутри
# Москвы сожгла бы ещё часть покрытия). Ключи здесь и в `_CITY_CENTROIDS_DEG`
# не должны пересекаться — это проверяется тестом.
#
# Города области, которые НЕ попадают ни в один московский радиус, здесь и не
# нужны: Наро-Фоминск (9.83 км до сетки), Кубинка (17.18), Чехов (22.25) отдают
# «город не определён» просто потому, что вне 8 км. При прежних 25 км все трое
# резолвились как «Москва» — это и был дефект, ради которого радиус переехал
# в свойство центроида.
_COVERAGE_NEGATIVE_CENTROIDS_DEG: dict[str, tuple[float, float]] = {
    "Андреевка": (55.9772, 37.1100),
    "Менделеево": (56.0333, 37.2333),
    "Сходня": (55.9500, 37.3000),
    "Чёрная Грязь": (55.9667, 37.3167),
    "Лунёво": (56.0000, 37.3500),
    "Поварово": (56.0667, 37.0667),
    "Дедовск": (55.8672, 37.1200),
    "Нахабино": (55.8500, 37.1833),
    "Реутов": (55.7614, 37.8564),
    "Подольск": (55.4312, 37.5450),
    "Апрелевка": (55.5500, 37.0700),
    "Немчиновка": (55.7050, 37.3450),
    "Химки": (55.8894, 37.4450),
    "Одинцово": (55.6789, 37.2639),
    "Лобня": (56.0100, 37.4750),
    "Мытищи": (55.9116, 37.7308),
    "Котельники": (55.6553, 37.8619),
    "Красногорск": (55.8317, 37.3300),
    "Люберцы": (55.6767, 37.8931),
    "Дзержинский": (55.6294, 37.8500),
    "Развилка": (55.5842, 37.7392),
    "Климовск": (55.3667, 37.5333),
    "Балашиха": (55.7969, 37.9386),
    "Истра": (55.9167, 36.8667),
    "Долгопрудный": (55.9386, 37.5100),
    "Королёв": (55.9142, 37.8256),
    "Видное": (55.5519, 37.7133),
    "Селятино": (55.5081, 36.9825),
    "Томилино": (55.6528, 37.9472),
    "Некрасовский": (56.0500, 37.5500),
    "Барвиха": (55.7333, 37.2333),
    "Железнодорожный": (55.7444, 38.0128),
}


# Радиус матчинга по центроиду: ключ — folded имя центроида, значение — км.
# Чего здесь нет — берёт `COVERAGE_CITY_MATCH_RADIUS_KM` (25 км), то есть все
# свердловские точки. Таблица выводится из самих словарей координат, чтобы
# радиус физически не мог разойтись со списком точек.
#
# Почему у отрицательных точек радиус бесконечный. Отрицательный центроид
# ничего не «накрывает»: он влияет на ответ только там, где он ближе всех
# (в своей ячейке Вороного), а выигравший центроид без порога даёт «город не
# определён» на ЛЮБОМ расстоянии — обе ветки `_resolve_coverage_city` (вышли
# за радиус / нет порога) возвращают один и тот же ответ. Поэтому конечное
# число здесь было бы декорацией: поведение оно не меняет, но читалось бы как
# «дальше отрицательная точка перестаёт действовать», чего не происходит.
# math.inf говорит правду — границу отрицательной точки задаёт геометрия
# соседей, а не радиус, и попытка «подрезать» её числом ничего не вернёт
# Москве, зато спрячет этот факт от следующего читателя.
_CENTROID_RADIUS_KM: dict[str, float] = {
    **{_fold_city(k): COVERAGE_MOSCOW_MATCH_RADIUS_KM for k in COVERAGE_MOSCOW_CENTROID_KEYS},
    **{_fold_city(k): math.inf for k in _COVERAGE_NEGATIVE_CENTROIDS_DEG},
}


def _centroid_radius_km(centroid_key: str) -> float:
    """Радиус матчинга КОНКРЕТНОГО центроида в км (см. `_CENTROID_RADIUS_KM`)."""
    return _CENTROID_RADIUS_KM.get(_fold_city(centroid_key), COVERAGE_CITY_MATCH_RADIUS_KM)


def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Расстояние по большому кругу (км), радиус Земли 6371 км."""
    r_earth_km = 6371.0
    phi1, phi2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlambda = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
    return 2 * r_earth_km * math.asin(math.sqrt(a))


def _resolve_coverage_city(lat: float, lon: float) -> tuple[str, int, bool]:
    """Резолвит (display_city, threshold, is_supported) для пробы покрытия — ПО КООРДИНАТАМ.

    Город = ближайший центроид из `_CITY_CENTROIDS_DEG`, если расстояние до него
    в пределах СОБСТВЕННОГО радиуса этого центроида (`_centroid_radius_km`:
    25 км у свердловских городов, 8 км у точек московской сетки — почему так,
    см. комментарий над `_MOSCOW_GRID_DEG`); иначе город не определён.
    Детерминированно
    и без участия клиента — см. комментарий над `_CITY_CENTROIDS_DEG` про то,
    почему `listings.city` (мода когорты) и `city_hint` (клиентский вход) сюда
    больше НЕ допускаются в качестве источника истины.

    В конкурсе ближайшего участвуют И отрицательные центроиды
    (`_COVERAGE_NEGATIVE_CENTROIDS_DEG`) — подмосковные точки без порога.
    Выигравший центроид без порога означает «город не определён»: так граница
    Москвы и области проходит по конкурсу центров, а не по кругу радиуса
    (см. «ГРАНИЦА С ОБЛАСТЬЮ» над `_CITY_CENTROIDS_DEG`).

    Возвращается display-имя из `_COVERAGE_CITY_THRESHOLDS`, а НЕ ключ
    центроида: у Москвы 67 центроидов на один город, и наружу все они обязаны
    отдавать «Москва» с одним порогом (см. `COVERAGE_MOSCOW_CENTROID_KEYS`).
    """
    nearest_city: str | None = None
    nearest_km = math.inf
    for city, (clat, clon) in (
        *_CITY_CENTROIDS_DEG.items(),
        *_COVERAGE_NEGATIVE_CENTROIDS_DEG.items(),
    ):
        distance_km = _haversine_km(lat, lon, clat, clon)
        if distance_km < nearest_km:
            nearest_km = distance_km
            nearest_city = city

    # Радиус берётся у ПОБЕДИТЕЛЯ конкурса, а не общий: круги плотной сетки
    # складываются, и один глобальный радиус либо режет Свердловскую область,
    # либо протекает из Москвы в область (см. `_CENTROID_RADIUS_KM`).
    if nearest_city is None or nearest_km > _centroid_radius_km(nearest_city):
        return "", 0, False

    # .get(), а не индексирование: ближайшим мог оказаться отрицательный
    # центроид (или новый центроид, для которого забыли завести порог) — это
    # «город не определён», а не KeyError и 500 на живом адресе.
    entry = _COVERAGE_CITY_THRESHOLDS.get(_fold_city(nearest_city))
    if entry is None:
        return "", 0, False

    display, threshold = entry
    return display, threshold, True


@router.post("/coverage", response_model=CoverageProbeResponse)
def coverage_probe(
    payload: CoverageProbeInput,
    db: Annotated[Session, Depends(get_db)],
) -> CoverageProbeResponse:
    """Бесплатная проба покрытия (issue #2894) — сколько похожих квартир рядом.

    Когорта — тот же дедуп/cap-канон, что radius-тиры в estimator._fetch_analogs
    (rn_dup по (source, source_id), rn_addr cap по адресу, реюз тех же
    приватных helper'ов эстиматора — импорт локальный, как и в остальных
    ручках этого файла, чтобы не тащить тяжёлый app.services.estimator
    в module-level import graph): ST_DWithin 1000м, rooms точное совпадение,
    area ±15%, scraped_at не старше 14 дней, is_active.

    MAJOR-1 fix (независимый ревью #2894): когорта пробы обязана быть
    ПОДМНОЖЕСТВОМ когорты платного эстиматора, не шире её — иначе проба честно
    отвечает "ok" там, где платный расчёт увидит 0. Три предиката ниже — тот же
    канон, что estimator._COMMON_WHERE (app/services/estimator.py:5441/5460) и
    inline-копия Tier W (estimator.py:5910/5916/5932, radius-тир, откуда реально
    берутся аналоги на 1000 м): guard новостроек, geo_precision != 'city'
    (#769 Part E — city-centroid листинги без реального адреса), price_rub > 0.

    В ответе НЕТ ни одной цены — см. CoverageProbeResponse docstring.

    MAJOR-2 (независимый ревью #2894): days_on_market на проде фактически
    заполнена только у ОДНОГО источника (yandex) — это ограничение данных, а
    не продуктовое решение. n_with_age в ответе честно считает, по скольким
    объявлениям взята медиана; ниже COVERAGE_MIN_AGE_SAMPLES — null (см. поле
    в ответе). Значения > COVERAGE_MAX_AGE_DAYS (почти наверняка мёртвое
    объявление) в расчёт медианы не берутся.

    #oblast (2026-08): house_placement_history.exposure_days — реальная (не
    цензурированная) экспозиция history-строк — НЕ используется здесь: это
    house-level архив (join по house_id, не привязан к текущей radius/rooms/
    area когорте один-в-один), а не активные листинги в подобранном радиусе;
    сведение двух разных когорт усложнило бы «один дешёвый SQL» без выигрыша
    в честности (у нас и так честное имя поля — age активного объявления, не
    срок продажи). См. openQuestions PR #2894 при ревью.

    Повторная проверка ручки (2026-08): город больше НЕ берётся из моды
    `listings.city` найденной когорты и НЕ зависит от `payload.city_hint` —
    оба источника ненадёжны (см. комментарий над `_CITY_CENTROIDS_DEG`).
    Город резолвится детерминированно по `payload.lat/lon` через
    `_resolve_coverage_city` — `city_hint` в payload остаётся только
    информационным полем (см. `CoverageProbeInput.city_hint`), на результат
    не влияет.
    """
    from app.services.estimator import _RN_DUP_WINDOW, MAX_ANALOGS_PER_ADDRESS

    area_min = payload.area_m2 * (1 - COVERAGE_AREA_TOLERANCE)
    area_max = payload.area_m2 * (1 + COVERAGE_AREA_TOLERANCE)

    row = (
        db.execute(
            text(
                f"""
                WITH base AS (
                    SELECT
                        days_on_market,
                        row_number() OVER (
                            PARTITION BY address ORDER BY scraped_at DESC
                        ) AS rn_addr,
{_RN_DUP_WINDOW}
                    FROM listings
                    WHERE is_active = true
                      AND rooms = :rooms
                      AND area_m2 BETWEEN :area_min AND :area_max
                      AND scraped_at > NOW() - (:fresh_days || ' days')::interval
                      AND ST_DWithin(
                              geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius
                          )
                      -- MAJOR-1: sync с estimator._COMMON_WHERE (5441) / Tier W (5916) —
                      AND price_rub > 0
                      -- MAJOR-1: sync с estimator._COMMON_WHERE (5460) / Tier W (5932) —
                      -- guard новостроек, NULL = legacy вторичка до м.011
                      AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
                      -- MAJOR-1: sync с estimator Tier W (5910/5945-5948, #769 Part E) —
                      -- исключает city-centroid листинги без реального адреса;
                      -- IS DISTINCT FROM пропускает NULL (неизвестная точность)
                      AND (geo_precision IS DISTINCT FROM 'city')
                )
                SELECT
                    count(*) AS n_listings,
                    count(*) FILTER (
                        WHERE days_on_market IS NOT NULL
                          AND days_on_market <= :max_age_days
                    ) AS n_with_age,
                    percentile_cont(0.5) WITHIN GROUP (ORDER BY days_on_market)
                        FILTER (
                            WHERE days_on_market IS NOT NULL
                              AND days_on_market <= :max_age_days
                        ) AS median_age_days
                FROM base
                WHERE rn_addr <= :max_per_addr
                  AND rn_dup = 1
                """
            ),
            {
                "rooms": payload.rooms,
                "area_min": area_min,
                "area_max": area_max,
                "fresh_days": COVERAGE_FRESH_DAYS,
                "lat": payload.lat,
                "lon": payload.lon,
                "radius": COVERAGE_RADIUS_M,
                "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
                "max_age_days": COVERAGE_MAX_AGE_DAYS,
            },
        )
        .mappings()
        .fetchone()
    )

    n_listings = int(row["n_listings"]) if row else 0
    n_with_age = int(row["n_with_age"]) if row and row["n_with_age"] is not None else 0
    median_age = (
        round(row["median_age_days"])
        if row is not None
        and row["median_age_days"] is not None
        and n_with_age >= COVERAGE_MIN_AGE_SAMPLES
        else None
    )

    city, threshold, supported = _resolve_coverage_city(payload.lat, payload.lon)

    if not supported or n_listings == 0:
        status: Literal["ok", "thin", "not_covered"] = "not_covered"
        # Nit-fix (повторная проверка #2894): threshold неприменим при
        # not_covered — см. CoverageProbeResponse.threshold docstring. Раньше
        # поддерживаемый (по координатам) город с пустой когортой отдавал
        # реальный порог (8/12) вместе с not_covered — противоречило докстрингу.
        threshold = 0
    elif n_listings >= threshold:
        status = "ok"
    else:
        status = "thin"

    logger.info(
        "coverage probe rooms=%d area=%.1f city=%r status=%s n=%d n_with_age=%d",
        payload.rooms,
        payload.area_m2,
        city,
        status,
        n_listings,
        n_with_age,
    )

    return CoverageProbeResponse(
        status=status,
        n_listings=n_listings,
        median_listing_age_days=median_age,
        n_with_age=n_with_age,
        radius_m=COVERAGE_RADIUS_M,
        city=city,
        threshold=threshold,
    )