fix: corrige le sur-comptage des revenus par immeuble

Les agrégats par immeuble joignaient revenus et locataires à partir du
lot. Comme un lot accumule ses occupants successifs, chaque ligne de
revenu était comptée autant de fois qu'il y a eu de locataires : aucune
erreur levée, juste des montants faux qui dérivent avec l'ancienneté du
parc.

Sur la base réelle (40 lots, 46 locataires), /api/revenus/immeubles et
le bloc by_immeuble de /api/revenus/summary annonçaient :

    total   348 111,95 € au lieu de 332 899,45 €   (+4,6 %)
    réglés  100 697,66 € au lieu de  85 545,16 €   (+17,7 %)
    taux           28,9 % au lieu de       25,7 %

Les montants sont désormais agrégés par immeuble AVANT la jointure des
locataires, qui ne sert plus qu'au dénombrement. Les deux endpoints
partagent ces sous-requêtes et la construction de la réponse.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-26 09:44:11 +02:00
parent 2ac7855cd5
commit 1b319dff4e
3 changed files with 202 additions and 59 deletions

View File

@@ -127,6 +127,65 @@ class RevenusSummaryResponse(BaseModel):
top_impayes: list[RevenuByLocataire]
# ============================================================
# Agregats par immeuble
# ============================================================
def _revenus_par_immeuble():
"""Sous-requete : montants des revenus agreges par immeuble.
Agreger AVANT de joindre les locataires est ce qui garantit la justesse des
totaux : un lot relouee plusieurs fois a plusieurs locataires, et une
jointure directe revenus x locataires compterait chaque revenu autant de
fois qu'il y a eu d'occupants.
"""
return (
select(
Lot.immeuble_id.label("immeuble_id"),
func.sum(Revenu.total).label("total_revenus"),
func.sum(Revenu.regles).label("total_regles"),
func.sum(Revenu.impayes).label("total_impayes"),
)
.join(Revenu, Revenu.lot_id == Lot.id)
.group_by(Lot.immeuble_id)
.subquery()
)
def _effectifs_par_immeuble():
"""Sous-requete : nombre de lots et de locataires par immeuble."""
return (
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()
)
def _immeuble_response(row) -> RevenuByImmeuble:
"""Construit la reponse d'un immeuble a partir d'une ligne agregee."""
revenus = row.total_revenus or 0.0
regles = row.total_regles or 0.0
taux = (regles / revenus * 100) if revenus > 0 else 100.0
return RevenuByImmeuble(
immeuble_id=row.id,
immeuble_code=row.code,
adresse=row.adresse,
ville=row.ville,
nb_lots=row.nb_lots or 0,
nb_locataires=row.nb_locataires or 0,
total_revenus=revenus,
total_regles=regles,
total_impayes=row.total_impayes or 0.0,
taux_recouvrement=round(taux, 1),
)
# ============================================================
# Endpoints
# ============================================================
@@ -241,47 +300,30 @@ async def get_revenus_summary(
)
# ========== By Immeuble ==========
# Seuls les immeubles portant des revenus apparaissent dans ce resume.
montants = _revenus_par_immeuble()
effectifs = _effectifs_par_immeuble()
immeuble_stmt = (
select(
Immeuble.id,
Immeuble.code,
Immeuble.adresse,
Immeuble.ville,
func.count(func.distinct(Lot.id)).label("nb_lots"),
func.count(func.distinct(Locataire.id)).label("nb_locataires"),
func.sum(Revenu.total).label("total_revenus"),
func.sum(Revenu.regles).label("total_regles"),
func.sum(Revenu.impayes).label("total_impayes"),
effectifs.c.nb_lots,
effectifs.c.nb_locataires,
montants.c.total_revenus,
montants.c.total_regles,
montants.c.total_impayes,
)
.join(Lot, Lot.immeuble_id == Immeuble.id)
.join(Revenu, Revenu.lot_id == Lot.id)
.outerjoin(Locataire, Locataire.lot_id == Lot.id)
.group_by(Immeuble.id)
.order_by(desc("total_revenus"))
.join(montants, montants.c.immeuble_id == Immeuble.id)
.outerjoin(effectifs, effectifs.c.immeuble_id == Immeuble.id)
.order_by(desc(montants.c.total_revenus))
)
if immeuble_id:
immeuble_stmt = immeuble_stmt.where(Immeuble.id == immeuble_id)
by_immeuble = []
for row in session.execute(immeuble_stmt):
rev = row.total_revenus or 0.0
reg = row.total_regles or 0.0
taux = (reg / rev * 100) if rev > 0 else 100.0
by_immeuble.append(
RevenuByImmeuble(
immeuble_id=row.id,
immeuble_code=row.code,
adresse=row.adresse,
ville=row.ville,
nb_lots=row.nb_lots or 0,
nb_locataires=row.nb_locataires or 0,
total_revenus=rev,
total_regles=reg,
total_impayes=row.total_impayes or 0.0,
taux_recouvrement=round(taux, 1),
)
)
by_immeuble = [_immeuble_response(row) for row in session.execute(immeuble_stmt)]
# ========== Top impayes by locataire ==========
impayes_stmt = (
@@ -471,43 +513,24 @@ async def get_immeubles_with_revenus(
session: Session = Depends(get_session),
) -> list[RevenuByImmeuble]:
"""Retourne la liste des immeubles avec leurs stats de revenus."""
# Tous les immeubles sont listes, y compris ceux sans aucun revenu.
montants = _revenus_par_immeuble()
effectifs = _effectifs_par_immeuble()
stmt = (
select(
Immeuble.id,
Immeuble.code,
Immeuble.adresse,
Immeuble.ville,
func.count(func.distinct(Lot.id)).label("nb_lots"),
func.count(func.distinct(Locataire.id)).label("nb_locataires"),
func.sum(Revenu.total).label("total_revenus"),
func.sum(Revenu.regles).label("total_regles"),
func.sum(Revenu.impayes).label("total_impayes"),
effectifs.c.nb_lots,
effectifs.c.nb_locataires,
montants.c.total_revenus,
montants.c.total_regles,
montants.c.total_impayes,
)
.outerjoin(Lot, Lot.immeuble_id == Immeuble.id)
.outerjoin(Revenu, Revenu.lot_id == Lot.id)
.outerjoin(Locataire, Locataire.lot_id == Lot.id)
.group_by(Immeuble.id)
.outerjoin(montants, montants.c.immeuble_id == Immeuble.id)
.outerjoin(effectifs, effectifs.c.immeuble_id == Immeuble.id)
.order_by(Immeuble.code)
)
results = []
for row in session.execute(stmt):
rev = row.total_revenus or 0.0
reg = row.total_regles or 0.0
taux = (reg / rev * 100) if rev > 0 else 100.0
results.append(
RevenuByImmeuble(
immeuble_id=row.id,
immeuble_code=row.code,
adresse=row.adresse,
ville=row.ville,
nb_lots=row.nb_lots or 0,
nb_locataires=row.nb_locataires or 0,
total_revenus=rev,
total_regles=reg,
total_impayes=row.total_impayes or 0.0,
taux_recouvrement=round(taux, 1),
)
)
return results
return [_immeuble_response(row) for row in session.execute(stmt)]