gendesign/tradein-mvp/backend/app/api/public/mera.py

"""Публичный API МЕРЫ (B2C, meraocenka.ru) — анонимный.

ЗАЧЕМ ОТДЕЛЬНЫЙ ПРЕФИКС, А НЕ ОТКРЫТИЕ КУСКА /api/v1/*
-------------------------------------------------------
На `meraocenka.ru` действует allowlist-by-default: Caddy проксирует поимённо
перечисленные пути, всё остальное — 404 (см. корневой Caddyfile, site-блок
meraocenka.ru; регресс — scripts/smoke-mera-perimeter.sh). Чтобы открыть там
API, нужно было выбрать одно из двух:

  (а) пробросить `/trade-in/api/v1/trade-in/coverage` и `.../geocode/suggest`
      поимённо — периметр остаётся узким, но одна опечатка в matcher'е
      (`/trade-in/api/*` вместо точного пути) открывает наружу ВЕСЬ v1: ~20
      ручек, включая PDF расчётов, фотографии объектов, историю и админку;
  (б) завести отдельный префикс, под которым по определению не может лежать
      ничего закрытого, и пробрасывать его целиком.

Выбрано (б). Разница не в удобстве, а в цене ошибки: при (а) безопасность
периметра держится на аккуратности матчера, при (б) — на структуре кода.
Добавить сюда ручку с приватными данными нужно СПЕЦИАЛЬНО (положить файл в
`app/api/public/`), случайно — нельзя.

Тот же принцип, что уже применён на фронте: публичный лэндинг вынесен в
`app/mera-public/` с guard-скриптом на граф импортов, а не помечен флагом
внутри общего дерева.

АНОНИМНОСТЬ
-----------
`rbac_guard` (app/core/rbac.py) требует `X-Authenticated-User` для любого
non-public пути. Все ручки перечислены в `_PUBLIC_PATHS` ТОЧНЫМИ строками —
не префиксом: множество там — frozenset с проверкой `path in ...`, и
добавление префиксной ветки расширило бы механизм, которым пользуется весь
бэкенд, ради одной фичи. Поэтому и чтение по токену — POST с постоянным
путём `/estimate/read`, а не `GET /estimate/{token}`: переменный сегмент
пути потребовал бы ровно такой префиксной ветки (плюс сам токен уехал бы в
access-лог Caddy, см. `PublicEstimateTokenInput`).

ЧТО ЭТИ РУЧКИ ДЕЛАЮТ С ДАННЫМИ
------------------------------
`/suggest` и `/coverage` не пишут в БД ничего: первый — прокси автокомплита,
второй — один SELECT. Это по-прежнему так и меняться не должно.

`/estimate` — единственная, которая ПИШЕТ строку с адресом физлица (issue
#2895), и поэтому устроена иначе: она закрыта флагом
`settings.public_estimate_enabled` (дефолт false → 404) и требует явного
согласия 152-ФЗ в payload'е строго `True` — на уровне схемы, то есть 422
прилетает до входа в хендлер и до любого обращения к БД. Пути удаления
данных в бэкенде всё ещё нет; включать флаг на проде — вместе с ним и с
правкой п.5.5 политики обработки ПДн.

`/estimate/read` возвращает по капабилити-токену ТОЛЬКО бесплатную часть
(`PublicEstimateResult`) — цену и прогнозы продаёт платный контур.

БЮДЖЕТЫ
-------
Общий `RateLimitMiddleware` (300/60с на IP) здесь недостаточен: `/suggest`
через DaData-тир геокодера — платный внешний вызов, то есть абуз стоит денег,
а не только CPU. Поэтому у каждой ручки свой, заведомо более узкий per-IP
бюджет поверх общего — тот же приём, что у анонимного чата поддержки
(app/api/v1/support.py, `_anon_ip_limiter`).

Лимитеры in-process: при нескольких репликах бэкенда бюджет умножится на их
число. Сейчас реплика одна (docker-compose, tradein-backend), что и делает
допущение верным; при масштабировании — выносить в Redis (issue заводить
тогда же, не раньше: преждевременный вынос добавит зависимость без выигрыша).
"""

from __future__ import annotations

import asyncio
import hashlib
import logging
import secrets
from collections.abc import Callable
from datetime import datetime
from typing import Annotated, Literal

from fastapi import APIRouter, Depends, HTTPException, Request, Response
from pydantic import BaseModel, Field
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.api.v1.geocode import SuggestResponse, effective_region_code, suggest_addresses
from app.api.v1.trade_in import coverage_probe, estimate
from app.core.config import settings
from app.core.db import get_db
from app.core.public_request import install_address_log_redaction, public_request_scope
from app.core.ratelimit import SlidingWindowLimiter, _client_ip
from app.schemas.trade_in import (
    CoverageProbeInput,
    CoverageProbeResponse,
    TradeInEstimateInput,
)

logger = logging.getLogger(__name__)

