gendesign/tradein-mvp/backend/app/services/buildings_query.py

"""SQL-билдеры страницы «доля квартир дома в продаже» (мигр. 143).

Читаем готовый view v_building_sale_share — логику доли НЕ пересчитываем.
Все запросы параметризованы (SQLAlchemy text + bind params, CAST(:x AS type) —
psycopg v3). Сортировка — строго по whitelist-колонкам, user-ввод в SQL не
интерполируется.
"""

from __future__ import annotations

# share_desc / active_desc / exposure_desc / price_asc / price_desc → безопасный ORDER BY.
# house_id как стабильный tiebreaker для детерминированной пагинации.
_SORT_SQL: dict[str, str] = {
    "share_desc": "sale_share_pct DESC NULLS LAST, house_id",
    "active_desc": "active_secondary DESC NULLS LAST, house_id",
    "exposure_desc": "avg_days_on_market DESC NULLS LAST, house_id",
    "price_asc": "median_price_rub ASC NULLS LAST, house_id",
    "price_desc": "median_price_rub DESC NULLS LAST, house_id",
}
ALLOWED_SORTS: frozenset[str] = frozenset(_SORT_SQL)

# Гистограмма распределения sale_share_pct — фиксированный порядок корзин.
HISTOGRAM_BUCKETS: tuple[str, ...] = (
    "0-5",
    "5-10",
    "10-20",
    "20-30",
    "30-50",
    "50-100",
    "100+",
)

# Колонки view, отдаваемые в список (в порядке SELECT). median_* округляем до
# bigint — percentile_cont отдаёт double precision (может быть x.5), а Pydantic
# int-поле на дробном float падает.
_LIST_COLUMNS = (
    "house_id, "
    "COALESCE(short_address, full_address, address) AS address, "
    "lat, lon, "
    "sale_share_pct, "
    "(sale_share_pct > 100) AS over_100, "
    "active_secondary, "
    "flat_count_effective, "
    "gar_match_method, "
    "CAST(round(median_price_rub) AS bigint) AS median_price_rub, "
    "CAST(round(median_price_per_m2) AS bigint) AS median_price_per_m2, "
    "avg_days_on_market, "
    "year_built, house_type, total_floors, series_name, is_emergency"
)


def build_sale_share_query(
    *,
    min_pct: float = 5.0,
    max_pct: float | None = None,
    city: str | None = None,
    price_min: int | None = None,
    price_max: int | None = None,
    year_min: int | None = None,
    year_max: int | None = None,
    house_type: str | None = None,
    sort: str = "share_desc",
    limit: int = 200,
) -> tuple[str, dict[str, object]]:
    """SELECT домов из v_building_sale_share по фильтрам.

    Всегда требует sale_share_pct IS NOT NULL (строки без знаменателя — только
    для coverage в summary, не в списке).
    """
    where: list[str] = ["sale_share_pct IS NOT NULL"]
    args: dict[str, object] = {}

    where.append("sale_share_pct >= CAST(:min_pct AS numeric)")
    args["min_pct"] = min_pct

    if max_pct is not None:
        where.append("sale_share_pct <= CAST(:max_pct AS numeric)")
        args["max_pct"] = max_pct

    if city:
        # case-insensitive substring по всем адресным полям (город может быть в любом).
        where.append(
            "(COALESCE(short_address, '') || ' ' || COALESCE(full_address, '') "
            "|| ' ' || COALESCE(address, '')) ILIKE CAST(:city_like AS text)"
        )
        args["city_like"] = f"%{city}%"

    if price_min is not None:
        where.append("median_price_rub >= CAST(:price_min AS bigint)")
        args["price_min"] = price_min
    if price_max is not None:
        where.append("median_price_rub <= CAST(:price_max AS bigint)")
        args["price_max"] = price_max

    if year_min is not None:
        where.append("year_built >= CAST(:year_min AS integer)")
        args["year_min"] = year_min
    if year_max is not None:
        where.append("year_built <= CAST(:year_max AS integer)")
        args["year_max"] = year_max

    if house_type:
        where.append("house_type = CAST(:house_type AS text)")
        args["house_type"] = house_type

    order_sql = _SORT_SQL.get(sort, _SORT_SQL["share_desc"])
    args["limit"] = max(1, min(int(limit), 1000))

    sql = (
        f"SELECT {_LIST_COLUMNS} "
        "FROM v_building_sale_share "
        f"WHERE {' AND '.join(where)} "
        f"ORDER BY {order_sql} "
        "LIMIT CAST(:limit AS integer)"
    )
    return sql, args


def build_listings_query(house_id: int) -> tuple[str, dict[str, object]]:
    """Активные вторичные объявления одного дома (ORDER BY price_rub)."""
    sql = (
        "SELECT id AS listing_id, source, source_url, price_rub, price_per_m2, "
        "rooms, CAST(area_m2 AS double precision) AS area_m2, floor, total_floors, "
        "days_on_market, listing_date, address "
        "FROM listings "
        "WHERE house_id_fk = CAST(:house_id AS bigint) "
        "AND is_active AND listing_segment = 'vtorichka' "
        "ORDER BY price_rub ASC NULLS LAST, id"
    )
    return sql, {"house_id": house_id}


def build_summary_scalars_query() -> str:
    """Скаляры сводки: всего домов, с знаменателем, coverage, max, p95."""
    return (
        "SELECT "
        "count(*) AS total_secondary_buildings, "
        "count(*) FILTER (WHERE sale_share_pct IS NOT NULL) "
        "AS buildings_with_denominator, "
        "COALESCE(round("
        "100.0 * count(*) FILTER (WHERE sale_share_pct IS NOT NULL) "
        "/ NULLIF(count(*), 0), 1), 0) AS coverage_pct, "
        "max(sale_share_pct) AS max_pct, "
        "percentile_cont(0.95) WITHIN GROUP (ORDER BY sale_share_pct) AS p95_pct "
        "FROM v_building_sale_share"
    )


def build_summary_histogram_query() -> str:
    """Гистограмма sale_share_pct по фиксированным корзинам (только non-null)."""
    return (
        "SELECT "
        "CASE "
        "WHEN sale_share_pct < 5 THEN '0-5' "
        "WHEN sale_share_pct < 10 THEN '5-10' "
        "WHEN sale_share_pct < 20 THEN '10-20' "
        "WHEN sale_share_pct < 30 THEN '20-30' "
        "WHEN sale_share_pct < 50 THEN '30-50' "
        "WHEN sale_share_pct <= 100 THEN '50-100' "
        "ELSE '100+' "
        "END AS bucket, "
        "count(*) AS count "
        "FROM v_building_sale_share "
        "WHERE sale_share_pct IS NOT NULL "
        "GROUP BY bucket"
    )