lafrite/pdf_oralia_vibe
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: décrit les logements dans un référentiel saisi à la main

Browse Source 
Les lots n'étaient connus que par l'extraction PDF : un numéro, un type
souvent vide, et rien sur le bien lui-même. Cette table de caractéristiques
(surface, étage, bâtiment, chauffage, DPE, rapprochement impôts) donne au
référentiel une source de vérité indépendante des comptes rendus.

Table séparée de `lots` à dessein : une ré-extraction ne peut alors pas
écraser la saisie, et le désaccord sur le type de lot reste visible au lieu
d'être arbitré en silence. La fiche gagne, le PDF comble les trous.

Échéance du DPE et écart de surface ne sont pas stockés mais calculés : une
colonne dérivée finirait par mentir après une correction.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Bertrand Benjamin
2026-07-26 18:00:42 +02:00
parent ac63922896
commit 5a08b0c7e5
13 changed files with 720 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/plesna_gerance/api/app.py
View File

@@ -18,6 +18,7 @@ from .routes import (
documents_router,
extraction_router,
ia_router,
referentiel_router,
revenus_router,
tags_router,
)
@@ -52,6 +53,7 @@ app.include_router(tags_router)
app.include_router(analytics_router)
app.include_router(dashboard_router)
app.include_router(revenus_router)
app.include_router(referentiel_router)
if FEATURE_IA:
app.include_router(ia_router)
app.include_router(config_router)

2
src/plesna_gerance/api/routes/__init__.py
View File

@@ -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 .referentiel import router as referentiel_router
from .revenus import router as revenus_router
from .tags import router as tags_router
@@ -16,6 +17,7 @@ __all__ = [
"analytics_router",
"dashboard_router",
"revenus_router",
"referentiel_router",
"ia_router",
"config_router",
]

23
src/plesna_gerance/api/routes/analytics.py
View File