# Публичная форма обещает, что введённый адрес нигде не сохраняется. По базам
# это так, по журналам не было — геокодер печатал запрос открытым текстом, а
# прод пишет stdout в persistent journald. Ставим редакцию логов в момент
# импорта модуля (его импортирует app/main.py) — то есть ровно тогда, когда
# публичные ручки вообще появляются в приложении. Разбор — в
# app/core/public_request.py.
install_address_log_redaction()

router = APIRouter()

# Бюджеты подобраны от живого сценария, а не «на глаз»: человек набирает адрес
# с debounce'ом — это единицы запросов на один адрес, поэтому 40/мин хватает
# на несколько попыток подряд и режет перебор словарём. Проба покрытия — шаг
# осознанный (нажатие кнопки), 15/мин с запасом покрывает «поправил площадь,
# нажал ещё раз».
_SUGGEST_LIMIT = 20
_COVERAGE_LIMIT = 15
# Витрина — один SELECT по своей же маленькой таблице, внешних вызовов нет,
# поэтому бюджет шире соседних: он здесь против перебора-в-цикле, а не против
# денежных трат. Лэндинг дёргает ручку один раз на загрузку страницы.
_SHOWCASE_LIMIT = 60
# Расчёт — самый дорогой шаг публичного контура (геокодер + десяток SQL по
# листингам + запись строки). Бюджет намеренно ниже пробы покрытия: живой
# человек нажимает «рассчитать» единицы раз, а анонимная месячная квота
# (settings.anon_estimate_quota_limit) обходится сменой IP — минутное окно
# делает такой обход дорогим по времени.
_ESTIMATE_LIMIT = 5
# Чтение по токену дешевле расчёта (один SELECT по уникальному индексу), но
# ослаблять его до бесконечности нельзя: перебор токенов — это перебор.
_ESTIMATE_READ_LIMIT = 30
_WINDOW_S = 60.0

_suggest_limiter = SlidingWindowLimiter(limit=_SUGGEST_LIMIT, window_s=_WINDOW_S)
_coverage_limiter = SlidingWindowLimiter(limit=_COVERAGE_LIMIT, window_s=_WINDOW_S)
_showcase_limiter = SlidingWindowLimiter(limit=_SHOWCASE_LIMIT, window_s=_WINDOW_S)
_estimate_limiter = SlidingWindowLimiter(limit=_ESTIMATE_LIMIT, window_s=_WINDOW_S)
_estimate_read_limiter = SlidingWindowLimiter(limit=_ESTIMATE_READ_LIMIT, window_s=_WINDOW_S)

# ── Общий суточный потолок публичных подсказок ──────────────────────────────
#
# Per-IP окна одного клиента ограничивают, но не ограничивают СУММУ. Считаем:
# 20 запросов/мин с одного адреса — это 28 800 в сутки, а весь бесплатный тир
# DaData у проекта — 10 000 в сутки И ОН ОБЩИЙ с закрытым контуром. То есть без
# этого потолка один настойчивый клиент (или один скрипт) за несколько часов
# выедает квоту, и подсказки перестают работать у ПЛАТЯЩИХ пилотов, а не только
# у него. Найдено состязательным ревью и подтверждено на проде: достаточно
# упомянуть в запросе не-екатеринбургский город, чтобы локальный кадастровый
# тир отключился и запрос гарантированно ушёл во внешний сервис.
#
# 2000/сутки — заведомо меньше десятой доли тира: публичная форма не должна
# уметь навредить закрытому контуру в принципе. Порог достижим только абузом
# (живой посетитель тратит единицы запросов на адрес), поэтому исчерпание —
# сигнал, а не штатный режим: логируем ошибкой.
_DAILY_SUGGEST_BUDGET = 2000
_daily_suggest_limiter = SlidingWindowLimiter(limit=_DAILY_SUGGEST_BUDGET, window_s=86_400.0)
_GLOBAL_KEY = "public-suggest"

# ── Потолок одновременных подсказок ─────────────────────────────────────────
#
# Кадастровый тир геокодера уходит в FDW-скан ЧУЖОЙ базы (gendesign) и на
# коротком вводе занимает около секунды, всё это время удерживая соединение из
# пула. Пул общий с закрытым контуром и невелик (дефолт SQLAlchemy 5+10), так
# что полтора десятка одновременных публичных подсказок способны положить
# B2B-запросы в том же процессе — при том, что per-IP лимиты каждого из них
# формально соблюдены.
#
# Ждём слот недолго и отвечаем 429, а не копим очередь: очередь под нагрузкой
# превращается в те же занятые соединения плюс растущий таймаут у клиента.
_SUGGEST_CONCURRENCY = 4
_SUGGEST_SLOT_WAIT_S = 2.0
_suggest_slots = asyncio.Semaphore(_SUGGEST_CONCURRENCY)


def _enforce(limiter: SlidingWindowLimiter, request: Request, what: str) -> None:
    """429 при превышении per-IP бюджета. Попытку регистрируем ДО работы ручки.

    В отличие от отправки сообщения в поддержку (там `record()` только на
    успех, чтобы неудача не съедала бюджет), здесь считаем каждую попытку:
    внешний вызов геокодера тратится и на запросе, который вернёт пусто, —
    иначе перебор мусорными строками не стоил бы атакующему ничего.
    """
    ip = _client_ip(request)
    retry_after = limiter.retry_after(ip)
    if retry_after is not None:
        logger.info("public mera %s rate-limited for %s", what, ip)
        raise HTTPException(
            status_code=429,
            detail="Слишком много запросов. Попробуйте через минуту.",
            headers={"Retry-After": str(int(retry_after) + 1)},
        )
    limiter.record(ip)


