gendesign/backend/app/services/analytics_queries.py

"""SQL queries for /api/v1/analytics endpoints.

One function per endpoint. All return plain dicts/lists ready for JSON.
Region 66 = Sverdlovskaya oblast. Developer 6208_0 = PRINZIP.
"""

from __future__ import annotations

from decimal import Decimal
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session


def _f(value: Any) -> float | None:
    if value is None:
        return None
    if isinstance(value, Decimal):
        return float(value)
    return value


def market_pulse(db: Session, region_code: int = 66) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT snapshot_date, rep_year, rep_month,
                   total_square, sold_perc, price_avg
              FROM domrf_realization
             WHERE region_code = :region_code
               AND endpoint_type = 'total'
               AND type_square = 'total'
             ORDER BY snapshot_date
            """
            ),
            {"region_code": region_code},
        )
        .mappings()
        .all()
    )
    return [
        {
            "snapshot_date": r["snapshot_date"].isoformat(),
            "rep_year": r["rep_year"],
            "rep_month": r["rep_month"],
            "total_square_th_sqm": _f(r["total_square"]),
            "sold_perc": _f(r["sold_perc"]),
            "price_avg": _f(r["price_avg"]),
        }
        for r in rows
    ]


def quartirography(db: Session, source: str, region_id: int = 66) -> list[dict[str, Any]]:
    """source: 'portfolio' (что строится) or 'deals' (реально покупают)."""
    if source == "portfolio":
        rows = (
            db.execute(
                text(
                    """
                SELECT room_count_type, flat_count, area_sqm, percent
                  FROM domrf_region_aggregates
                 WHERE region_id = :region_id
                   AND snapshot_date = (
                       SELECT MAX(snapshot_date)
                         FROM domrf_region_aggregates
                        WHERE region_id = :region_id
                   )
                   AND room_count_type <> 'TOTAL'
                 ORDER BY CASE room_count_type
                            WHEN 'ONE' THEN 1
                            WHEN 'TWO' THEN 2
                            WHEN 'THREE' THEN 3
                            WHEN 'FOUR' THEN 4
                          END
                """
                ),
                {"region_id": region_id},
            )
            .mappings()
            .all()
        )
        return [
            {
                "bucket": {
                    "ONE": "1-к",
                    "TWO": "2-к",
                    "THREE": "3-к",
                    "FOUR": "4+",
                }.get(r["room_count_type"], r["room_count_type"]),
                "flat_count": r["flat_count"],
                "area_sqm": _f(r["area_sqm"]),
                "percent": r["percent"],
                "avg_area": _f(r["area_sqm"] / r["flat_count"]) if r["flat_count"] else None,
            }
            for r in rows
        ]

    # deals: bucketize Rosreestr area into 5 segments (студия, 1-к, 2-к, 3-к, 4+).
    # Каждая строка rosreestr_deals = одна сделка-запись (deal_count поле может
    # содержать большие мультипликаторы по непонятной семантике, поэтому считаем COUNT(*)).
    rows = (
        db.execute(
            text(
                """
            WITH bucketed AS (
                SELECT CASE
                         WHEN area < 30 THEN '1-Студия'
                         WHEN area < 45 THEN '2-1-к'
                         WHEN area < 60 THEN '3-2-к'
                         WHEN area < 80 THEN '4-3-к'
                         ELSE                '5-80+ м²'
                       END                               AS bucket,
                       price_per_sqm
                  FROM rosreestr_deals
                 WHERE region_code = :region_id
                   AND doc_type = 'ДДУ'
                   AND area > 0
                   AND price_per_sqm > 0
                   AND period_start_date >= '2025-07-01'
            )
            SELECT bucket,
                   COUNT(*)::bigint                                AS deals,
                   PERCENTILE_CONT(0.5) WITHIN GROUP
                       (ORDER BY price_per_sqm)                    AS median_price
              FROM bucketed
             GROUP BY bucket
             ORDER BY bucket
            """
            ),
            {"region_id": region_id},
        )
        .mappings()
        .all()
    )
    pretty = {
        "1-Студия": "Студии 15-30",
        "2-1-к": "1-к 30-45",
        "3-2-к": "2-к 45-60",
        "4-3-к": "3-к 60-80",
        "5-80+ м²": "80+ м²",
    }
    total = sum(r["deals"] or 0 for r in rows) or 1
    return [
        {
            "bucket": pretty[r["bucket"]],
            "deals": int(r["deals"] or 0),
            "percent": round((r["deals"] or 0) * 100 / total, 1),
            "median_price": _f(r["median_price"]),
        }
        for r in rows
    ]


def pipeline_by_year(db: Session, region_code: int = 66) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT subject_desc                AS year,
                   total_square                AS total_th_sqm,
                   sold_perc, unsold_perc, unopened_perc
              FROM domrf_realization
             WHERE region_code = :region_code
               AND endpoint_type = 'ready_year'
               AND type_square = 'total'
               AND snapshot_date = (
                   SELECT MAX(snapshot_date)
                     FROM domrf_realization
                    WHERE region_code = :region_code
                      AND endpoint_type = 'ready_year'
               )
             ORDER BY subject
            """
            ),
            {"region_code": region_code},
        )
        .mappings()
        .all()
    )
    return [
        {
            "year": r["year"],
            "total_th_sqm": _f(r["total_th_sqm"]),
            "sold_perc": _f(r["sold_perc"]),
            "unsold_perc": _f(r["unsold_perc"]),
            "unopened_perc": _f(r["unopened_perc"]),
        }
        for r in rows
    ]


