pdf_oralia_vibe/src/plesna_gerance/api/routes/dashboard.py

"""Dashboard routes - Vue annuelle des revenus et des depenses."""

from datetime import date

from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel
from sqlalchemy import desc, func, select
from sqlalchemy.orm import Session

from ...database import get_session
from ...database.models import (
    Depense,
    Document,
    Immeuble,
    Locataire,
    Lot,
    Revenu,
)
from ...services.revenus_query import (
    est_flux,
    flux_par,
    restant_du_par,
    taux_de_recouvrement,
)
from ...services.tresorerie import depenses_par, net_reverse, somme_soldes_annonces

router = APIRouter(prefix="/api/dashboard", tags=["dashboard"])


# ============================================================
# Response models
# ============================================================


class PerimetreResponse(BaseModel):
    """De quoi l'annee affichee est faite."""

    nb_comptes_rendus: int
    premiere_date: str | None = None
    derniere_date: str | None = None
    nb_immeubles: int
    #: Lots effectivement factures dans l'annee. La table `lots` en contient
    #: davantage : d'anciennes ecritures de numeros y subsistent sans revenu.
    nb_lots_factures: int


class MoisResponse(BaseModel):
    """Un mois de l'annee, tel qu'il se compare aux autres."""

    mois: str  # "2026-06"
    facture: float
    encaisse: float
    depenses: float
    net: float


class ResumeAnnuelResponse(BaseModel):
    """Revenus et depenses de l'annee civile, avec leur contrepartie annoncee."""

    annee: int
    annees_disponibles: list[int]
    perimetre: PerimetreResponse

    # Flux cumules sur l'annee, report exclu (cf. services/revenus_query)
    facture: float
    encaisse: float
    taux_recouvrement: float
    depenses_debit: float
    depenses_credit: float

    # Ce qui revient au proprietaire, et l'ecart avec les soldes des CR
    net_reverse: float
    soldes_annonces: float
    ecart_soldes: float

    # Stock : photo du dernier compte rendu de l'annee, non cumulable
    restant_du: float
    restant_du_date: str | None = None

    par_mois: list[MoisResponse] = []


class ImmeubleShortcutResponse(BaseModel):
    """Raccourci immeuble pour acces rapide."""

    id: int
    code: str
    adresse: str | None
    ville: str | None
    nb_lots: int
    nb_locataires: int
    total_revenus: float
    total_impayes: float


# ============================================================
# Endpoints
# ============================================================


def _annees_disponibles(session: Session) -> list[int]:
    """Annees ayant au moins un compte rendu, de la plus recente a la plus ancienne."""
    # `select(colonne).distinct()` plutot que `func.distinct(colonne)` : le
    # second perd le type de la colonne et rendrait les dates sous forme de
    # chaines, sans annee a lire.
    dates = session.execute(select(Document.date).distinct()).scalars().all()
    return sorted({jour.year for jour in dates}, reverse=True)


def _annee_par_defaut(session: Session, annee: int | None) -> tuple[int, list[int]]:
    """Annee a afficher et annees proposables.

    Le defaut est l'annee du dernier compte rendu, et non l'annee en cours :
    sans import depuis quelques mois, un accueil cale sur la date du jour se
    viderait alors que la base est pleine.
    """
    annees = _annees_disponibles(session)
    if annee is None:
        annee = annees[0] if annees else date.today().year
    return annee, annees