def _budget(limiter: SlidingWindowLimiter, what: str) -> Callable[[Request], None]:
    """Per-IP бюджет КАК ЗАВИСИМОСТЬ, а не первой строкой тела.

    Ровно та же поправка места, что уже сделана у
    `_require_public_estimate_enabled` (см. его докстринг): FastAPI решает
    зависимости РАНЬШЕ, чем валидирует тело, поэтому проверка в теле не
    срабатывает на запросах, которые падают на разборе тела — до неё просто не
    доходит.

    Для флага это стоило утечки схемы, для лимитера — неограниченного потока
    отказов: 12 запросов подряд с некорректным телом на `/coverage` дали
    двенадцать ответов и ни одного 429 (замер 30.08.2026). Пока такой вход
    ронял сериализацию 422 (чинится в app/core/http_errors.py), это был поток
    500 с одного адреса; после починки — поток 422, но всё так же мимо бюджета.

    ЧЕГО ЭТА ЗАВИСИМОСТЬ НЕ ЛОВИТ: тело, которое не разбирается как JSON
    вообще — там `RequestValidationError` летит до решения зависимостей. Такой
    запрос стоит один `json.loads` и остаётся под общим `RateLimitMiddleware`
    (300/60с на IP); заводить ради него middleware поверх middleware смысла нет.
    """

    def _dep(request: Request) -> None:
        _enforce(limiter, request, what)

    return _dep


class PublicSuggestInput(BaseModel):
    """Вход публичного автокомплита.

    Телом, а не query-параметрами — см. `public_suggest`.
    """

    q: str = Field(min_length=2, max_length=200)
    limit: int = Field(default=8, ge=1, le=10)
    city_hint: str | None = Field(default=None, max_length=100)
    # #3051: явный регион с фронта (если он его когда-нибудь пришлёт) —
    # приоритетнее вывода из city_hint, см. effective_region_code.
    region_code: int | None = Field(default=None)


def _fold(text: str) -> str:
    """ёЁ→еЕ + casefold — та же нормализация, что у городов в trade_in.py."""
    return text.translate(str.maketrans("ёЁ", "ее")).casefold()


def _query_with_city(query: str, city_hint: str | None) -> str:
    """Подставить выбранный город В САМУ СТРОКУ запроса.

    ЗАЧЕМ. `city_hint` доезжает до геокодера, но НА ВЫДАЧУ ПОДСКАЗОК НЕ ВЛИЯЕТ:
    его использует только екатеринбургский кадастровый тир (как признак «речь
    не про ЕКБ, тир пропускаем»), а DaData-тир ограничен регионом целиком и
    хинта не принимает. Замер на проде 16.08.2026: выбран Серов, введено
    «Ленина 1» → первой подсказкой «Невьянский р-н, пгт Верх-Нейвинский».
    Человек выбирает верхний вариант и считает совсем чужой дом — ровно тот
    баг #2576, ради которого город и спрашивают.

    С городом в строке («Серов Ленина 1») выдача становится серовской целиком —
    проверено там же.

    Для Екатеринбурга подстановка безвредна: три разных адреса дали
    побайтово тот же результат с префиксом и без (кадастровый тир парсит
    улицу и дом одинаково). Поэтому правило одно на все города, без
    исключения для основного трафика — исключение пришлось бы поддерживать.

    Чинится ЗДЕСЬ, а не в геокодере: там от `city_hint` зависит поведение
    закрытого контура (`target_city_ambiguous`), и менять его смысл ради
    публичной формы значит трогать чужой контракт.
    """
    if not city_hint:
        return query
    if _fold(city_hint) in _fold(query):
        return query
    return f"{city_hint}, {query}"