def districts(db: Session) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT district_name, zk_count, flat_count, area_m2,
                   median_price_per_m2, mean_price_per_m2
              FROM ekb_districts
             WHERE district_name <> 'не определён'
             ORDER BY zk_count DESC NULLS LAST
            """
            )
        )
        .mappings()
        .all()
    )
    return [
        {
            "district_name": r["district_name"],
            "zk_count": r["zk_count"],
            "flat_count": r["flat_count"],
            "area_m2": _f(r["area_m2"]),
            "median_price_per_m2": _f(r["median_price_per_m2"]),
            "mean_price_per_m2": _f(r["mean_price_per_m2"]),
        }
        for r in rows
    ]


def yandex_listings(db: Session) -> dict[str, Any]:
    rows = (
        db.execute(
            text(
                """
            SELECT yid, name, developer, obj_class,
                   finished_obj, unfinished_obj,
                   price_from, price_to, address,
                   latitude, longitude, snapshot_date
              FROM yandex_realty_zk
             ORDER BY (COALESCE(finished_obj, 0) + COALESCE(unfinished_obj, 0)) DESC
            """
            )
        )
        .mappings()
        .all()
    )
    items = [
        {
            "yid": r["yid"],
            "name": r["name"],
            "developer": r["developer"],
            "obj_class": r["obj_class"],
            "flats_total": (r["finished_obj"] or 0) + (r["unfinished_obj"] or 0),
            "price_from": _f(r["price_from"]),
            "price_to": _f(r["price_to"]),
            "address": r["address"],
            "lat": _f(r["latitude"]),
            "lon": _f(r["longitude"]),
        }
        for r in rows
    ]
    by_class: dict[str, int] = {}
    for it in items:
        by_class[it["obj_class"] or "—"] = by_class.get(it["obj_class"] or "—", 0) + 1
    return {
        "snapshot_date": rows[0]["snapshot_date"].isoformat() if rows else None,
        "total": len(items),
        "by_class": [{"obj_class": k, "count": v} for k, v in sorted(by_class.items())],
        "items": items,
    }


def top_developers(db: Session, region_code: int = 66, limit: int = 15) -> list[dict[str, Any]]:
    """Top developers in Sverdl by sqm + Δ sold% over the available history.

    Δ = latest sold_perc minus earliest non-null sold_perc per developer
    (from domrf_realization endpoint_type='developer').
    """
    rows = (
        db.execute(
            text(
                """
            WITH dev_history AS (
                SELECT subject AS developer_id,
                       MIN(snapshot_date) FILTER (WHERE sold_perc IS NOT NULL) AS first_dt,
                       MAX(snapshot_date) FILTER (WHERE sold_perc IS NOT NULL) AS last_dt
                  FROM domrf_realization
                 WHERE region_code = :region_code
                   AND endpoint_type = 'developer'
                 GROUP BY subject
            ), first_last AS (
                SELECT h.developer_id,
                       (SELECT sold_perc FROM domrf_realization r
                         WHERE r.region_code = :region_code
                           AND r.endpoint_type = 'developer'
                           AND r.subject = h.developer_id
                           AND r.snapshot_date = h.first_dt
                           AND r.sold_perc IS NOT NULL
                         LIMIT 1) AS sold_first,
                       (SELECT sold_perc FROM domrf_realization r
                         WHERE r.region_code = :region_code
                           AND r.endpoint_type = 'developer'
                           AND r.subject = h.developer_id
                           AND r.snapshot_date = h.last_dt
                           AND r.sold_perc IS NOT NULL
                         LIMIT 1) AS sold_last,
                       h.first_dt, h.last_dt
                  FROM dev_history h
            )
            SELECT m.developer_id, m.developer_name,
                   m.jk_count, m.jk_flats_total,
                   m.sverdl_sqm, m.sverdl_sold_pct,
                   m.avg_area_sqm, m.pct_one, m.pct_three_plus,
                   fl.sold_first, fl.sold_last,
                   (fl.sold_last - fl.sold_first)              AS sold_delta_pp,
                   fl.first_dt, fl.last_dt
              FROM v_developer_full_metrics m
              LEFT JOIN first_last fl ON fl.developer_id = m.developer_id
             WHERE m.sverdl_sqm IS NOT NULL
             ORDER BY m.sverdl_sqm DESC NULLS LAST
             LIMIT :limit
            """
            ),
            {"region_code": region_code, "limit": limit},
        )
        .mappings()
        .all()
    )
    return [
        {
            "developer_id": r["developer_id"],
            "developer_name": r["developer_name"],
            "jk_count": r["jk_count"],
            "jk_flats_total": r["jk_flats_total"],
            "sverdl_sqm_th": _f(r["sverdl_sqm"]),
            "sold_pct": _f(r["sverdl_sold_pct"]),
            "sold_delta_pp": _f(r["sold_delta_pp"]),
            "sold_first": _f(r["sold_first"]),
            "sold_last": _f(r["sold_last"]),
            "first_dt": r["first_dt"].isoformat() if r["first_dt"] else None,
            "last_dt": r["last_dt"].isoformat() if r["last_dt"] else None,
            "avg_area_sqm": _f(r["avg_area_sqm"]),
            "pct_one": _f(r["pct_one"]),
            "pct_three_plus": _f(r["pct_three_plus"]),
        }
        for r in rows
    ]


def developer_detail(db: Session, developer_id: str) -> dict[str, Any] | None:
    row = (
        db.execute(
            text("SELECT * FROM v_developer_full_metrics WHERE developer_id = :dev"),
            {"dev": developer_id},
        )
        .mappings()
        .first()
    )
    if not row:
        return None
    return {
        "developer_id": row["developer_id"],
        "developer_name": row["developer_name"],
        "jk_count": row["jk_count"],
        "jk_flats_total": row["jk_flats_total"],
        "jk_sqm_total": _f(row["jk_sqm_total"]),
        "jk_ekb": row["jk_ekb"],
        "jk_completed": row["jk_completed"],
        "jk_in_progress": row["jk_in_progress"],
        "jk_escrow": row["jk_escrow"],
        "agg_flats_total": row["agg_flats_total"],
        "agg_one_room": row["agg_one_room"],
        "agg_two_room": row["agg_two_room"],
        "agg_three_room": row["agg_three_room"],
        "agg_four_plus": row["agg_four_plus"],
        "pct_one": _f(row["pct_one"]),
        "pct_three_plus": _f(row["pct_three_plus"]),
        "avg_area_sqm": _f(row["avg_area_sqm"]),
        "sverdl_sqm_th": _f(row["sverdl_sqm"]),
        "sverdl_sold_pct": _f(row["sverdl_sold_pct"]),
        "sverdl_unsold_pct": _f(row["sverdl_unsold_pct"]),
        "sverdl_price_avg": _f(row["sverdl_price_avg"]),
    }


def developer_history(
    db: Session,
    developer_ids: list[str],
    region_code: int = 66,
) -> list[dict[str, Any]]:
    """Per-month sold_perc for one or more developers in the region."""
    rows = (
        db.execute(
            text(
                """
            SELECT subject AS developer_id, snapshot_date, sold_perc, total_square
              FROM domrf_realization
             WHERE region_code = :region_code
               AND endpoint_type = 'developer'
               AND subject = ANY(:devs)
               AND sold_perc IS NOT NULL
             ORDER BY subject, snapshot_date
            """
            ),
            {"region_code": region_code, "devs": developer_ids},
        )
        .mappings()
        .all()
    )
    return [
        {
            "developer_id": r["developer_id"],
            "snapshot_date": r["snapshot_date"].isoformat(),
            "sold_perc": _f(r["sold_perc"]),
            "total_th_sqm": _f(r["total_square"]),
        }
        for r in rows
    ]


def developer_portfolio(db: Session, developer_id: str) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT obj_id, comm_name, addr, region_cd, flat_count,
                   square_living, ready_dt, obj_class, escrow,
                   problem_flag, latitude, longitude, is_ekb
              FROM domrf_kn_objects
             WHERE dev_id = :dev
             ORDER BY ready_dt DESC NULLS LAST
            """
            ),
            {"dev": developer_id},
        )
        .mappings()
        .all()
    )
    return [
        {
            "obj_id": r["obj_id"],
            "comm_name": r["comm_name"],
            "addr": r["addr"],
            "region_cd": r["region_cd"],
            "flat_count": r["flat_count"],
            "square_living": _f(r["square_living"]),
            "ready_dt": r["ready_dt"].isoformat() if r["ready_dt"] else None,
            "obj_class": r["obj_class"],
            "escrow": r["escrow"],
            "problem_flag": r["problem_flag"],
            "lat": _f(r["latitude"]),
            "lon": _f(r["longitude"]),
            "is_ekb": r["is_ekb"],
        }
        for r in rows
    ]