@router.get("/resume-annuel", response_model=ResumeAnnuelResponse)
async def get_resume_annuel(
    annee: int | None = Query(None, description="Annee civile a afficher"),
    session: Session = Depends(get_session),
) -> ResumeAnnuelResponse:
    """Retourne les revenus et depenses d'une annee civile.

    Les montants cumulables (facture, encaisse, depenses) sont bornes a l'annee ;
    le restant du est lu dans le dernier compte rendu de la periode, car un stock
    ne s'additionne pas dans le temps (cf. services/revenus_query).
    """
    annee, annees = _annee_par_defaut(session, annee)
    debut = date(annee, 1, 1)
    fin = date(annee, 12, 31)

    perimetre = _perimetre(session, debut, fin)

    # ========== Flux de l'annee ==========
    flux = flux_par(Document.immeuble_id, debut, fin)
    totaux = session.execute(
        select(
            func.sum(flux.c.facture).label("facture"),
            func.sum(flux.c.encaisse).label("encaisse"),
            func.sum(flux.c.facture_regle).label("facture_regle"),
        )
    ).one()

    depenses = depenses_par(Document.immeuble_id, debut, fin)
    cumul = session.execute(
        select(
            func.sum(depenses.c.debit).label("debit"),
            func.sum(depenses.c.credit).label("credit"),
        )
    ).one()

    facture = totaux.facture or 0.0
    encaisse = totaux.encaisse or 0.0
    debit = cumul.debit or 0.0
    credit = cumul.credit or 0.0

    net = net_reverse(encaisse, debit, credit)
    annonces = somme_soldes_annonces(session, debut, fin)

    # ========== Restant du (stock) ==========
    stock = restant_du_par(Document.immeuble_id, debut, fin)
    restant_du = session.execute(select(func.sum(stock.c.restant_du))).scalar() or 0.0

    return ResumeAnnuelResponse(
        annee=annee,
        annees_disponibles=annees,
        perimetre=perimetre,
        facture=round(facture, 2),
        encaisse=round(encaisse, 2),
        taux_recouvrement=taux_de_recouvrement(facture, totaux.facture_regle),
        depenses_debit=round(debit, 2),
        depenses_credit=round(credit, 2),
        net_reverse=net,
        soldes_annonces=annonces,
        ecart_soldes=round(net - annonces, 2),
        restant_du=round(restant_du, 2),
        restant_du_date=perimetre.derniere_date,
        par_mois=_par_mois(session, debut, fin),
    )


def _perimetre(session: Session, debut: date, fin: date) -> PerimetreResponse:
    """Ce que couvre l'annee : combien de comptes rendus, sur quoi, jusqu'a quand."""
    documents = session.execute(
        select(
            func.count(Document.id).label("nb"),
            func.min(Document.date).label("premiere"),
            func.max(Document.date).label("derniere"),
            func.count(func.distinct(Document.immeuble_id)).label("nb_immeubles"),
        ).where(Document.date.between(debut, fin))
    ).one()

    # Les lots comptes sont ceux qui ont ete factures dans l'annee : un lot sorti
    # de la gestion ne fait plus partie du perimetre qu'on resume, et la table
    # `lots` retient par ailleurs d'anciennes ecritures de numeros sans revenu.
    nb_lots_factures = (
        session.execute(
            select(func.count(func.distinct(Revenu.lot_id)))
            .join(Document, Revenu.document_id == Document.id)
            .where(Document.date.between(debut, fin))
        ).scalar()
        or 0
    )

    return PerimetreResponse(
        nb_comptes_rendus=documents.nb or 0,
        premiere_date=str(documents.premiere) if documents.premiere else None,
        derniere_date=str(documents.derniere) if documents.derniere else None,
        nb_immeubles=documents.nb_immeubles or 0,
        nb_lots_factures=nb_lots_factures,
    )