@router.post(
    "/suggest",
    response_model=SuggestResponse,
    dependencies=[Depends(_budget(_suggest_limiter, "suggest"))],
)
async def public_suggest(
    payload: PublicSuggestInput,
    db: Annotated[Session, Depends(get_db)],
) -> SuggestResponse:
    """Автокомплит адреса для публичной формы (Свердловская область).

    ПОЧЕМУ POST У ЧИТАЮЩЕЙ РУЧКИ. Каноничнее был бы GET с `?q=`. Но на
    публичном домене включён access-лог (`/var/log/caddy/meraocenka.ru.log`), а
    он пишет URI целиком — то есть адрес квартиры лёг бы в файл рядом с IP
    посетителя. Мы публично обещаем на этой же странице, что введённый адрес
    нигде не сохраняем; лог — это сохранение. Тело запроса в лог не попадает,
    поэтому обещание остаётся правдой без правки конфигурации логирования (её
    легко потерять при следующем рефакторинге Caddyfile — а тип запроса
    потерять нельзя, сломается сразу и заметно).

    Тот же довод, что у черновика с лэндинга: он едет через sessionStorage, а
    не через query-параметры (frontend `estimate-draft.ts`).

    Делегирует В ТУ ЖЕ функцию, что обслуживает B2B-экран
    (`app.api.v1.geocode.suggest_addresses`), а не повторяет её логику:
    публичная форма обязана резолвить адрес ровно так же, как платный расчёт,
    иначе аноним выберет дом, которого потом «не окажется».

    Отличие от v1 ровно одно — потолок `limit` 10 вместо 15: выдача сверх
    десятка в публичном UI не показывается, а каждый лишний кандидат может
    стоить внешнего вызова.
    """
    # Суточный потолок — ПОСЛЕ per-IP: сначала отсекаем одиночного абузера его
    # собственным лимитом, и только оставшееся считаем в общий бюджет.
    daily_retry = _daily_suggest_limiter.retry_after(_GLOBAL_KEY)
    if daily_retry is not None:
        logger.error(
            "публичные подсказки исчерпали суточный бюджет (%d) — квота геокодера "
            "защищена, но форма на лэндинге сейчас без автокомплита",
            _DAILY_SUGGEST_BUDGET,
        )
        raise HTTPException(
            status_code=429,
            detail="Подсказки адреса временно недоступны. Введите адрес полностью.",
            headers={"Retry-After": str(int(daily_retry) + 1)},
        )
    _daily_suggest_limiter.record(_GLOBAL_KEY)

    try:
        await asyncio.wait_for(_suggest_slots.acquire(), timeout=_SUGGEST_SLOT_WAIT_S)
    except TimeoutError:
        raise HTTPException(
            status_code=429,
            detail="Сервис сейчас занят. Попробуйте ещё раз через несколько секунд.",
            headers={"Retry-After": "5"},
        ) from None

    try:
        # Пометка публичного запроса нужна ровно здесь: внутри `suggest_addresses`
        # геокодер логирует введённую строку, а публичная форма обещает, что
        # адрес не попадает в журналы.
        with public_request_scope():
            return await suggest_addresses(
                q=_query_with_city(payload.q, payload.city_hint),
                limit=payload.limit,
                db=db,
                city_hint=payload.city_hint,
                region_code=effective_region_code(payload.region_code, payload.city_hint),
            )
    finally:
        _suggest_slots.release()


@router.post(
    "/coverage",
    response_model=CoverageProbeResponse,
    dependencies=[Depends(_budget(_coverage_limiter, "coverage"))],
)
def public_coverage(
    payload: CoverageProbeInput,
    db: Annotated[Session, Depends(get_db)],
) -> CoverageProbeResponse:
    """Бесплатная проба покрытия (issue #2894) для публичной формы.

    Делегирует в `app.api.v1.trade_in.coverage_probe` — ту же функцию, что
    вызывает закрытый контур. Копии SQL здесь нет намеренно: разбор #2894
    показал, что стоит когорте пробы разойтись с когортой платного расчёта —
    проба честно отвечает «есть данные» там, где расчёт увидит ноль.

    Ответ не содержит ни одной цены (см. `CoverageProbeResponse`) — бесплатный
    шаг доказывает наличие данных, цену продаёт платный.
    """
    return coverage_probe(payload=payload, db=db)


class LandingStat(BaseModel):
    """Одна витринная величина лэндинга.

    `sample_n` и `note` едут наружу вместе со значением намеренно: цифра без
    размера выборки и без описания измеренного — это ровно тот литерал, который
    лежал во фронте до появления landing_stats. Пусть фронт решает, показывать
    ли их мелким шрифтом, но получить число БЕЗ них он не может.
    """

    value: float | str | None
    sample_n: int | None
    note: str | None
    computed_at: datetime


_STATS_LIMIT = 30
_stats_limiter = SlidingWindowLimiter(limit=_STATS_LIMIT, window_s=_WINDOW_S)

# Все строки витрины — их единицы, LIMIT не нужен, но потолок пусть будет:
# таблица наполняется только ночной задачей, и если она когда-нибудь начнёт
# писать метрику на город, ручка не должна молча вырасти в мегабайты.
_STATS_SQL = text("""
    SELECT metric, value_num, value_text, sample_n, note, computed_at
    FROM landing_stats
    ORDER BY metric
    LIMIT 200
""")