def prinzip_district_distribution(
    db: Session, developer_id: str = "6208_0"
) -> list[dict[str, Any]]:
    """Spatial-join PRINZIP buildings to ЕКБ districts via lat/lon polygons.

    Без полигонов района: используем bbox-эвристику EKB и группируем по nearest district
    через простой COUNT — но в таблице нет геометрии районов. Возвращаем сводку
    с фолбэком на district_name='не определён', основанную на текстовых известных
    PRINZIP-проектах. Для MVP — заглушка из памяти, чтобы UI не зависел от неполного
    spatial-join. TODO: добавить geometry в ekb_districts.
    """
    # Hard-coded from PRINZIP_Strategy_Apr27 — verified mapping.
    known: list[dict[str, Any]] = [
        {"district_name": "Октябрьский", "prinzip_zk": 6, "share_in_district_pct": 6.7},
        {"district_name": "Верх-Исетский", "prinzip_zk": 4, "share_in_district_pct": 2.6},
        {"district_name": "Ленинский", "prinzip_zk": 4, "share_in_district_pct": 1.9},
        {"district_name": "Кировский", "prinzip_zk": 2, "share_in_district_pct": 1.7},
        {"district_name": "Орджоникидзевский", "prinzip_zk": 1, "share_in_district_pct": 0.7},
        {"district_name": "Академический", "prinzip_zk": 0, "share_in_district_pct": 0.0},
        {"district_name": "Чкаловский", "prinzip_zk": 0, "share_in_district_pct": 0.0},
        {"district_name": "Железнодорожный", "prinzip_zk": 0, "share_in_district_pct": 0.0},
    ]
    return known