def _par_mois(session: Session, debut: date, fin: date) -> list[MoisResponse]:
    """Detail mensuel de l'annee, un point par mois ayant un compte rendu.

    Les mois sans compte rendu sont omis plutot que mis a zero : un mois vide
    est un mois non importe, pas un mois sans loyer, et une barre a zero le
    ferait lire comme une chute d'activite.

    L'encaisse retient toutes les lignes, report compris : regler une vieille
    dette est bien un encaissement du mois. Le facture, lui, exclut le report,
    sans quoi les mois cesseraient d'etre comparables.
    """
    revenus = session.execute(
        select(
            Document.date,
            func.sum(Revenu.total).filter(est_flux()).label("facture"),
            func.sum(Revenu.regles).label("encaisse"),
        )
        .join(Revenu, Revenu.document_id == Document.id)
        .where(Document.date.between(debut, fin))
        .group_by(Document.date)
    ).all()

    depenses = session.execute(
        select(
            Document.date,
            func.sum(Depense.debit).label("debit"),
            func.sum(Depense.credit).label("credit"),
        )
        .join(Depense, Depense.document_id == Document.id)
        .where(Document.date.between(debut, fin))
        .group_by(Document.date)
    ).all()

    mois: dict[str, dict[str, float]] = {}
    for ligne in revenus:
        cumul = mois.setdefault(ligne.date.strftime("%Y-%m"), {})
        cumul["facture"] = cumul.get("facture", 0.0) + (ligne.facture or 0.0)
        cumul["encaisse"] = cumul.get("encaisse", 0.0) + (ligne.encaisse or 0.0)
    for ligne in depenses:
        cumul = mois.setdefault(ligne.date.strftime("%Y-%m"), {})
        cumul["debit"] = cumul.get("debit", 0.0) + (ligne.debit or 0.0)
        cumul["credit"] = cumul.get("credit", 0.0) + (ligne.credit or 0.0)

    return [
        MoisResponse(
            mois=cle,
            facture=round(cumul.get("facture", 0.0), 2),
            encaisse=round(cumul.get("encaisse", 0.0), 2),
            depenses=round(cumul.get("debit", 0.0), 2),
            net=net_reverse(
                cumul.get("encaisse"), cumul.get("debit"), cumul.get("credit")
            ),
        )
        for cle, cumul in sorted(mois.items())
    ]


@router.get("/immeubles-shortcuts", response_model=list[ImmeubleShortcutResponse])
async def get_immeubles_shortcuts(
    annee: int | None = Query(None, description="Annee civile a afficher"),
    limit: int = 5,
    session: Session = Depends(get_session),
) -> list[ImmeubleShortcutResponse]:
    """Retourne les immeubles pour acces rapide avec stats.

    Trie par nombre de documents (plus actifs en premier). Borne sur la meme
    annee que le resume : deux perimetres sur un meme ecran donneraient deux
    montants de revenus sans que rien ne les distingue.

    - **limit**: Nombre maximum d'immeubles (defaut: 5)
    """
    annee, _ = _annee_par_defaut(session, annee)
    debut = date(annee, 1, 1)
    fin = date(annee, 12, 31)

    # Effectifs et activite, puis flux et restant du : trois granularites
    # differentes, jointes plutot que melangees pour ne pas se multiplier.
    effectifs = (
        select(
            Lot.immeuble_id.label("immeuble_id"),
            func.count(func.distinct(Lot.id)).label("nb_lots"),
            func.count(func.distinct(Locataire.id)).label("nb_locataires"),
        )
        .outerjoin(Locataire, Locataire.lot_id == Lot.id)
        .group_by(Lot.immeuble_id)
        .subquery()
    )
    activite = (
        select(
            Document.immeuble_id.label("immeuble_id"),
            func.count(Document.id).label("nb_documents"),
        )
        .where(Document.date.between(debut, fin))
        .group_by(Document.immeuble_id)
        .subquery()
    )
    flux = flux_par(Document.immeuble_id, debut, fin)
    stock = restant_du_par(Document.immeuble_id, debut, fin)

    # Jointure fermee sur l'activite : un immeuble sans compte rendu cette
    # annee-la n'a rien a montrer et sortirait avec des montants vides.
    stmt = (
        select(
            Immeuble,
            effectifs.c.nb_lots,
            effectifs.c.nb_locataires,
            flux.c.facture,
            stock.c.restant_du,
        )
        .join(activite, activite.c.immeuble_id == Immeuble.id)
        .outerjoin(effectifs, effectifs.c.immeuble_id == Immeuble.id)
        .outerjoin(flux, flux.c.cle == Immeuble.id)
        .outerjoin(stock, stock.c.cle == Immeuble.id)
        .order_by(desc(activite.c.nb_documents))
        .limit(limit)
    )

    return [
        ImmeubleShortcutResponse(
            id=row.Immeuble.id,
            code=row.Immeuble.code,
            adresse=row.Immeuble.adresse,
            ville=row.Immeuble.ville,
            nb_lots=row.nb_lots or 0,
            nb_locataires=row.nb_locataires or 0,
            total_revenus=row.facture or 0.0,
            total_impayes=row.restant_du or 0.0,
        )
        for row in session.execute(stmt)
    ]