@router.get(
    "/stats",
    response_model=dict[str, LandingStat],
    dependencies=[Depends(_budget(_stats_limiter, "stats"))],
)
def public_stats(
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, LandingStat]:
    """Витринные метрики лэндинга — готовый ночной срез (issue: числа по проду).

    GET, в отличие от соседей: здесь в запросе нет ни адреса, ни чего-либо
    относящегося к посетителю, поэтому довод «URI попадает в access-лог» не
    работает, а кэшируемость GET'а для страницы, которую открывают все, полезна.

    Читает готовые строки, НЕ считает на лету: агрегаты по offer_price_history с
    подзапросами на листинг — секунды, а анонимная ручка, которая стоит секунду
    CPU, это рычаг DoS. Считает их app/tasks/landing_stats.py раз в сутки.

    Пустая таблица — валидные `{}` и 200. Это штатное состояние сразу после
    накатки миграции (задача ещё не отработала) и оно же — состояние «данных для
    метрики нет»: задача не пишет строку, когда мерить нечего. Фронт обязан это
    пережить и не рисовать блок, а не получить 500 и сломанную страницу.

    `value` — числовое value_num, если оно есть; иначе value_text (для метрик,
    у которых значение не число). Оба NULL — отдаём null, а не выдуманный ноль.
    """
    rows = db.execute(_STATS_SQL).fetchall()
    return {
        row.metric: LandingStat(
            value=(float(row.value_num) if row.value_num is not None else row.value_text),
            sample_n=row.sample_n,
            note=row.note,
            computed_at=row.computed_at,
        )
        for row in rows
    }


class ShowcaseDeal(BaseModel):
    """Одна строка витрины «МЕРА сказала X — продали за Y».

    `district` / `floor` / `total_floors` НУЛЛАБЕЛЬНЫ намеренно: этих величин в
    ДКП-данных может не быть, и фронт обязан пережить null, а не получить
    правдоподобную подстановку.

    `street_name` / `street_scheme` — УЛИЦА, А НЕ ДОМ. Номер дома есть у 2.7%
    сделок (разбор в миграции 276), поэтому дома в витрине нет и не будет.
    Оба поля НУЛЛАБЕЛЬНЫ, и null — штатный случай: название сматчилось с OSM у
    550 названий из 654 (92.3% сделок, замер 2026-08-29), остальным схемы нет и
    фронт показывает район.

    `street_scheme` — уже спроецированные SVG-пути окна 840×840 м вокруг центра
    улицы::

        {"street": "улица Краснолесья", "w": 1000, "h": 1000,
         "target": ["M…L…"],                       # подсвеченная улица
         "roads":  [{"c": "primary", "d": "M…L…"}],  # фон, c = класс дороги
         "water":  ["M…L…"],
         "labels": [{"t": "Чкалова", "x": 431.2, "y": 88.0}]}

    Координат окна и констант проекции в схеме НЕТ намеренно: по ней нельзя
    положить (lon, lat) в её систему координат, то есть нарисовать точку дома
    невозможно даже случайно. Схема готовая, а не геометрия, потому что GeoJSON
    того же окна — 7-8 КБ на строку против 2.8-2.9 КБ схемы (замер 2026-08-29).
    правдоподобную подстановку. Улицы и дома в модели нет вовсе — номер дома
    есть у 2.7% сделок (разбор в миграции 276).

    `lat` / `lon` — ЦЕНТРОИД УЛИЦЫ, не дом: 991 различная координата на 34 017
    сделок выборки (≈34 сделки в одной точке) при тех же 2.7% известных домов
    (замер 2026-08-29, миграция 280). Точка честна на масштабе района и улицы
    и НЕ честна на масштабе дома — то же самое написано в `note` строки,
    которая едет рядом. Тоже нуллабельны: у части сделок координаты нет, и
    такая строка остаётся на витрине без точки, а не выбрасывается.
    """

    district: str | None
    rooms: int
    area_m2: float
    floor: int | None
    total_floors: int | None
    deal_quarter: str
    predicted_rub: int
    fact_rub: int
    err_pct: float
    n_analogs: int
    note: str
    street_name: str | None
    street_scheme: dict | None
    lat: float | None = None
    lon: float | None = None


class ShowcaseStats(BaseModel):
    """Итог прогона, который дал показанные строки. Подпись под витриной.

    Без этих чисел «20 отличных строк» неотличимо от «столько и было»:
    посетитель не может отличить выборку из работы оценщика от её лучшего
    хвоста. `eligible` минус `written` — сколько годных строк не поместилось
    в витрину; `rejection_rule` — по какому правилу отсеяно остальное,
    записанное ТЕМ прогоном, который эти строки посчитал.
    """

    considered: int
    priced: int
    no_prediction: int
    incomplete: int
    eligible: int
    written: int
    with_district: int
    rejection_rule: str


class ShowcaseResponse(BaseModel):
    """Витрина целиком: когда считали, что показываем и из чего это отобрано.

    `stats` = None только до первого пересчёта — тогда и `deals` пуст.
    """

    computed_at: str | None
    deals: list[ShowcaseDeal]
    stats: ShowcaseStats | None = None


# Последний прогон — единственная точка отсчёта: и `computed_at`, и счётчики, и
# набор строк берутся ИЗ НЕГО. Брать строки по своему max(computed_at) значило
# бы, что прогон, не давший ни одной строки, показывает вчерашние строки под
# сегодняшними счётчиками.
_SHOWCASE_RUN_SQL = text(
    """
    SELECT computed_at, considered, priced, no_prediction, incomplete,
           eligible, written, with_district, rejection_rule
    FROM landing_showcase_runs
    ORDER BY computed_at DESC, id DESC
    LIMIT 1
    """
)