def prinzip_insights() -> dict[str, Any]:
    """Static text/recommendations from PRINZIP_Strategy_Apr27 (knowledge graph)."""
    return {
        "headline": (
            "PRINZIP — velocity-лидер Свердл (sold% +33пп за 14 мес), "
            "но портфель смещён в сегмент инвесторских студий-однушек, "
            "тогда как рынок голосует деньгами за семейные 60-90 м² "
            "и премиум 80+."
        ),
        "key_gaps": [
            {
                "label": "Средний метраж",
                "prinzip": 38.1,
                "market": 49.0,
                "brusnika": 60.0,
                "forum": 61.0,
                "unit": "м²",
            },
            {
                "label": "Доля 1-к",
                "prinzip": 75.4,
                "market": 52.0,
                "brusnika": 47.0,
                "forum": 44.3,
                "unit": "%",
            },
            {
                "label": "Доля 3-к+",
                "prinzip": 5.4,
                "market": 13.0,
                "brusnika": 18.1,
                "forum": 21.5,
                "unit": "%",
            },
            {
                "label": "sold% Свердл",
                "prinzip": 48.0,
                "market": 29.0,
                "brusnika": 47.0,
                "forum": 54.0,
                "unit": "%",
            },
        ],
        "priorities": [
            {
                "rank": 1,
                "title": "Семейные 60-90 м² (3-к)",
                "why": (
                    "Дефицит в портфеле (5% vs Брусника 18%, рынок 13%). "
                    "Реальные сделки Q3'25-Q1'26: 3-к 60-80 м² = 8% сделок "
                    "при медиане 126 934 ₽/м². Средний чек ≈ 10.5 М ₽ — "
                    "выше текущих 6.15 М CRM."
                ),
            },
            {
                "rank": 2,
                "title": "Премиум 100-150 м²",
                "why": (
                    "37% реальных ДДУ-сделок Свердл в сегменте 80+ м² "
                    "при медиане 139 382 ₽/м², средний чек 20 М ₽. "
                    "Премиум кад.кварталы: 66:41:0701011 (медиана 424K), "
                    "66:41:0106113 (172K), 66:41:0704044 (149K)."
                ),
            },
        ],
        "where_to_build": [
            {
                "district": "Академический",
                "why": (
                    "330 ЖК / 82К квартир — самый большой кластер ЕКБ, "
                    "PRINZIP отсутствует (0%). Семейный сегмент молодых покупателей."
                ),
            },
            {
                "district": "Верх-Исетский (расширение)",
                "why": (
                    "Кад.квартал 66:41:0106113 — ср.метраж 113 м² × 172K ₽/м², "
                    "ниша бизнес 80-130 м²."
                ),
            },
            {
                "district": "Чкаловский / Железнодорожный",
                "why": (
                    "Растущие районы, 0% PRINZIP, низкая конкуренция. "
                    "Тест 60-80 м² без премиума."
                ),
            },
        ],
        "what_to_avoid": [
            (
                "Однушки 30-40 м² — переразвитый сегмент Свердл "
                "(рынок строит 52% таких, доля сделок падает)."
            ),
            (
                "Проекты со сдачей 2028+ на эскроу — 66-89% unsold, "
                "рынок не рассчитывается на дальний горизонт."
            ),
        ],
        "benchmarks": [
            {
                "name": "Брусника",
                "model": ("350 тыс м² × sold 47% × Δ +11пп. 3-к доля 18%, ср. метраж 60 м²."),
            },
            {
                "name": "Холдинг Форум-групп",
                "model": (
                    "113 тыс м² × sold 54% × Δ +21пп лидер velocity. " "3-к доля 21.5%, ср. 61 м²."
                ),
            },
        ],
    }


