diff --git a/src/plesna_gerance/api/app.py b/src/plesna_gerance/api/app.py index 66a0144..64733ba 100644 --- a/src/plesna_gerance/api/app.py +++ b/src/plesna_gerance/api/app.py @@ -18,6 +18,7 @@ from .routes import ( documents_router, extraction_router, ia_router, + lot_analyse_router, referentiel_router, revenus_router, tags_router, @@ -54,6 +55,7 @@ app.include_router(analytics_router) app.include_router(dashboard_router) app.include_router(revenus_router) app.include_router(referentiel_router) +app.include_router(lot_analyse_router) if FEATURE_IA: app.include_router(ia_router) app.include_router(config_router) diff --git a/src/plesna_gerance/api/routes/__init__.py b/src/plesna_gerance/api/routes/__init__.py index 51bd5ab..ea9dbdf 100644 --- a/src/plesna_gerance/api/routes/__init__.py +++ b/src/plesna_gerance/api/routes/__init__.py @@ -6,6 +6,7 @@ from .dashboard import router as dashboard_router from .documents import router as documents_router from .extraction import router as extraction_router from .ia import router as ia_router +from .lot_analyse import router as lot_analyse_router from .referentiel import router as referentiel_router from .revenus import router as revenus_router from .tags import router as tags_router @@ -18,6 +19,7 @@ __all__ = [ "dashboard_router", "revenus_router", "referentiel_router", + "lot_analyse_router", "ia_router", "config_router", ] diff --git a/src/plesna_gerance/api/routes/lot_analyse.py b/src/plesna_gerance/api/routes/lot_analyse.py new file mode 100644 index 0000000..7380fea --- /dev/null +++ b/src/plesna_gerance/api/routes/lot_analyse.py @@ -0,0 +1,363 @@ +"""Vue d'un lot : ce que les comptes rendus disent de lui, et rien de plus. + +Les autres pages agrègent le parc ; celle-ci descend à un lot et remet ses +lignes bout à bout — loyers facturés, règlements, interventions — dans l'ordre +où les comptes rendus les ont portées. + +Deux limites sont assumées plutôt que contournées : + +- **les dépenses d'un lot sont celles que le compte rendu lui impute**, pas une + quote-part des charges d'immeuble. Aucune clé de répartition n'existe en base + (ni tantièmes, ni surfaces complètes) : en inventer une donnerait des montants + qu'aucun document ne justifie. Le solde d'un lot n'est donc pas un résultat + net, et `depenses_immeuble_non_reparties` rappelle ce qui reste dehors ; +- **les lignes sont rendues telles qu'extraites**, sans regroupement ni + dédoublonnage. Un acompte et son solde restent deux lignes, parce que le + compte rendu les porte ainsi. +""" + +from datetime import date + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from ...database import get_session +from ...database.models import ( + Depense, + Document, + Immeuble, + Locataire, + Lot, + Revenu, + Tag, +) +from ...services.referentiel import type_effectif +from ...services.revenus_query import ( + TYPE_LIGNE_REPORT, + flux_par, + restant_du_par, + taux_de_recouvrement, +) + +router = APIRouter(prefix="/api", tags=["lots"]) + + +class LotIdentite(BaseModel): + """Qui est ce lot : son rattachement, et ce que sa fiche en dit.""" + + id: int + numero: str + immeuble_id: int + immeuble_code: str | None + immeuble_denomination: str | None + type_effectif: str | None + + # Fiche saisie : `null` tant qu'elle ne l'est pas. La page montre le trou + # plutôt que de le combler. + surface: float | None = None + etage: str | None = None + bat: str | None = None + chauffage: str | None = None + dpe_classe: str | None = None + + #: Noms portés par les comptes rendus. Les dates d'entrée et de sortie ne + #: sont pas extraites : l'ordre n'a pas de sens ici. + locataires: list[str] = [] + + +class LotChiffres(BaseModel): + """Totaux du lot sur tout son historique.""" + + # Recettes. `facture` exclut les reports de solde, `encaisse` les inclut : + # un règlement de vieille dette est bien un encaissement de la période. + facture: float = 0.0 + encaisse: float = 0.0 + loyers: float = 0.0 + provisions: float = 0.0 + #: Dette du lot au dernier compte rendu de son immeuble (photo, non cumulée). + restant_du: float = 0.0 + taux_recouvrement: float = 100.0 + + # Dépenses imputées au lot par le compte rendu. + depenses_debit: float = 0.0 + depenses_credit: float = 0.0 + depenses_deductible: float = 0.0 + depenses_locatif: float = 0.0 + nb_operations: int = 0 + + #: Encaissé moins décaissé sur le lot. Pas un résultat : les charges + #: d'immeuble n'y sont pas (voir le module). + solde: float = 0.0 + + #: Charges de l'immeuble non imputées à un lot, sur toute la période. + #: Affiché comme contexte, jamais ventilé. + depenses_immeuble_non_reparties: float = 0.0 + + +class LigneChronologie(BaseModel): + """Une ligne de compte rendu concernant le lot, recette ou dépense.""" + + date: date + document_id: int + nature: str # "recette" | "depense" + libelle: str + categorie: str | None = None + + # Dépense + fournisseur: str | None = None + tag: str | None = None + + # Recette + type_ligne: str | None = None + periode_debut: date | None = None + periode_fin: date | None = None + regle: float | None = None + impaye: float | None = None + + #: Montant de la ligne : total facturé pour une recette, débit net de crédit + #: pour une dépense. + montant: float = 0.0 + #: Report du solde antérieur : déjà compté par un compte rendu précédent, il + #: s'affiche mais n'entre dans aucun cumul. + est_report: bool = False + + +class Intervenant(BaseModel): + """Une entreprise intervenue sur le lot.""" + + fournisseur: str + nb_interventions: int + #: Somme des lignes du fournisseur, crédit déduit — le total que le détail + #: déplié doit retrouver ligne à ligne. + montant: float + derniere_date: date | None = None + + +class LotAnalyseResponse(BaseModel): + """Fiche complète d'un lot.""" + + identite: LotIdentite + chiffres: LotChiffres + chronologie: list[LigneChronologie] + intervenants: list[Intervenant] + + +def _identite(session: Session, lot: Lot) -> LotIdentite: + """Identité du lot, fiche saisie comprise quand elle existe.""" + immeuble = session.get(Immeuble, lot.immeuble_id) + fiche = lot.caracteristiques + + noms = session.execute( + select(Locataire.nom).where(Locataire.lot_id == lot.id).order_by(Locataire.nom) + ).scalars() + + return LotIdentite( + id=lot.id, + numero=lot.numero, + immeuble_id=lot.immeuble_id, + immeuble_code=immeuble.code if immeuble else None, + immeuble_denomination=immeuble.denomination if immeuble else None, + type_effectif=type_effectif(lot), + surface=fiche.surface if fiche else None, + etage=fiche.etage if fiche else None, + bat=fiche.bat if fiche else None, + chauffage=fiche.chauffage if fiche else None, + dpe_classe=fiche.dpe_classe if fiche else None, + locataires=list(noms), + ) + + +def _chiffres(session: Session, lot: Lot) -> LotChiffres: + """Totaux du lot, en réutilisant les règles flux/stock des revenus. + + Passer par `flux_par` et `restant_du_par` plutôt que de resommer ici : ces + fonctions portent la distinction entre ce qui se cumule et ce qui est une + photo, et la rejouer à la main la ferait diverger de la page Recettes. + """ + flux = flux_par(Revenu.lot_id) + dette = restant_du_par(Revenu.lot_id) + + recettes = session.execute( + select( + flux.c.facture, + flux.c.encaisse, + flux.c.facture_regle, + flux.c.loyers, + flux.c.provisions, + ).where(flux.c.cle == lot.id) + ).first() + + restant_du = session.execute( + select(dette.c.restant_du).where(dette.c.cle == lot.id) + ).scalar() + + depenses = session.execute( + select( + func.coalesce(func.sum(Depense.debit), 0.0), + func.coalesce(func.sum(Depense.credit), 0.0), + func.coalesce(func.sum(Depense.deductible), 0.0), + func.coalesce(func.sum(Depense.locatif), 0.0), + func.count(Depense.id), + ).where(Depense.lot_id == lot.id) + ).one() + debit, credit, deductible, locatif, nb_operations = depenses + + # Charges de l'immeuble laissées hors des lots, pour situer le solde. + commun = session.execute( + select(func.coalesce(func.sum(Depense.debit), 0.0)).where( + Depense.immeuble_id == lot.immeuble_id, Depense.lot_id.is_(None) + ) + ).scalar_one() + + encaisse = (recettes.encaisse if recettes else 0.0) or 0.0 + + return LotChiffres( + facture=(recettes.facture if recettes else 0.0) or 0.0, + encaisse=encaisse, + loyers=(recettes.loyers if recettes else 0.0) or 0.0, + provisions=(recettes.provisions if recettes else 0.0) or 0.0, + restant_du=restant_du or 0.0, + taux_recouvrement=taux_de_recouvrement( + recettes.facture if recettes else None, + recettes.facture_regle if recettes else None, + ), + depenses_debit=debit, + depenses_credit=credit, + depenses_deductible=deductible, + depenses_locatif=locatif, + nb_operations=nb_operations, + solde=round(encaisse - debit + credit, 2), + depenses_immeuble_non_reparties=commun, + ) + + +def _libelle_recette(revenu: Revenu) -> str: + """Ce que la ligne de recette dit d'elle-même. + + Le libellé d'une ligne « divers » porte le motif réel (régularisation, + ordures ménagères…) : le préférer au type générique, qui n'apprendrait rien. + """ + return revenu.divers_libelle or revenu.type_ligne + + +def _chronologie(session: Session, lot: Lot) -> list[LigneChronologie]: + """Recettes et dépenses du lot, remises dans l'ordre des comptes rendus.""" + lignes: list[LigneChronologie] = [] + + revenus = session.execute( + select(Revenu, Document.date) + .join(Document, Revenu.document_id == Document.id) + .where(Revenu.lot_id == lot.id) + ).all() + + for revenu, date_document in revenus: + lignes.append( + LigneChronologie( + date=date_document, + document_id=revenu.document_id, + nature="recette", + libelle=_libelle_recette(revenu), + type_ligne=revenu.type_ligne, + periode_debut=revenu.periode_debut, + periode_fin=revenu.periode_fin, + montant=revenu.total or 0.0, + regle=revenu.regles or 0.0, + impaye=revenu.impayes or 0.0, + est_report=revenu.type_ligne == TYPE_LIGNE_REPORT, + ) + ) + + depenses = session.execute( + select(Depense, Document.date, Tag.nom) + .join(Document, Depense.document_id == Document.id) + .outerjoin(Tag, Tag.id == Depense.tag_id) + .where(Depense.lot_id == lot.id) + ).all() + + for depense, date_document, tag in depenses: + lignes.append( + LigneChronologie( + date=date_document, + document_id=depense.document_id, + nature="depense", + libelle=depense.description or depense.sous_categorie or "", + categorie=depense.sous_categorie, + fournisseur=depense.fournisseur, + tag=tag, + montant=round((depense.debit or 0.0) - (depense.credit or 0.0), 2), + ) + ) + + # Tri en Python : les deux sources sont déjà chargées et un lot en porte + # quelques dizaines de lignes. Les recettes d'abord à date égale, parce + # qu'un compte rendu présente la situation locative avant les opérations. + lignes.sort(key=lambda ligne: (ligne.date, ligne.nature != "recette")) + return lignes + + +def _intervenants(session: Session, lot: Lot) -> list[Intervenant]: + """Entreprises intervenues sur le lot, la plus engagée en tête. + + Simple regroupement sur le fournisseur porté par chaque opération : rien + n'est rapproché ni déduit au-delà de ce que la colonne contient. + + Le montant est net du crédit, comme celui de chaque ligne de la + chronologie : déplier une entreprise doit retrouver son total, pas un autre + chiffre. Un avoir (« Remise état des lieux ») rend d'ailleurs ce montant + négatif, ce qui est bien ce que le compte rendu porte. + """ + montant_net = func.coalesce( + func.sum( + func.coalesce(Depense.debit, 0.0) - func.coalesce(Depense.credit, 0.0) + ), + 0.0, + ) + + rows = session.execute( + select( + Depense.fournisseur, + func.count(Depense.id), + montant_net, + func.max(Document.date), + ) + .join(Document, Depense.document_id == Document.id) + .where(Depense.lot_id == lot.id, Depense.fournisseur.is_not(None)) + .group_by(Depense.fournisseur) + .order_by(montant_net.desc()) + ).all() + + return [ + Intervenant( + fournisseur=fournisseur, + nb_interventions=nb, + montant=montant, + derniere_date=derniere_date, + ) + for fournisseur, nb, montant, derniere_date in rows + ] + + +@router.get("/lots/{lot_id}/analyse", response_model=LotAnalyseResponse) +async def analyser_lot( + lot_id: int, + session: Session = Depends(get_session), +) -> LotAnalyseResponse: + """Tout ce que les comptes rendus portent sur un lot. + + - **lot_id**: ID du lot + + Sans borne de période : l'historique est court et le montrer entier évite + qu'un filtre par défaut cache des opérations sans le dire. + """ + lot = session.get(Lot, lot_id) + if lot is None: + raise HTTPException(status_code=404, detail="Lot introuvable.") + + return LotAnalyseResponse( + identite=_identite(session, lot), + chiffres=_chiffres(session, lot), + chronologie=_chronologie(session, lot), + intervenants=_intervenants(session, lot), + ) diff --git a/tests/test_lot_analyse.py b/tests/test_lot_analyse.py new file mode 100644 index 0000000..9a476a9 --- /dev/null +++ b/tests/test_lot_analyse.py @@ -0,0 +1,225 @@ +"""Tests de la fiche d'un lot. + +Cette page restitue un lot tel que les comptes rendus le portent. Les tests +protègent donc ce qui la rendrait fausse ou trompeuse : cumuler un report de +solde, ventiler des charges d'immeuble qu'aucun document n'attribue, ou +regrouper des lignes que le compte rendu a émises séparément. +""" + +import pytest + +from plesna_gerance.database.models import Immeuble, Lot +from plesna_gerance.database.service import DatabaseService + + +@pytest.fixture +def donnees(db_session, sample_data): + """Deux comptes rendus sur un lot : un loyer réglé, puis un report impayé. + + Le second document facture un loyer resté impayé et reporte le solde du + premier — la configuration exacte où un cumul naïf compterait deux fois la + même dette. + """ + service = DatabaseService(db_session) + service.save_document(data=sample_data) + + suivant = { + **sample_data, + "metadata": { + **sample_data["metadata"], + "document": { + "reference": "REF002", + "date": "2024-02-15", + "type": "COMPTE RENDU DE GESTION", + }, + }, + "situation_locataires": [ + { + "lot": {"numero": "01", "type": "Appartement"}, + "locataire": {"nom": "DUPONT"}, + "lignes": [ + { + "type": "solde_anterieur", + "total": 300.0, + "regles": 0.0, + "impayes": 300.0, + }, + { + "type": "loyer", + "periode": {"debut": "2024-02-01", "fin": "2024-02-29"}, + "loyers": 500.0, + "total": 500.0, + "regles": 200.0, + "impayes": 300.0, + }, + ], + } + ], + "recapitulatif_operations": [ + { + "categorie": "DEPENSES_NON_RECUPERABLES", + "sous_categorie": "Travaux divers", + "fournisseur": "PLOMBERIE", + "description": "S01 ACOMPTE 40% remplacement chaudière", + # Le parser déduit ce numéro du préfixe de la description ; la + # fixture le fournit tel qu'il arrive en base. + "lot_numero": "01", + "montants": {"debit": 400.0, "deductible": 400.0}, + }, + { + "categorie": "DEPENSES_NON_RECUPERABLES", + "sous_categorie": "Travaux divers", + "fournisseur": "PLOMBERIE", + "description": "S01 SOLDE remplacement chaudière", + "lot_numero": "01", + "montants": {"debit": 600.0, "deductible": 600.0}, + }, + { + "categorie": "HONORAIRES_DE_GESTION", + "sous_categorie": "Frais d'expert", + "fournisseur": "EXPERTISE", + "description": "S01 - Remise état des lieux sortie", + "lot_numero": "01", + # Un avoir : la ligne rend au propriétaire au lieu de lui coûter. + "montants": {"debit": 0.0, "credit": 35.4}, + }, + ], + } + service.save_document(data=suivant) + + immeuble = db_session.query(Immeuble).filter(Immeuble.code == "IMM1").one() + lot = db_session.query(Lot).filter(Lot.immeuble_id == immeuble.id).one() + return immeuble, lot + + +def test_lot_inconnu_donne_404(api_client, donnees): + assert api_client.get("/api/lots/999999/analyse").status_code == 404 + + +def test_identite_montre_les_trous_de_la_fiche(api_client, donnees): + """Une caractéristique non saisie reste nulle : la page doit le montrer.""" + _, lot = donnees + + identite = api_client.get(f"/api/lots/{lot.id}/analyse").json()["identite"] + + assert identite["numero"] == "01" + assert identite["immeuble_code"] == "IMM1" + assert identite["type_effectif"] == "Appartement" + assert identite["surface"] is None + assert identite["dpe_classe"] is None + assert identite["locataires"] == ["DUPONT"] + + +def test_le_report_de_solde_ne_gonfle_pas_le_facture(api_client, donnees): + """Le facturé ne retient que les loyers, jamais la dette reportée. + + Deux loyers de 500 € : cumuler en plus le report de 300 € afficherait 1300 € + facturés pour un lot qui n'a jamais rien facturé de tel. + """ + _, lot = donnees + + chiffres = api_client.get(f"/api/lots/{lot.id}/analyse").json()["chiffres"] + + assert chiffres["facture"] == 1000.0 + assert chiffres["encaisse"] == 700.0 + #: Photo du dernier compte rendu (300 de report + 300 de loyer), pas un cumul. + assert chiffres["restant_du"] == 600.0 + assert chiffres["taux_recouvrement"] == 70.0 + + +def test_les_charges_d_immeuble_restent_hors_du_lot(api_client, donnees): + """Le nettoyage de l'immeuble ne doit pas atterrir dans un lot. + + Aucune donnée ne dit quelle part revient à quel lot : la ventiler + inventerait des montants. Elle est exposée à part, comme contexte. + """ + _, lot = donnees + + chiffres = api_client.get(f"/api/lots/{lot.id}/analyse").json()["chiffres"] + + # 400 + 600 de travaux imputés au lot, sans le nettoyage de l'immeuble. + assert chiffres["depenses_debit"] == 1000.0 + assert chiffres["depenses_credit"] == 35.4 + assert chiffres["nb_operations"] == 3 + # Le nettoyage du premier compte rendu, resté sans lot. + assert chiffres["depenses_immeuble_non_reparties"] == 50.0 + # Encaissé (700) moins décaissé (1000), avoir rendu (35,40) : charges + # communes exclues. + assert chiffres["solde"] == -264.6 + + +def test_la_chronologie_garde_les_lignes_telles_qu_extraites(api_client, donnees): + """Acompte et solde restent deux lignes : le compte rendu les porte ainsi.""" + _, lot = donnees + + chronologie = api_client.get(f"/api/lots/{lot.id}/analyse").json()["chronologie"] + + travaux = [ligne for ligne in chronologie if ligne["fournisseur"] == "PLOMBERIE"] + assert len(travaux) == 2 + assert {ligne["montant"] for ligne in travaux} == {400.0, 600.0} + + +def test_la_chronologie_est_ordonnee_et_signale_les_reports(api_client, donnees): + """L'ordre des comptes rendus est l'ordre de lecture de la page.""" + _, lot = donnees + + chronologie = api_client.get(f"/api/lots/{lot.id}/analyse").json()["chronologie"] + + dates = [ligne["date"] for ligne in chronologie] + assert dates == sorted(dates) + assert dates[0] == "2024-01-15" + + reports = [ligne for ligne in chronologie if ligne["est_report"]] + assert len(reports) == 1 + assert reports[0]["montant"] == 300.0 + + +def test_les_intervenants_agregent_le_fournisseur_du_compte_rendu(api_client, donnees): + """Une entreprise, ses interventions et son montant — rien de déduit.""" + _, lot = donnees + + intervenants = api_client.get(f"/api/lots/{lot.id}/analyse").json()["intervenants"] + + par_nom = {ligne["fournisseur"]: ligne for ligne in intervenants} + assert set(par_nom) == {"PLOMBERIE", "EXPERTISE"} + assert par_nom["PLOMBERIE"]["nb_interventions"] == 2 + assert par_nom["PLOMBERIE"]["montant"] == 1000.0 + assert par_nom["PLOMBERIE"]["derniere_date"] == "2024-02-15" + + +def test_un_avoir_rend_le_montant_de_l_intervenant_negatif(api_client, donnees): + """Le crédit est déduit : une remise ne doit pas s'afficher comme un coût.""" + _, lot = donnees + + intervenants = api_client.get(f"/api/lots/{lot.id}/analyse").json()["intervenants"] + + expertise = next( + ligne for ligne in intervenants if ligne["fournisseur"] == "EXPERTISE" + ) + assert expertise["montant"] == -35.4 + # La plus engagée en tête : un avoir se classe donc en dernier. + assert intervenants[-1]["fournisseur"] == "EXPERTISE" + + +def test_le_total_d_un_intervenant_est_celui_de_ses_lignes(api_client, donnees): + """Invariant du dépliage : le détail doit retrouver le total affiché. + + Les deux chiffres viennent de calculs séparés (agrégat SQL d'un côté, lignes + de la chronologie de l'autre) ; les laisser diverger ferait mentir la ligne + qu'on vient d'ouvrir. + """ + _, lot = donnees + + analyse = api_client.get(f"/api/lots/{lot.id}/analyse").json() + + for intervenant in analyse["intervenants"]: + lignes = [ + ligne + for ligne in analyse["chronologie"] + if ligne["fournisseur"] == intervenant["fournisseur"] + ] + assert len(lignes) == intervenant["nb_interventions"] + assert ( + round(sum(ligne["montant"] for ligne in lignes), 2) + == intervenant["montant"] + )