_SHOWCASE_SQL = text(
    """
    SELECT district, rooms, area_m2, floor, total_floors, deal_quarter,
           predicted_rub, fact_rub, err_pct, n_analogs, note,
           lat, lon, street_name, street_scheme
    FROM landing_showcase_deals
    WHERE computed_at = CAST(:computed_at AS timestamptz)
    ORDER BY id
    """
)


@router.get(
    "/showcase",
    response_model=ShowcaseResponse,
    dependencies=[Depends(_budget(_showcase_limiter, "showcase"))],
)
def public_showcase(
    db: Annotated[Session, Depends(get_db)],
) -> ShowcaseResponse:
    """Витрина лэндинга: реальные ДКП-сделки против прогноза МЕРЫ.

    Читает готовый батч из `landing_showcase_deals` (пересчёт —
    `app/tasks/landing_showcase_deals.py`), а не считает прогноз на лету:
    один прогноз — это несколько пространственных SELECT'ов, двадцать штук на
    анонимный GET были бы рычагом для DoS.

    Пустой список — штатный ответ, а не ошибка: до первого пересчёта показывать
    нечего, и это ровно то, что фронт должен увидеть вместо выдуманных строк.

    Вместе со строками едет `stats` — сколько сделок рассмотрено, сколько
    годных строк не поместилось и по какому правилу отсеяно остальное. Числа
    считает пересчёт; без них витрина не имеет права подписаться честно.
    """
    run = db.execute(_SHOWCASE_RUN_SQL).mappings().first()
    if run is None:
        return ShowcaseResponse(computed_at=None, deals=[], stats=None)
    rows = db.execute(_SHOWCASE_SQL, {"computed_at": run["computed_at"]}).mappings().all()
    return ShowcaseResponse(
        computed_at=run["computed_at"].isoformat(),
        stats=ShowcaseStats(
            considered=int(run["considered"]),
            priced=int(run["priced"]),
            no_prediction=int(run["no_prediction"]),
            incomplete=int(run["incomplete"]),
            eligible=int(run["eligible"]),
            written=int(run["written"]),
            with_district=int(run["with_district"]),
            rejection_rule=run["rejection_rule"],
        ),
        deals=[
            ShowcaseDeal(
                district=r["district"],
                rooms=int(r["rooms"]),
                area_m2=float(r["area_m2"]),
                floor=(int(r["floor"]) if r["floor"] is not None else None),
                total_floors=(int(r["total_floors"]) if r["total_floors"] is not None else None),
                deal_quarter=r["deal_quarter"],
                predicted_rub=int(r["predicted_rub"]),
                fact_rub=int(r["fact_rub"]),
                err_pct=float(r["err_pct"]),
                n_analogs=int(r["n_analogs"]),
                note=r["note"],
                street_name=r["street_name"],
                street_scheme=r["street_scheme"],
                # Порядок не менять: lat — широта (~56.8 в ЕКБ), lon — долгота
                # (~60.6). Перепутанные местами координаты остаются валидными
                # float и уедут на карту точкой в другой стране.
                lat=(float(r["lat"]) if r["lat"] is not None else None),
                lon=(float(r["lon"]) if r["lon"] is not None else None),
            )
            for r in rows
        ],
    )


# ── Анонимный расчёт и повторное чтение его бесплатной части ────────────────
#
# ФЛАГ. Обе ручки ниже мертвы, пока `settings.public_estimate_enabled` не
# включён в .env.runtime: это первая пара публичных ручек, которая ПИШЕТ в
# `trade_in_estimates` адрес физлица, а такое включение — продуктовое решение
# владельца (нужна правка п.5.5 политики обработки ПДн, она сегодня анонимный
# расчёт не описывает), а не следствие мержа кода. Тот же приём, что у
# `payments_enabled`. Отвечаем 404, а не 403: выключенная ручка не должна
# подтверждать, что она существует.
#
# ОБЩИЙ СУТОЧНЫЙ ПОТОЛОК. Per-IP окно ограничивает одного клиента, но не сумму:
# 5/мин с адреса — это 7200 расчётов в сутки, каждый из которых дёргает
# геокодер и пишет строку. Потолок ниже — про защиту закрытого контура и
# квоты геокодера (тот же довод, что у `_DAILY_SUGGEST_BUDGET`), исчерпание —
# сигнал абуза, не штатный режим.
_DAILY_ESTIMATE_BUDGET = 300
_daily_estimate_limiter = SlidingWindowLimiter(limit=_DAILY_ESTIMATE_BUDGET, window_s=86_400.0)
_ESTIMATE_GLOBAL_KEY = "public-estimate"

# Срок жизни капабилити-ссылки. Ссылка — единственный способ анонима вернуться
# к своему расчёту (аккаунта у него нет), поэтому недели мало не будет для
# «оплатил, закрыл вкладку, вернулся вечером», а бесконечной она быть не может:
# это ссылка на данные о конкретной квартире конкретного человека.
_TOKEN_TTL = "7 days"