# ── Per-object drill-in (поверх extras-таблиц из 51_schema_kn_extras.sql) ────


def object_detail(db: Session, obj_id: int) -> dict[str, Any] | None:
    """Базовая инфа объекта из domrf_kn_objects (последний snapshot)."""
    row = (
        db.execute(
            text(
                """
            SELECT obj_id, hobj_id, comm_name, addr, short_addr, region_cd,
                   dev_id, dev_name, floor_min, floor_max, flat_count, square_living,
                   ready_dt, site_status, escrow, obj_class, latitude, longitude,
                   obj_status, snapshot_date
              FROM domrf_kn_objects
             WHERE obj_id = :obj
             ORDER BY snapshot_date DESC
             LIMIT 1
            """
            ),
            {"obj": obj_id},
        )
        .mappings()
        .first()
    )
    if not row:
        return None
    return {
        "obj_id": row["obj_id"],
        "hobj_id": row["hobj_id"],
        "comm_name": row["comm_name"],
        "addr": row["addr"],
        "short_addr": row["short_addr"],
        "region_cd": row["region_cd"],
        "dev_id": row["dev_id"],
        "dev_name": row["dev_name"],
        "floor_min": row["floor_min"],
        "floor_max": row["floor_max"],
        "flat_count": row["flat_count"],
        "square_living": _f(row["square_living"]),
        "ready_dt": row["ready_dt"].isoformat() if row["ready_dt"] else None,
        "site_status": row["site_status"],
        "escrow": row["escrow"],
        "obj_class": row["obj_class"],
        "latitude": _f(row["latitude"]),
        "longitude": _f(row["longitude"]),
        "obj_status": row["obj_status"],
        "snapshot_date": row["snapshot_date"].isoformat() if row["snapshot_date"] else None,
    }