@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session
from ...database import get_session
from ...database.models import Depense, Document, Immeuble, Lot, Tag
from ...services.referentiel import TYPE_LOT_EFFECTIF, joindre_fiche
from ..schemas.models import (
CategorySummary,
DepenseDetail,
@@ -77,11 +78,15 @@ async def list_lots(
- **immeuble_id**: ID de l'immeuble pour filtrer (optionnel)
"""
stmt = (
select(Lot, Immeuble.code.label("immeuble_code"))
.join(Immeuble, Lot.immeuble_id == Immeuble.id)
.order_by(Immeuble.code, Lot.numero)
)
stmt = joindre_fiche(
select(
Lot.id,
Lot.numero,
Lot.immeuble_id,
TYPE_LOT_EFFECTIF.label("type"),
Immeuble.code.label("immeuble_code"),
).join(Immeuble, Lot.immeuble_id == Immeuble.id)
).order_by(Immeuble.code, Lot.numero)
if immeuble_id is not None:
stmt = stmt.where(Lot.immeuble_id == immeuble_id)
@@ -91,10 +96,10 @@ async def list_lots(
return [
LotResponse(
id=row.Lot.id,
numero=row.Lot.numero,
type=row.Lot.type,
immeuble_id=row.Lot.immeuble_id,
id=row.id,
numero=row.numero,
type=row.type,
immeuble_id=row.immeuble_id,
immeuble_code=row.immeuble_code,
)
for row in rows

148
src/plesna_gerance/api/routes/referentiel.py Normal file
View File

@@ -0,0 +1,148 @@
"""Référentiel des logements — caractéristiques saisies à la main.
Ces données ne viennent pas des PDF : elles décrivent le bien (surface, étage,
DPE, chauffage) et donnent au référentiel une source de vérité indépendante de
l'extraction.
"""
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from ...database import get_session
from ...database.models import Depense, Immeuble, Lot, LotCaracteristiques, Revenu
from ...services.referentiel import type_effectif
from ...utils.logements import delta_surface, dpe_echeance, type_en_ecart
from ..schemas.models import (
CaracteristiquesBody,
CaracteristiquesResponse,
LotReferentielResponse,
)
router = APIRouter(prefix="/api", tags=["referentiel"])
#: Champs de la fiche, dans l'ordre de saisie. Sert à recopier le corps de la
#: requête vers le modèle sans énumérer les champs à chaque fois.
CHAMPS_FICHE = tuple(CaracteristiquesBody.model_fields)
def _caracteristiques_response(
fiche: LotCaracteristiques | None,
) -> CaracteristiquesResponse | None:
"""Fiche augmentée de ses valeurs dérivées, ou None si elle n'existe pas."""
if fiche is None:
return None
return CaracteristiquesResponse(
**{champ: getattr(fiche, champ) for champ in CHAMPS_FICHE},
dpe_echeance=dpe_echeance(fiche.dpe_date_realisation),
delta_surface=delta_surface(fiche.surface, fiche.surface_impots),
updated_at=fiche.updated_at,
)
def _lot_response(
lot: Lot,
immeuble_code: str | None,
nb_revenus: int = 0,
nb_depenses: int = 0,
) -> LotReferentielResponse:
"""Assemble la ligne de référentiel d'un lot."""
fiche = lot.caracteristiques
return LotReferentielResponse(
id=lot.id,
numero=lot.numero,
immeuble_id=lot.immeuble_id,
immeuble_code=immeuble_code,
type_extrait=lot.type,
type_effectif=type_effectif(lot),
type_ecart=type_en_ecart(lot.type, fiche.type if fiche else None),
caracteristiques=_caracteristiques_response(fiche),
nb_revenus=nb_revenus,
nb_depenses=nb_depenses,
)
@router.get(
"/immeubles/{immeuble_id}/lots/referentiel",
response_model=list[LotReferentielResponse],
)
async def list_lots_referentiel(
immeuble_id: int,
session: Session = Depends(get_session),
) -> list[LotReferentielResponse]:
"""Liste les lots d'un immeuble avec leur fiche de caractéristiques.
- **immeuble_id**: ID de l'immeuble
Les lots sans fiche sont renvoyés avec `caracteristiques` à `null` : le
tableau de saisie doit montrer les lignes vides autant que les remplies.
"""
immeuble = session.get(Immeuble, immeuble_id)
if immeuble is None:
raise HTTPException(status_code=404, detail="Immeuble introuvable.")
# Sous-requêtes corrélées plutôt que des jointures : compter revenus et
# dépenses dans la même jointure multiplierait les lignes entre elles.
nb_revenus = (
select(func.count(Revenu.id))
.where(Revenu.lot_id == Lot.id)
.correlate(Lot)
.scalar_subquery()
)
nb_depenses = (
select(func.count(Depense.id))
.where(Depense.lot_id == Lot.id)
.correlate(Lot)
.scalar_subquery()
)
stmt = (
select(Lot, nb_revenus.label("nb_revenus"), nb_depenses.label("nb_depenses"))
.where(Lot.immeuble_id == immeuble_id)
.order_by(Lot.numero)
)
return [
_lot_response(
row.Lot,
immeuble.code,
nb_revenus=row.nb_revenus or 0,
nb_depenses=row.nb_depenses or 0,
)
for row in session.execute(stmt).all()
]
@router.put("/lots/{lot_id}/caracteristiques", response_model=LotReferentielResponse)
async def upsert_caracteristiques(
lot_id: int,
body: CaracteristiquesBody,
session: Session = Depends(get_session),
) -> LotReferentielResponse:
"""Enregistre la fiche d'un lot, en la créant si elle n'existe pas encore.
- **lot_id**: ID du lot
Le corps décrit la fiche complète : un champ omis ou vidé efface la valeur
précédente, pour qu'une correction dans le tableau ne laisse pas de reste.
"""
lot = session.get(Lot, lot_id)
if lot is None:
raise HTTPException(status_code=404, detail="Lot introuvable.")
fiche = lot.caracteristiques
if fiche is None:
fiche = LotCaracteristiques(lot_id=lot.id)
session.add(fiche)
lot.caracteristiques = fiche
for champ in CHAMPS_FICHE:
setattr(fiche, champ, getattr(body, champ))
session.commit()
session.refresh(lot)
immeuble = session.get(Immeuble, lot.immeuble_id)
return _lot_response(lot, immeuble.code if immeuble else None)

41
src/plesna_gerance/api/routes/revenus.py
View File

@@ -15,6 +15,7 @@ from ...database.models import (
Lot,
Revenu,
)
from ...services.referentiel import TYPE_LOT_EFFECTIF, joindre_fiche
from ...services.revenus_query import (
est_flux,
flux_par,
@@ -390,26 +391,28 @@ async def get_revenus_by_lot(
flux = flux_par(Revenu.lot_id)
dette = restant_du_par(Revenu.lot_id)
stmt = (
select(
Lot.id,
Lot.numero,
Lot.type,
Immeuble.code,
func.max(Locataire.nom).label("locataire_nom"),
flux.c.facture,
flux.c.encaisse,
dette.c.restant_du,
func.max(Document.date).label("derniere_date"),
joindre_fiche(
select(
Lot.id,
Lot.numero,
TYPE_LOT_EFFECTIF.label("type"),
Immeuble.code,
func.max(Locataire.nom).label("locataire_nom"),
flux.c.facture,
flux.c.encaisse,
dette.c.restant_du,
func.max(Document.date).label("derniere_date"),
)
.join(Immeuble, Lot.immeuble_id == Immeuble.id)
.join(Revenu, Revenu.lot_id == Lot.id)
.join(Document, Revenu.document_id == Document.id)
.outerjoin(
Locataire,
and_(Locataire.lot_id == Lot.id, Locataire.date_fin.is_(None)),
)
.outerjoin(flux, flux.c.cle == Lot.id)
.outerjoin(dette, dette.c.cle == Lot.id)
)
.join(Immeuble, Lot.immeuble_id == Immeuble.id)
.join(Revenu, Revenu.lot_id == Lot.id)
.join(Document, Revenu.document_id == Document.id)
.outerjoin(
Locataire,
and_(Locataire.lot_id == Lot.id, Locataire.date_fin.is_(None)),
)
.outerjoin(flux, flux.c.cle == Lot.id)
.outerjoin(dette, dette.c.cle == Lot.id)
.group_by(Lot.id)
.order_by(desc("restant_du"), desc("facture"))
.limit(limit)

6
src/plesna_gerance/api/schemas/__init__.py
View File

@@ -1,10 +1,13 @@
"""Pydantic schemas for API request/response models."""
from .models import (
CaracteristiquesBody,
CaracteristiquesResponse,
DepenseDetail,
DepensesSummary,
DocumentSummary,
ImmeubleResponse,
LotReferentielResponse,
LotResponse,
PredictTagsRequest,
SaveRequest,
@@ -20,4 +23,7 @@ __all__ = [
"DepensesSummary",
"ImmeubleResponse",
"LotResponse",
"CaracteristiquesBody",
"CaracteristiquesResponse",
"LotReferentielResponse",
]

85
src/plesna_gerance/api/schemas/models.py
View File

@@ -1,9 +1,11 @@
"""Pydantic models for API requests and responses."""
from datetime import date
from datetime import date, datetime
from typing import Any
from pydantic import BaseModel
from pydantic import BaseModel, Field, field_validator
from ...utils.logements import DPE_CLASSES
# ============================================================
# Requests
@@ -180,3 +182,82 @@ class DepensesSummary(BaseModel):
by_tag: list[TagSummary]
by_month: list[MonthlySummary]
by_fournisseur: list[FournisseurSummary]
# ============================================================
# Référentiel des logements
# ============================================================
class CaracteristiquesBody(BaseModel):
"""Caractéristiques d'un logement telles que saisies.
Tous les champs sont optionnels : la fiche se remplit progressivement, et
une fiche partielle vaut mieux qu'une fiche refusée.
"""
bat: str | None = None
etage: str | None = None
type: str | None = None
surface: float | None = Field(None, ge=0)
surface_date_diag: date | None = None
chauffage: str | None = None
dpe_classe: str | None = None
dpe_date_realisation: date | None = None
numero_fiscal: str | None = None
surface_impots: float | None = Field(None, ge=0)
note_impots: str | None = None
@field_validator("dpe_classe")
@classmethod
def _classe_connue(cls, value: str | None) -> str | None:
"""Refuse une classe hors A-G, faute de quoi les KPI DPE mentiraient."""
if value is None or value == "":
return None
classe = value.strip().upper()
if classe not in DPE_CLASSES:
raise ValueError(f"Classe DPE inconnue : {value} (attendu A-G)")
return classe
@field_validator(
"bat", "etage", "type", "chauffage", "numero_fiscal", "note_impots"
)
@classmethod
def _texte_vide_vaut_absent(cls, value: str | None) -> str | None:
"""Un champ vide dans le tableau doit effacer la valeur, pas la figer."""
if value is None:
return None
value = value.strip()
return value or None
class CaracteristiquesResponse(CaracteristiquesBody):
"""Caractéristiques saisies, augmentées de leurs valeurs dérivées."""
#: Péremption du DPE, déduite de la date de réalisation (+10 ans).
dpe_echeance: date | None = None
#: Surface impôts moins surface mesurée ; None si une des deux manque.
delta_surface: float | None = None
updated_at: datetime | None = None
class LotReferentielResponse(BaseModel):
"""Un lot et sa fiche, tels qu'affichés dans le tableau du référentiel."""
id: int
numero: str
immeuble_id: int
immeuble_code: str | None
#: Type de lot vu par l'extraction PDF, conservé tel quel.
type_extrait: str | None
#: Type retenu : celui de la fiche s'il existe, sinon celui du PDF.
type_effectif: str | None
#: Vrai quand les deux sources se contredisent (comparaison normalisée).
type_ecart: bool = False
caracteristiques: CaracteristiquesResponse | None = None
#: Rattachements existants : un lot qui en a n'est pas supprimable.
nb_revenus: int = 0
nb_depenses: int = 0

13
src/plesna_gerance/database/__init__.py
View File

@@ -2,7 +2,17 @@
from . import storage
from .connection import get_engine, get_session, get_session_factory, init_db
from .models import Base, Depense, Document, Immeuble, Locataire, Lot, Revenu, Setting
from .models import (
Base,
Depense,
Document,
Immeuble,
Locataire,
Lot,
LotCaracteristiques,
Revenu,
Setting,
)
from .service import DatabaseService, DuplicateDocumentError
__all__ = [
@@ -14,6 +24,7 @@ __all__ = [
"Document",
"Immeuble",
"Lot",
"LotCaracteristiques",
"Locataire",
"Revenu",
"Depense",

47
src/plesna_gerance/database/models.py
View File

@@ -103,11 +103,58 @@ class Lot(Base):
)
revenus = relationship("Revenu", back_populates="lot")
depenses = relationship("Depense", back_populates="lot")
caracteristiques = relationship(
"LotCaracteristiques",
back_populates="lot",
uselist=False,
cascade="all, delete-orphan",
)
def __repr__(self) -> str:
return f"<Lot(numero={self.numero}, type={self.type})>"
class LotCaracteristiques(Base):
"""Caractéristiques d'un logement, saisies à la main.
Table séparée de `lots` à dessein : `lots` porte ce que l'extraction PDF
sait d'un lot, celle-ci ce que le propriétaire en sait. Une ré-extraction ne
peut donc structurellement pas écraser la saisie, et l'écart entre les deux
sources (le type de lot) reste calculable au lieu d'être perdu.
"""
__tablename__ = "lot_caracteristiques"
id = Column(Integer, primary_key=True, autoincrement=True)
lot_id = Column(Integer, ForeignKey("lots.id"), nullable=False, unique=True)
# Description physique
bat = Column(String(50), nullable=True) # "Rue", "Cour"
etage = Column(String(20), nullable=True) # "RC", "1", "SS", "Combles"
type = Column(String(100), nullable=True) # prioritaire sur Lot.type
surface = Column(Float, nullable=True) # m², mesure Oralia
surface_date_diag = Column(Date, nullable=True)
chauffage = Column(String(100), nullable=True)
# Réglementaire
dpe_classe = Column(String(1), nullable=True) # A..G
dpe_date_realisation = Column(Date, nullable=True) # échéance = +10 ans
# Rapprochement avec les impôts
numero_fiscal = Column(String(50), nullable=True) # clé de recherche impots.gouv
surface_impots = Column(Float, nullable=True) # m² déclarés
note_impots = Column(Text, nullable=True)
created_at = Column(DateTime, default=_utcnow)
updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow)
# Relations
lot = relationship("Lot", back_populates="caracteristiques")
def __repr__(self) -> str:
return f"<LotCaracteristiques(lot_id={self.lot_id}, surface={self.surface})>"
class Locataire(Base):
"""Table des locataires avec historique."""

34
src/plesna_gerance/services/referentiel.py Normal file
View File

@@ -0,0 +1,34 @@
"""Arbitrage entre ce que dit le PDF d'un lot et ce que sa fiche en dit.
Le type d'un lot est connu de deux sources : l'extraction PDF, qui le remplit
parfois mal et souvent pas du tout, et la fiche saisie à la main, qui fait foi.
La règle est donc « la fiche gagne, le PDF comble les trous » — et elle vit ici,
en un seul endroit, sous ses deux formes : une expression SQL pour les
agrégats, une fonction Python pour l'ORM. Les faire diverger reviendrait à
afficher deux types différents pour un même lot selon la page consultée.
Le désaccord entre les deux sources n'est jamais résolu en silence : il reste
visible via `type_en_ecart` (voir `utils.logements`).
"""
from sqlalchemy import Select, func
from ..database.models import Lot, LotCaracteristiques
#: Type de lot retenu, en SQL. Requiert la jointure de `joindre_fiche`.
TYPE_LOT_EFFECTIF = func.coalesce(LotCaracteristiques.type, Lot.type)
def joindre_fiche(stmt: Select) -> Select:
"""Ajoute à une requête sur `Lot` la jointure vers sa fiche.
La relation est 1↔1 : la jointure ne multiplie aucune ligne, elle peut donc
s'ajouter à une requête agrégée sans fausser les totaux.
"""
return stmt.outerjoin(LotCaracteristiques, LotCaracteristiques.lot_id == Lot.id)
def type_effectif(lot: Lot) -> str | None:
"""Type de lot retenu, depuis un objet chargé par l'ORM."""
fiche = lot.caracteristiques
return (fiche.type if fiche else None) or lot.type

88
src/plesna_gerance/utils/logements.py Normal file
View File

@@ -0,0 +1,88 @@
"""Valeurs dérivées des caractéristiques d'un logement.
Ni l'échéance du DPE ni l'écart de surface ne sont stockés : ce sont des
conséquences de valeurs saisies, et une colonne dérivée finit toujours par
mentir après une correction. Elles se calculent ici, en un seul endroit, pour
que l'API et les futurs KPI donnent le même résultat.
"""
import re
from datetime import date
#: Durée de validité d'un DPE réalisé après la réforme de 2021 (10 ans).
DPE_VALIDITE_ANNEES = 10
#: Classes possibles d'un DPE, de la plus performante à la moins performante.
DPE_CLASSES = ("A", "B", "C", "D", "E", "F", "G")
def dpe_echeance(date_realisation: date | None) -> date | None:
"""Date de péremption d'un DPE réalisé à `date_realisation`.
Args:
date_realisation: Date de réalisation du diagnostic, ou None
Returns:
Date de fin de validité, ou None si la date de réalisation manque
"""
if date_realisation is None:
return None
annee = date_realisation.year + DPE_VALIDITE_ANNEES
try:
return date_realisation.replace(year=annee)
except ValueError:
# 29 février d'une année bissextile vers une année qui ne l'est pas.
return date(annee, 2, 28)
def delta_surface(surface: float | None, surface_impots: float | None) -> float | None:
"""Écart entre la surface déclarée aux impôts et la surface mesurée.
Signe positif : les impôts retiennent plus de surface que la mesure Oralia.
Args:
surface: Surface mesurée (m²)
surface_impots: Surface déclarée aux impôts (m²)
Returns:
L'écart en m², ou None si une des deux surfaces manque
"""
if surface is None or surface_impots is None:
return None
return round(surface_impots - surface, 2)
def _type_comparable(valeur: str | None) -> str:
"""Écriture normalisée d'un type de lot, pour comparaison seulement."""
if not valeur:
return ""
return re.sub(r"[^a-z0-9]+", " ", valeur.lower()).strip()
def type_en_ecart(type_extrait: str | None, type_saisi: str | None) -> bool:
"""Le PDF et la fiche annoncent-ils deux types de lot différents ?
La comparaison ignore casse et ponctuation : le PDF écrit
"Loc. Commercial" là où une saisie donne "Loc, Commercial" ou
"loc commercial". Signaler ces trois-là comme un désaccord noierait le seul
écart qui compte, celui où les deux sources ne parlent pas du même logement.
Un type absent d'un côté n'est pas un écart : c'est une information qui
manque, pas une contradiction.
Args:
type_extrait: Type de lot vu par l'extraction PDF
type_saisi: Type de lot saisi dans la fiche
Returns:
True si les deux valeurs sont renseignées et se contredisent
"""
extrait = _type_comparable(type_extrait)
saisi = _type_comparable(type_saisi)
if not extrait or not saisi:
return False
return extrait != saisi