def _require_public_estimate_enabled() -> None:
    """404, пока публичный расчёт не включён владельцем явно.

    ВЕШАЕТСЯ ЧЕРЕЗ `dependencies=[Depends(...)]` НА ДЕКОРАТОР, А НЕ ВЫЗЫВАЕТСЯ
    ПЕРВОЙ СТРОКОЙ ТЕЛА. Разница не косметическая: FastAPI решает зависимости
    РАНЬШЕ, чем валидирует тело запроса. Пока проверка стояла в теле, до неё
    просто не доходило — невалидное тело отбивалось 422 ещё на разборе, и
    выключенная ручка отвечала так:

        POST /api/public/mera/estimate  {}          → 422 + перечень полей
        POST /api/public/mera/estimate  {валидное}  → 404
        POST /api/public/mera/nosuchthing           → 401 (rbac)

    То есть 422 подтверждал существование ручки (несуществующий путь даёт 401)
    и заодно выдавал её схему: address, area_m2, rooms, consent. Замер на проде
    29.08.2026, ровно то, что этот докстринг обещал не делать.

    Как зависимость проверка срабатывает до разбора тела, и выключенная ручка
    неотличима от отсутствующей при ЛЮБОМ входе.
    """
    if not settings.public_estimate_enabled:
        raise HTTPException(status_code=404, detail="Not Found")


def _token_hash(token: str) -> str:
    """sha256(hex) — в БД лежит только это, сам токен не хранится нигде.

    Без соли намеренно: вход — `secrets.token_urlsafe(32)` (256 бит), перебор
    по словарю невозможен, а соль сделала бы невозможным поиск по равенству.
    """
    return hashlib.sha256(token.encode()).hexdigest()


class PublicEstimateInput(TradeInEstimateInput):
    """Вход анонимного расчёта: тот же payload, что у закрытого контура, но
    согласие 152-ФЗ — ОБЯЗАТЕЛЬНОЕ и строго True.

    В базовой схеме `consent: bool | None = None`: сделать его обязательным там
    нельзя — B2B-пилоты согласия в UI не дают, у них договор, и их фронт поля
    не шлёт. Здесь же анонимный посетитель — единственный источник согласия,
    поэтому `Literal[True]`: без него Pydantic отвечает 422 ДО входа в
    хендлер, то есть до первого обращения к БД. Это не дубль гейта в
    `estimate_quality` (тот ловит любых вызывающих), а его сдвиг на самую
    раннюю возможную границу — «согласие фиксируется до первого INSERT»
    перестаёт зависеть от порядка строк внутри эстиматора.
    """

    consent: Literal[True]


class PublicEstimateTokenInput(BaseModel):
    """Токен едет ТЕЛОМ, а ручка чтения — POST, а не GET /{token}.

    Причина ровно та же, по которой POST'ом сделан `/suggest`: access-лог Caddy
    на публичном домене пишет URI целиком, поэтому капабилити-ссылка в пути
    легла бы в файл рядом с IP посетителя — и любой, у кого есть доступ к
    логам (или их бэкапу), открыл бы чужой расчёт. Токен — это пароль; пароли
    в URL не кладут.
    """

    token: str = Field(min_length=16, max_length=128)


class PublicEstimateResult(BaseModel):
    """БЕСПЛАТНАЯ часть расчёта — ровно то, что можно показать до оплаты.

    Здесь СОЗНАТЕЛЬНО нет ни одного поля из `AggregatedEstimate` с ценой,
    прогнозом или списком аналогов (`median_price_rub`, `range_*`,
    `expected_sold_*`, `analogs`, `actual_deals`, `market_percentile`,
    `est_days_on_market`, `cian_valuation`, `avito_imv`, `dkp_corridor`…).
    Модель отдельная, а не `AggregatedEstimate` с `exclude`: список исключений
    надо помнить и дополнять при каждом новом поле эстиматора, а отдельная
    модель молчит по умолчанию — новое платное поле не утечёт само.

    `coverage` — тот же ответ, что у бесплатной пробы `/coverage` (в нём цен
    нет по построению, см. `CoverageProbeResponse`): число похожих квартир,
    возраст объявлений, вердикт покрытия. null — координаты дома не
    разрезолвились, честнее отдать «неизвестно», чем правдоподобное число.
    """

    token: str
    token_expires_at: datetime
    n_analogs: int
    coverage: CoverageProbeResponse | None = None


def _coverage_for(
    db: Session, lat: float | None, lon: float | None, rooms: int, area_m2: float
) -> CoverageProbeResponse | None:
    """Вердикт покрытия для уже посчитанной оценки — через ту же `coverage_probe`.

    Своей копии порогов здесь нет намеренно: разъедься она с бесплатной пробой,
    один и тот же адрес получил бы «есть данные» на одном экране и «мало» на
    соседнем.
    """
    if lat is None or lon is None:
        return None
    return coverage_probe(
        payload=CoverageProbeInput(lat=lat, lon=lon, rooms=rooms, area_m2=area_m2),
        db=db,
    )