def object_sale_graph(
    db: Session, obj_id: int, type_filter: str | None = None
) -> list[dict[str, Any]]:
    """Time-series продаж per-ЖК. Latest snapshot."""
    where_type = ""
    params: dict[str, Any] = {"obj": obj_id}
    if type_filter:
        where_type = "AND type = :type_filter"
        params["type_filter"] = type_filter
    rows = (
        db.execute(
            text(
                f"""
            SELECT obj_id, report_month, type, realised, contracted,
                   area_sq, price_avg, snapshot_date
              FROM domrf_kn_sale_graph
             WHERE obj_id = :obj
               {where_type}
               AND snapshot_date = (
                   SELECT MAX(snapshot_date) FROM domrf_kn_sale_graph
                    WHERE obj_id = :obj {where_type}
               )
             ORDER BY type, report_month
            """
            ),
            params,
        )
        .mappings()
        .all()
    )
    return [
        {
            "report_month": r["report_month"].isoformat() if r["report_month"] else None,
            "type": r["type"],
            "realised": r["realised"],
            "contracted": r["contracted"],
            "area_sq": _f(r["area_sq"]),
            "price_avg": _f(r["price_avg"]),
        }
        for r in rows
    ]


def object_sales_agg(db: Session, obj_id: int) -> list[dict[str, Any]]:
    """3 строки текущих агрегатов: apartments / nonliv / parking."""
    rows = (
        db.execute(
            text(
                """
            SELECT type, name, total, realised, perc, snapshot_date
              FROM domrf_kn_sales_agg
             WHERE obj_id = :obj
               AND snapshot_date = (
                   SELECT MAX(snapshot_date) FROM domrf_kn_sales_agg WHERE obj_id = :obj
               )
             ORDER BY CASE type
                        WHEN 'apartments' THEN 1
                        WHEN 'nonliv' THEN 2
                        ELSE 3
                      END
            """
            ),
            {"obj": obj_id},
        )
        .mappings()
        .all()
    )
    return [
        {
            "type": r["type"],
            "name": r["name"],
            "total": r["total"],
            "realised": r["realised"],
            "perc": _f(r["perc"]),
        }
        for r in rows
    ]


