pdf_oralia_vibe/src/plesna_gerance/services/revenus_query.py

"""Règles d'agrégation des revenus : ce qui se cumule et ce qui ne se cumule pas.

Deux natures de grandeurs cohabitent dans la table ``revenus``, et les
confondre fausse tous les totaux :

- un **flux** est un événement daté (un loyer facturé en mars, un paiement reçu
  en avril). Il s'additionne dans le temps et entre les lots ;
- un **stock** est une photo à un instant (ce qui reste dû au 22/06). Il
  s'additionne entre les lots à une même date, jamais dans le temps :
  additionner deux photos du même solde compte deux fois la même dette.

Chaque compte rendu reporte la dette du précédent dans une ligne
``solde_anterieur``. Cumuler ces lignes sur une période revient donc à recompter
la même créance autant de fois qu'il y a de documents — et rend surtout
impossible d'enregistrer un remboursement : une dette soldée resterait dans le
total à vie.

Attention, **c'est la colonne qui porte la nature, pas la ligne**. Une ligne de
report contient les deux : sa colonne ``total`` est un stock déjà compté le mois
précédent, mais sa colonne ``regles`` est un encaissement bien réel de la
période. Écarter la ligne entière ferait disparaître de l'argent reçu.
"""

from datetime import date

from sqlalchemy import and_, func, select

from ..database.models import Document, Revenu

#: Type des lignes qui reportent le solde du compte rendu précédent.
#:
#: Repris tel quel côté frontend (``utils/totauxLocataire.js``), où il sépare le
#: solde antérieur des loyers de la période. Les deux décrivent la sortie du
#: même parser et doivent bouger ensemble.
TYPE_LIGNE_REPORT = "solde_anterieur"


def est_flux():
    """Condition : la ligne décrit un événement de la période, pas un report."""
    return Revenu.type_ligne != TYPE_LIGNE_REPORT


def _borner(stmt, date_debut: date | None, date_fin: date | None):
    if date_debut is not None:
        stmt = stmt.where(Document.date >= date_debut)
    if date_fin is not None:
        stmt = stmt.where(Document.date <= date_fin)
    return stmt


def derniers_comptes_rendus(
    date_debut: date | None = None, date_fin: date | None = None
):
    """Sous-requête : date du dernier compte rendu de chaque immeuble.

    Le dernier compte rendu est retenu **par immeuble** : avec plusieurs
    immeubles, le dernier document tous immeubles confondus n'en décrirait
    qu'un seul et les autres perdraient leur solde.
    """
    stmt = select(
        Document.immeuble_id.label("immeuble_id"),
        func.max(Document.date).label("date"),
    ).group_by(Document.immeuble_id)
    return _borner(stmt, date_debut, date_fin).subquery()


def restant_du_par(
    cle, date_debut: date | None = None, date_fin: date | None = None
):
    """Sous-requête : restant dû (stock) regroupé par `cle`.

    Seules les lignes du dernier compte rendu de chaque immeuble sont lues :
    c'est la seule photo à jour. Un lot absent de ce compte rendu est sorti de
    la gestion et ne compte plus — les présences observées sont contiguës, une
    absence n'est jamais un simple trou.

    Args:
        cle: colonne de regroupement (``Revenu.lot_id``, ``Document.immeuble_id``…)
        date_debut: borne basse optionnelle sur la date du document
        date_fin: borne haute optionnelle

    Returns:
        Sous-requête exposant `cle` et ``restant_du``.
    """
    derniers = derniers_comptes_rendus(date_debut, date_fin)
    return (
        select(
            cle.label("cle"),
            func.sum(Revenu.impayes).label("restant_du"),
        )
        .join(Document, Revenu.document_id == Document.id)
        .join(
            derniers,
            and_(
                derniers.c.immeuble_id == Document.immeuble_id,
                derniers.c.date == Document.date,
            ),
        )
        .group_by(cle)
        .subquery()
    )


def flux_par(cle, date_debut: date | None = None, date_fin: date | None = None):
    """Sous-requête : montants cumulables (flux) regroupés par `cle`.

    Expose :

    - ``facture`` : ce qui a été facturé sur la période, report exclu ;
    - ``encaisse`` : tout ce qui a été reçu, **y compris** les règlements de
      dettes anciennes portés par les lignes de report ;
    - ``facture_regle`` : la part de ``facture`` qui a été réglée. Sert au taux
      de recouvrement, qui doit comparer un périmètre homogène : rapporter
      ``encaisse`` à ``facture`` ferait dépasser 100 % dès qu'une vieille dette
      est rattrapée.
    """
    flux = est_flux()
    stmt = (
        select(
            cle.label("cle"),
            func.sum(Revenu.loyers).filter(flux).label("loyers"),
            func.sum(Revenu.taxes).filter(flux).label("taxes"),
            func.sum(Revenu.provisions).filter(flux).label("provisions"),
            func.sum(Revenu.total).filter(flux).label("facture"),
            func.sum(Revenu.regles).label("encaisse"),
            func.sum(Revenu.regles).filter(flux).label("facture_regle"),
        )
        .join(Document, Revenu.document_id == Document.id)
        .group_by(cle)
    )
    return _borner(stmt, date_debut, date_fin).subquery()


def taux_de_recouvrement(facture: float | None, facture_regle: float | None) -> float:
    """Part du facturé qui a été réglée, en pourcentage.

    Sans rien de facturé, il n'y a rien à recouvrer : le taux vaut 100 %.
    """
    facture = facture or 0.0
    if facture <= 0:
        return 100.0
    return round((facture_regle or 0.0) / facture * 100, 1)