@router.post(
    "/estimate",
    response_model=PublicEstimateResult,
    # Порядок несущий: флаг ПЕРВЫМ. Лимитер впереди него отвечал бы 429 на
    # выключенной ручке, а несуществующий путь даёт 401 — то есть 429 снова
    # подтверждал бы существование ручки, ровно то, что чинил флаг-гейт.
    dependencies=[
        Depends(_require_public_estimate_enabled),
        Depends(_budget(_estimate_limiter, "estimate")),
    ],
)
async def public_estimate(
    request: Request,
    response: Response,
    payload: PublicEstimateInput,
    db: Annotated[Session, Depends(get_db)],
) -> PublicEstimateResult:
    """Анонимный расчёт: считает полную оценку, отдаёт только бесплатную часть.

    Делегирует в `app.api.v1.trade_in.estimate` — ту же функцию, что обслуживает
    закрытый контур, а не копию её тела. Оттуда же бесплатно достаётся всё
    анти-абузное хозяйство: анонимная квота на связку (подписанная cookie + IP)
    с лимитом `settings.anon_estimate_quota_limit`, семафор одновременности,
    503 вместо непрозрачного 502 при сбое и consent-гейт (`require_consent`
    включается ровно потому, что мы зовём её без `X-Authenticated-User`).

    Полный расчёт при этом СОХРАНЯЕТСЯ в `trade_in_estimates` целиком — платный
    контур (сосед, `/api/v1/trade-in/r/{token}`) открывает его после оплаты по
    своему токену. Наружу здесь уезжает только `PublicEstimateResult`.
    """
    daily_retry = _daily_estimate_limiter.retry_after(_ESTIMATE_GLOBAL_KEY)
    if daily_retry is not None:
        logger.error(
            "публичные расчёты исчерпали суточный бюджет (%d) — закрытый контур "
            "защищён, но форма на лэндинге сейчас не считает",
            _DAILY_ESTIMATE_BUDGET,
        )
        raise HTTPException(
            status_code=429,
            detail="Расчёт временно недоступен. Попробуйте позже.",
            headers={"Retry-After": str(int(daily_retry) + 1)},
        )
    _daily_estimate_limiter.record(_ESTIMATE_GLOBAL_KEY)

    # Пометка публичного запроса — как у `/suggest`: внутри цепочки геокодер
    # печатает введённый адрес, а публичная форма обещает обратное.
    with public_request_scope():
        result = await estimate(
            payload=payload,
            request=request,
            response=response,
            db=db,
            x_authenticated_user=None,
        )

    # Токен выдаём ПОСЛЕ успешного расчёта: ссылка на несуществующий результат
    # не нужна никому, а строка в БД уже есть — estimate() её закоммитил.
    token = secrets.token_urlsafe(32)
    row = db.execute(
        text(
            """
            UPDATE trade_in_estimates
               SET public_token_hash = :token_hash,
                   public_token_expires_at = NOW() + CAST(:ttl AS interval)
             WHERE id = CAST(:id AS uuid)
         RETURNING public_token_expires_at
            """
        ),
        {"token_hash": _token_hash(token), "ttl": _TOKEN_TTL, "id": str(result.estimate_id)},
    ).fetchone()
    db.commit()
    if row is None:  # pragma: no cover — оценка только что записана этой же транзакцией
        raise HTTPException(status_code=503, detail="estimate temporarily unavailable")

    return PublicEstimateResult(
        token=token,
        token_expires_at=row.public_token_expires_at,
        n_analogs=result.n_analogs,
        coverage=_coverage_for(
            db, result.target_lat, result.target_lon, payload.rooms, payload.area_m2
        ),
    )


@router.post(
    "/estimate/read",
    response_model=PublicEstimateResult,
    # Флаг первым — по той же причине, что у `/estimate`.
    dependencies=[
        Depends(_require_public_estimate_enabled),
        Depends(_budget(_estimate_read_limiter, "estimate-read")),
    ],
)
def public_estimate_read(
    payload: PublicEstimateTokenInput,
    db: Annotated[Session, Depends(get_db)],
) -> PublicEstimateResult:
    """Повторное чтение бесплатной части по капабилити-токену.

    Существует потому, что у анонима нет аккаунта: без этой ручки результат
    жил бы ровно в теле POST-ответа и не переживал бы перезагрузку страницы
    (`_assert_estimate_access` в закрытом контуре отдаёт 404 на оценку с
    `created_by IS NULL` всем, кроме админа — и это правильно, менять его
    ради анонима значило бы ослабить IDOR-гейт для всех).

    Просрочка и «нет такого токена» отвечают ОДИНАКОВО (404): различать их
    значит подтверждать существование расчёта тому, кто угадал токен.
    """
    row = db.execute(
        text(
            """
            SELECT n_analogs, lat, lon, rooms, area_m2, public_token_expires_at
              FROM trade_in_estimates
             WHERE public_token_hash = :token_hash
               AND public_token_expires_at > NOW()
            """
        ),
        {"token_hash": _token_hash(payload.token)},
    ).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="Расчёт не найден или ссылка устарела.")

    return PublicEstimateResult(
        token=payload.token,
        token_expires_at=row.public_token_expires_at,
        n_analogs=row.n_analogs,
        coverage=_coverage_for(db, row.lat, row.lon, row.rooms, row.area_m2),
    )