def object_infrastructure(
    db: Session,
    obj_id: int,
    category: str | None = None,
    max_distance: int = 5000,
) -> list[dict[str, Any]]:
    """POI вокруг ЖК с фильтром по категории и радиусу."""
    where_cat = "AND poi_category = :cat" if category else ""
    params: dict[str, Any] = {"obj": obj_id, "dist": max_distance}
    if category:
        params["cat"] = category
    rows = (
        db.execute(
            text(
                f"""
            SELECT poi_name, poi_subtitle, poi_category, poi_address,
                   poi_lat, poi_lon, distance_m
              FROM domrf_kn_infrastructure
             WHERE obj_id = :obj
               AND distance_m <= :dist
               {where_cat}
               AND snapshot_date = (
                   SELECT MAX(snapshot_date) FROM domrf_kn_infrastructure WHERE obj_id = :obj
               )
             ORDER BY distance_m ASC
            """
            ),
            params,
        )
        .mappings()
        .all()
    )
    return [
        {
            "poi_name": r["poi_name"],
            "poi_subtitle": r["poi_subtitle"],
            "poi_category": r["poi_category"],
            "poi_address": r["poi_address"],
            "lat": _f(r["poi_lat"]),
            "lon": _f(r["poi_lon"]),
            "distance_m": _f(r["distance_m"]),
        }
        for r in rows
    ]


def object_photos(db: Session, obj_id: int, limit: int = 100) -> list[dict[str, Any]]:
    """Фото-метаданные, последние сверху."""
    rows = (
        db.execute(
            text(
                """
            SELECT obj_file_id, ord_num, photo_url, photo_dttm, period_dt,
                   size_bytes, photo_name, ready_desc, build_type, hidden, local_path
              FROM domrf_kn_photos
             WHERE obj_id = :obj AND COALESCE(hidden, FALSE) = FALSE
             ORDER BY period_dt DESC NULLS LAST, ord_num DESC NULLS LAST
             LIMIT :lim
            """
            ),
            {"obj": obj_id, "lim": limit},
        )
        .mappings()
        .all()
    )
    return [
        {
            "obj_file_id": r["obj_file_id"],
            "ord_num": r["ord_num"],
            "photo_url": r["photo_url"],
            "photo_dttm": r["photo_dttm"].isoformat() if r["photo_dttm"] else None,
            "period_dt": r["period_dt"].isoformat() if r["period_dt"] else None,
            "size_bytes": r["size_bytes"],
            "photo_name": r["photo_name"],
            "ready_desc": r["ready_desc"],
            "build_type": r["build_type"],
            "local_path": r["local_path"],
        }
        for r in rows
    ]


def prinzip_funnel_monthly(db: Session, months: int = 24) -> list[dict[str, Any]]:
    """Воронка по месяцам из materialized view."""
    rows = (
        db.execute(
            text(
                """
            SELECT month, source, leads, engaged, converted, conv_pct
              FROM prinzip_funnel_monthly
             WHERE month >= (CURRENT_DATE - (:months || ' months')::interval)::date
             ORDER BY month DESC, leads DESC
            """
            ),
            {"months": months},
        )
        .mappings()
        .all()
    )
    return [
        {
            "month": r["month"].isoformat() if r["month"] else None,
            "source": r["source"],
            "leads": r["leads"],
            "engaged": r["engaged"],
            "converted": r["converted"],
            "conv_pct": _f(r["conv_pct"]),
        }
        for r in rows
    ]


