gendesign/backend/app/api/v1/admin_leads.py

"""Admin endpoint for browsing PRINZIP CRM leads.

GET  /api/v1/admin/leads         — list with filters + pagination
GET  /api/v1/admin/leads/stats   — KPI summary (total/conversion/revenue)

Auth: gendsgn.ru-wide Caddy basic_auth gate (PR #426). App-level X-Admin-Token
header removed 2026-05-23 — двойная auth избыточна для pilot.
PII compliance: only phone_last4 and email_hash exposed; raw phone/email
were never persisted (см. data/sql/52_import_prinzip_crm.py).
"""

from __future__ import annotations

from typing import Annotated, Any, Literal

from fastapi import APIRouter, Depends, Query
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import get_db
from app.services import analytics_queries as q

router = APIRouter()


@router.get("/")
def list_leads(
    db: Annotated[Session, Depends(get_db)],
    status: Annotated[str | None, Query()] = None,
    source: Annotated[str | None, Query()] = None,
    converted: Annotated[bool | None, Query()] = None,
    obj_id: Annotated[int | None, Query()] = None,
    search: Annotated[str | None, Query(min_length=2, max_length=64)] = None,
    date_from: Annotated[str | None, Query(description="YYYY-MM-DD")] = None,
    date_to: Annotated[str | None, Query(description="YYYY-MM-DD")] = None,
    sort: Annotated[Literal["created_desc", "created_asc"], Query()] = "created_desc",
    limit: Annotated[int, Query(ge=1, le=200)] = 50,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> dict[str, Any]:
    where: list[str] = []
    params: dict[str, Any] = {"lim": limit, "off": offset}
    if status:
        where.append("l.status = :status")
        params["status"] = status
    if source:
        where.append("l.source = :source")
        params["source"] = source
    if converted is not None:
        where.append("l.converted = :converted")
        params["converted"] = converted
    if obj_id is not None:
        where.append("l.obj_interest = :obj_id")
        params["obj_id"] = obj_id
    if date_from:
        where.append("l.created_at >= :date_from")
        params["date_from"] = date_from
    if date_to:
        where.append("l.created_at < (CAST(:date_to AS date) + INTERVAL '1 day')")
        params["date_to"] = date_to
    if search:
        where.append("(l.name ILIKE :search OR l.phone_last4 = :exact_search)")
        params["search"] = f"%{search}%"
        params["exact_search"] = search

    where_sql = "WHERE " + " AND ".join(where) if where else ""
    order_sql = "ORDER BY l.created_at DESC" if sort == "created_desc" else "ORDER BY l.created_at"

    rows = (
        db.execute(
            text(
                f"""
            SELECT l.lead_id, l.created_at, l.source, l.channel, l.name,
                   l.phone_last4, l.email_hash, l.obj_interest, l.rooms,
                   l.budget_min, l.budget_max, l.status, l.converted, l.deal_id,
                   o.comm_name AS obj_name,
                   d.deal_price, d.closed_at AS deal_closed_at
              FROM prinzip_leads l
              LEFT JOIN domrf_kn_objects o
                ON o.obj_id = l.obj_interest
               AND o.snapshot_date = (SELECT MAX(snapshot_date) FROM domrf_kn_objects)
              LEFT JOIN prinzip_deals d ON d.deal_id = l.deal_id
            {where_sql}
            {order_sql}
             LIMIT :lim OFFSET :off
            """
            ),
            params,
        )
        .mappings()
        .all()
    )
    total = db.execute(
        text(f"SELECT COUNT(*) FROM prinzip_leads l {where_sql}"),
        {k: v for k, v in params.items() if k not in ("lim", "off")},
    ).scalar_one()

    return {
        "total": int(total or 0),
        "limit": limit,
        "offset": offset,
        "rows": [
            {
                "lead_id": r["lead_id"],
                "created_at": r["created_at"].isoformat() if r["created_at"] else None,
                "source": r["source"],
                "channel": r["channel"],
                "name": r["name"],
                "phone_last4": r["phone_last4"],
                "email_hash": r["email_hash"],
                "obj_interest": r["obj_interest"],
                "obj_name": r["obj_name"],
                "rooms": r["rooms"],
                "budget_min": float(r["budget_min"]) if r["budget_min"] is not None else None,
                "budget_max": float(r["budget_max"]) if r["budget_max"] is not None else None,
                "status": r["status"],
                "converted": r["converted"],
                "deal_id": r["deal_id"],
                "deal_price": float(r["deal_price"]) if r["deal_price"] is not None else None,
                "deal_closed_at": (
                    r["deal_closed_at"].isoformat() if r["deal_closed_at"] else None
                ),
            }
            for r in rows
        ],
    }


@router.get("/stats")
def leads_stats(
    db: Annotated[Session, Depends(get_db)],
    months: Annotated[int, Query(ge=1, le=120)] = 12,
) -> dict[str, Any]:
    """KPI summary за последние N месяцев.

    Суффикс `_window` — за окно `months`, `_total` — за всё время.
    """
    # Почему это важно и почему поля переименованы (#2464): revenue_total и
    # deals_total считались по CTE window_leads, то есть за окно, а суффиксом
    # обещали итог за всё время — рядом с честными leads_total/sources_total.
    # Админка из-за этого показывала карточку «Revenue (всего)» с 12-месячной
    # цифрой. Рационал держим комментарием, а не docstring'ом: docstring уходит
    # в OpenAPI description и дальше в сгенерированные типы фронта.
    row = (
        db.execute(
            text(
                """
            WITH window_leads AS (
                SELECT *
                  FROM prinzip_leads
                 WHERE created_at >= NOW() - make_interval(months => :m)
            )
            SELECT
                (SELECT COUNT(*) FROM prinzip_leads) AS leads_total,
                COUNT(*) FILTER (WHERE TRUE) AS leads_window,
                COUNT(*) FILTER (WHERE converted) AS converted_window,
                ROUND(
                    100.0 * COUNT(*) FILTER (WHERE converted) / NULLIF(COUNT(*), 0),
                    2
                ) AS conv_pct_window,
                (SELECT COUNT(DISTINCT source) FROM prinzip_leads) AS sources_total,
                (
                    SELECT SUM(d.deal_price)
                      FROM prinzip_deals d
                     WHERE d.deal_id IN (
                         SELECT deal_id FROM window_leads WHERE deal_id IS NOT NULL
                     )
                ) AS revenue_window,
                (
                    SELECT COUNT(*)
                      FROM prinzip_deals d
                     WHERE d.deal_id IN (
                         SELECT deal_id FROM window_leads WHERE deal_id IS NOT NULL
                     )
                ) AS deals_window
              FROM window_leads
            """
            ),
            {"m": months},
        )
        .mappings()
        .first()
    )
    if not row:
        return {
            "leads_total": 0,
            "leads_window": 0,
            "converted_window": 0,
            "conv_pct_window": None,
            "sources_total": 0,
            "revenue_window": None,
            "deals_window": 0,
            # window_months раньше отдавался ТОЛЬКО в непустой ветке — формы ответа
            # различались. Оговорка про достижимость: этот `if not row` СЕГОДНЯ не
            # срабатывает — запрос агрегатный и всегда возвращает ровно одну строку
            # (проверено на пустых таблицах: leads_total=0, leads_window=0, строка
            # truthy). То есть правка здесь — согласованность, а не наблюдаемая
            # починка; ветка остаётся защитой на случай смены формы запроса, и
            # расходиться с основной ей нельзя — именно так пропажа поля и возникла.
            "window_months": months,
        }
    return {
        "leads_total": row["leads_total"] or 0,
        "leads_window": row["leads_window"] or 0,
        "converted_window": row["converted_window"] or 0,
        "conv_pct_window": (
            float(row["conv_pct_window"]) if row["conv_pct_window"] is not None else None
        ),
        "sources_total": row["sources_total"] or 0,
        "revenue_window": (
            float(row["revenue_window"]) if row["revenue_window"] is not None else None
        ),
        "deals_window": row["deals_window"] or 0,
        "window_months": months,
    }


@router.get("/funnel/monthly")
def funnel_monthly(
    db: Annotated[Session, Depends(get_db)],
    months: Annotated[int, Query(ge=1, le=120)] = 24,
) -> list[dict[str, Any]]:
    """Воронка по месяцам: leads → engaged → converted (по source)."""
    return q.prinzip_funnel_monthly(db, months=months)


@router.get("/funnel/by-source")
def funnel_by_source(
    db: Annotated[Session, Depends(get_db)],
    months: Annotated[int, Query(ge=1, le=120)] = 12,
) -> list[dict[str, Any]]:
    """Splitting по source: кто конвертит лучше за последние N месяцев."""
    return q.prinzip_funnel_by_source(db, months=months)


@router.get("/funnel/by-object")
def funnel_by_object(
    db: Annotated[Session, Depends(get_db)],
) -> list[dict[str, Any]]:
    """Воронка для каждого ЖК PRINZIP: leads / deals / revenue."""
    return q.prinzip_funnel_by_object(db)


@router.get("/sources")
def list_sources(
    db: Annotated[Session, Depends(get_db)],
) -> list[str]:
    """Distinct list of source values for filter dropdown."""
    rows = db.execute(
        text(
            """
            SELECT source, COUNT(*) AS n
              FROM prinzip_leads
             WHERE source IS NOT NULL AND source <> ''
             GROUP BY source
             ORDER BY n DESC
             LIMIT 50
            """
        )
    ).all()
    return [r[0] for r in rows]