def prinzip_funnel_by_source(db: Session, months: int = 12) -> list[dict[str, Any]]:
    """Агрегаты по source за последние N месяцев."""
    rows = (
        db.execute(
            text(
                """
            SELECT source,
                   SUM(leads) AS leads,
                   SUM(engaged) AS engaged,
                   SUM(converted) AS converted,
                   ROUND(100.0 * SUM(converted) / NULLIF(SUM(leads), 0), 2) AS conv_pct
              FROM prinzip_funnel_monthly
             WHERE month >= (CURRENT_DATE - (:months || ' months')::interval)::date
             GROUP BY source
             ORDER BY leads DESC
            """
            ),
            {"months": months},
        )
        .mappings()
        .all()
    )
    return [
        {
            "source": r["source"],
            "leads": int(r["leads"] or 0),
            "engaged": int(r["engaged"] or 0),
            "converted": int(r["converted"] or 0),
            "conv_pct": _f(r["conv_pct"]),
        }
        for r in rows
    ]


def prinzip_funnel_by_object(db: Session) -> list[dict[str, Any]]:
    """Conversion per ЖК."""
    rows = (
        db.execute(
            text(
                """
            SELECT obj_id, comm_name, leads_count, deals_count, conv_pct,
                   total_revenue, avg_deal_price
              FROM prinzip_funnel_by_object
             ORDER BY total_revenue DESC NULLS LAST
            """
            ),
        )
        .mappings()
        .all()
    )
    return [
        {
            "obj_id": r["obj_id"],
            "comm_name": r["comm_name"],
            "leads_count": r["leads_count"],
            "deals_count": r["deals_count"],
            "conv_pct": _f(r["conv_pct"]),
            "total_revenue": _f(r["total_revenue"]),
            "avg_deal_price": _f(r["avg_deal_price"]),
        }
        for r in rows
    ]


def prinzip_objects_with_velocity(db: Session) -> list[dict[str, Any]]:
    """Список 28 PRINZIP-ЖК с агрегатами + apartments-velocity sparkline data."""
    rows = (
        db.execute(
            text(
                """
            WITH agg AS (
                SELECT obj_id, total, realised, perc
                  FROM domrf_kn_sales_agg
                 WHERE type = 'apartments'
                   AND snapshot_date = (SELECT MAX(snapshot_date) FROM domrf_kn_sales_agg)
            ),
            velocity AS (
                SELECT obj_id,
                       ARRAY_AGG(realised ORDER BY report_month) AS sparkline_realised,
                       ARRAY_AGG(report_month::text ORDER BY report_month) AS months
                  FROM domrf_kn_sale_graph
                 WHERE type = 'apartments'
                   AND snapshot_date = (SELECT MAX(snapshot_date) FROM domrf_kn_sale_graph)
                 GROUP BY obj_id
            )
            SELECT o.obj_id, o.comm_name, o.addr, o.flat_count, o.square_living, o.ready_dt,
                   o.site_status,
                   a.total, a.realised, a.perc,
                   v.sparkline_realised, v.months
              FROM domrf_kn_objects o
              LEFT JOIN agg a ON a.obj_id = o.obj_id
              LEFT JOIN velocity v ON v.obj_id = o.obj_id
             WHERE o.dev_id = '6208_0'
               AND o.snapshot_date = (
                   SELECT MAX(snapshot_date) FROM domrf_kn_objects WHERE dev_id = '6208_0'
               )
             ORDER BY a.total DESC NULLS LAST
            """
            ),
        )
        .mappings()
        .all()
    )
    return [
        {
            "obj_id": r["obj_id"],
            "comm_name": r["comm_name"],
            "addr": r["addr"],
            "flat_count": r["flat_count"],
            "square_living": _f(r["square_living"]),
            "ready_dt": r["ready_dt"].isoformat() if r["ready_dt"] else None,
            "site_status": r["site_status"],
            "total": r["total"],
            "realised": r["realised"],
            "perc": _f(r["perc"]),
            "sparkline_realised": (
                list(r["sparkline_realised"]) if r["sparkline_realised"] else []
            ),
            "months": list(r["months"]) if r["months"] else [],
        }
        for r in rows
    ]