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

feat: add depense tagging

Browse Source
This commit is contained in:
Bertrand Benjamin
2026-01-19 05:19:11 +01:00
parent c5e51a7513
commit 0fd6bdaeb3
8 changed files with 711 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,6 +14,7 @@ from .. import __version__
from ..extractor import extract_compte_rendu
from ..database import init_db, get_session, DatabaseService
from ..database.service import DuplicateDocumentError
from ..services.tag_predictor import TagPredictor
app = FastAPI(
title="Plesna Gérance API",
@@ -38,6 +39,14 @@ class SaveRequest(BaseModel):
source_file: str | None = None
data: dict[str, Any]
depenses_tags: list[dict] | None = None # Liste des tags par index de dépense
overwrite: bool = False # Si True, écrase le document existant
class PredictTagsRequest(BaseModel):
"""Request body for predicting tags."""
depenses: list[dict] # Liste des dépenses à prédire
class SaveResponse(BaseModel):
@@ -159,6 +168,8 @@ async def save_document(
- **source_file**: Nom du fichier PDF original (optionnel)
- **data**: Donnees extraites (format identique a la reponse de /api/extract)
- **depenses_tags**: Liste des tags par index de depense (optionnel)
- **overwrite**: Si True, ecrase le document existant (optionnel)
Retourne un message de succes avec l'ID du document cree,
ou une erreur si le document existe deja (doublon).
@@ -166,7 +177,10 @@ async def save_document(
try:
db_service = DatabaseService(session)
document = db_service.save_document(
data=request.data, source_file=request.source_file
data=request.data,
source_file=request.source_file,
depenses_tags=request.depenses_tags,
overwrite=request.overwrite,
)
return SaveResponse(
@@ -328,6 +342,54 @@ async def check_duplicate(
}
@app.get("/api/tags", tags=["tags"])
async def list_tags(
session: Session = Depends(get_session),
) -> list[dict]:
"""Liste tous les tags disponibles pour le tagging des depenses.
Retourne une liste de tags avec leur id et nom.
"""
db_service = DatabaseService(session)
tags = db_service.list_tags()
return [{"id": tag.id, "nom": tag.nom} for tag in tags]
@app.post("/api/predict-tags", tags=["tags"])
async def predict_tags(
request: PredictTagsRequest,
session: Session = Depends(get_session),
) -> dict:
"""Predit les tags pour une liste de depenses basee sur l'historique.
- **depenses**: Liste des depenses a predire (avec fournisseur, sous_categorie, etc.)
Retourne une liste de predictions avec tag suggere, confiance et raison.
"""
try:
predictor = TagPredictor(session)
predictions = predictor.predict_batch(request.depenses)
return {
"predictions": [
{
"index": idx,
"tag_id": pred.tag_id,
"tag_name": pred.tag_name,
"confidence": pred.confidence,
"reason": pred.reason,
}
for idx, pred in enumerate(predictions)
]
}
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Erreur lors de la prediction: {str(e)}"
)
# Mount static files for production (if dist exists)
if FRONTEND_DIST.exists():
# Serve static assets

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

@@ -24,6 +24,22 @@ class Base(DeclarativeBase):
pass
class Tag(Base):
"""Table des tags pour catégoriser les dépenses."""
__tablename__ = "tags"
id = Column(Integer, primary_key=True, autoincrement=True)
nom = Column(String(100), unique=True, nullable=False, index=True)
created_at = Column(DateTime, default=datetime.utcnow)
# Relations
depenses = relationship("Depense", back_populates="tag")
def __repr__(self) -> str:
return f"<Tag(nom={self.nom})>"
class Immeuble(Base):
"""Table des immeubles gérés."""
@@ -201,8 +217,11 @@ class Depense(Base):
lot_id = Column(
Integer, ForeignKey("lots.id"), nullable=True
) # NULL si dépense immeuble
tag_id = Column(
Integer, ForeignKey("tags.id"), nullable=True
) # Tag pour catégorisation manuelle
# Catégorisation
# Catégorisation (ancienne, conservée pour historique)
categorie = Column(String(100), nullable=True) # DEPENSES_LOCATIVES, etc.
sous_categorie = Column(String(255), nullable=True) # Nettoyage immeuble, etc.
fournisseur = Column(String(255), nullable=True)
@@ -221,12 +240,14 @@ class Depense(Base):
Index("ix_depense_document", "document_id"),
Index("ix_depense_immeuble", "immeuble_id"),
Index("ix_depense_categorie", "categorie"),
Index("ix_depense_tag", "tag_id"),
)
# Relations
document = relationship("Document", back_populates="depenses")
immeuble = relationship("Immeuble", back_populates="depenses")
lot = relationship("Lot", back_populates="depenses")
tag = relationship("Tag", back_populates="depenses")
def __repr__(self) -> str:
return f"<Depense(categorie={self.categorie}, debit={self.debit})>"

49
src/plesna_gerance/database/service.py
View File

@@ -8,7 +8,7 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from .models import Document, Immeuble, Lot, Locataire, Revenu, Depense
from .models import Document, Immeuble, Lot, Locataire, Revenu, Depense, Tag
class DuplicateDocumentError(Exception):
@@ -94,19 +94,27 @@ class DatabaseService:
except (ValueError, TypeError):
return None
def save_document(self, data: dict[str, Any], source_file: str = None) -> Document:
def save_document(
self,
data: dict[str, Any],
source_file: str = None,
depenses_tags: list[dict] = None,
overwrite: bool = False,
) -> Document:
"""Save extracted JSON data to database.
Args:
data: The 'data' portion of the extracted JSON (contains metadata,
situation_locataires, recapitulatif_operations)
source_file: Original PDF filename
depenses_tags: List of tags for expenses
overwrite: If True, delete existing document and recreate it
Returns:
The created Document instance
Raises:
DuplicateDocumentError: If document already exists
DuplicateDocumentError: If document already exists and overwrite=False
"""
metadata = data.get("metadata", {})
doc_info = metadata.get("document", {})
@@ -124,7 +132,12 @@ class DatabaseService:
# Check for duplicates
existing = self.check_duplicate(reference, doc_date)
if existing:
raise DuplicateDocumentError(reference, doc_date)
if overwrite:
# Delete existing document (cascade will delete related data)
self.session.delete(existing)
self.session.flush()
else:
raise DuplicateDocumentError(reference, doc_date)
# Get or create immeuble
immeuble = self.get_or_create_immeuble(
@@ -156,8 +169,15 @@ class DatabaseService:
self._save_situation_locataire(document, immeuble, situation)
# Process recapitulatif_operations (depenses)
for operation in data.get("recapitulatif_operations", []):
self._save_operation(document, immeuble, operation)
# Créer un mapping index -> tag_id si des tags sont fournis
tag_mapping = {}
if depenses_tags:
for item in depenses_tags:
tag_mapping[item.get("index")] = item.get("tag_id")
for idx, operation in enumerate(data.get("recapitulatif_operations", [])):
tag_id = tag_mapping.get(idx)
self._save_operation(document, immeuble, operation, tag_id=tag_id)
self.session.commit()
return document
@@ -205,7 +225,11 @@ class DatabaseService:
self.session.add(revenu)
def _save_operation(
self, document: Document, immeuble: Immeuble, operation: dict
self,
document: Document,
immeuble: Immeuble,
operation: dict,
tag_id: int = None,
) -> None:
"""Save operation (depense)."""
montants = operation.get("montants", {})
@@ -221,6 +245,7 @@ class DatabaseService:
document_id=document.id,
immeuble_id=immeuble.id,
lot_id=lot_id, # Can be NULL for immeuble-level expenses
tag_id=tag_id, # Tag assigné manuellement
categorie=operation.get("categorie"),
sous_categorie=operation.get("sous_categorie"),
fournisseur=operation.get("fournisseur"),
@@ -312,3 +337,13 @@ class DatabaseService:
"total_credit": total_credit,
"count": len(depenses),
}
def list_tags(self) -> list[Tag]:
"""List all available tags."""
stmt = select(Tag).order_by(Tag.nom)
result = self.session.execute(stmt)
return list(result.scalars().all())
def get_tag_by_id(self, tag_id: int) -> Tag | None:
"""Get a tag by ID."""
return self.session.get(Tag, tag_id)

76
src/plesna_gerance/scripts/seed_tags.py Normal file
View File

@@ -0,0 +1,76 @@
"""Script pour initialiser les tags prédéfinis dans la base de données."""
from sqlalchemy.exc import IntegrityError
from ..database import init_db, get_session
from ..database.models import Tag
# Liste des tags prédéfinis
PREDEFINED_TAGS = [
"Ascenseur",
"Assurance",
"Contentieux",
"Diagnostics",
"Eau",
"Elec",
"Entretien",
"Hono E/S",
"Hono Gestion",
"Loyer Charge",
"Tel",
"Travaux",
]
def seed_tags():
"""Initialise les tags prédéfinis en base."""
# Initialize database
init_db()
# Get session
session = next(get_session())
try:
created_count = 0
skipped_count = 0
for tag_name in PREDEFINED_TAGS:
try:
# Check if tag already exists
existing = session.query(Tag).filter(Tag.nom == tag_name).first()
if existing:
print(f" ⏭️ Tag '{tag_name}' existe déjà (id={existing.id})")
skipped_count += 1
else:
tag = Tag(nom=tag_name)
session.add(tag)
session.flush()
print(f" ✅ Tag '{tag_name}' créé (id={tag.id})")
created_count += 1
except IntegrityError:
session.rollback()
print(f" ⚠️ Erreur lors de la création du tag '{tag_name}'")
skipped_count += 1
session.commit()
print(f"\n📊 Résumé:")
print(f" - {created_count} tags créés")
print(f" - {skipped_count} tags existants")
print(f" - Total: {len(PREDEFINED_TAGS)} tags")
except Exception as e:
session.rollback()
print(f"❌ Erreur: {e}")
raise
finally:
session.close()
if __name__ == "__main__":
print("🏷️ Initialisation des tags prédéfinis...\n")
seed_tags()
print("\n✨ Terminé!")

1
src/plesna_gerance/services/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Services for business logic."""

165
src/plesna_gerance/services/tag_predictor.py Normal file
View File

@@ -0,0 +1,165 @@
"""Service de prédiction de tags pour les dépenses basé sur l'historique."""
from typing import Optional
from collections import Counter
from sqlalchemy import select, func
from sqlalchemy.orm import Session
from ..database.models import Depense, Tag
class TagPrediction:
"""Représente une prédiction de tag."""
def __init__(
self,
tag_id: Optional[int],
tag_name: Optional[str],
confidence: float,
reason: str,
):
self.tag_id = tag_id
self.tag_name = tag_name
self.confidence = confidence
self.reason = reason
def to_dict(self) -> dict:
"""Convertit en dictionnaire."""
return {
"tag_id": self.tag_id,
"tag_name": self.tag_name,
"confidence": self.confidence,
"reason": self.reason,
}
class TagPredictor:
"""Service de prédiction de tags basé sur l'historique."""
def __init__(self, session: Session):
self.session = session
def predict_for_depense(self, depense_data: dict) -> TagPrediction:
"""Prédit le tag pour une dépense donnée.
Args:
depense_data: Dict contenant 'fournisseur', 'sous_categorie', 'description'
Returns:
TagPrediction avec le tag suggéré et la confiance
"""
fournisseur = (depense_data.get("fournisseur") or "").strip()
sous_categorie = (depense_data.get("sous_categorie") or "").strip()
# Stratégie 1: Recherche par fournisseur exact
if fournisseur:
prediction = self._predict_by_fournisseur(fournisseur)
if prediction:
return prediction
# Stratégie 2: Recherche par sous-catégorie
if sous_categorie:
prediction = self._predict_by_sous_categorie(sous_categorie)
if prediction:
return prediction
# Aucune prédiction trouvée
return TagPrediction(
tag_id=None,
tag_name=None,
confidence=0.0,
reason="Aucun historique trouvé",
)
def _predict_by_fournisseur(self, fournisseur: str) -> Optional[TagPrediction]:
"""Prédit le tag basé sur le fournisseur.
Retourne le tag le plus fréquemment utilisé pour ce fournisseur.
"""
# Requête pour trouver toutes les dépenses avec ce fournisseur et un tag
stmt = (
select(Depense.tag_id, Tag.nom, func.count(Depense.id))
.join(Tag, Depense.tag_id == Tag.id)
.where(
Depense.fournisseur.ilike(f"%{fournisseur}%"),
Depense.tag_id.is_not(None),
)
.group_by(Depense.tag_id, Tag.nom)
.order_by(func.count(Depense.id).desc())
)
result = self.session.execute(stmt).first()
if result:
tag_id, tag_name, count = result
total_stmt = select(func.count(Depense.id)).where(
Depense.fournisseur.ilike(f"%{fournisseur}%")
)
total = self.session.execute(total_stmt).scalar() or 0
confidence = (count / total * 100) if total > 0 else 0
return TagPrediction(
tag_id=tag_id,
tag_name=tag_name,
confidence=round(confidence, 1),
reason=f"Basé sur {count} occurrence(s) pour ce fournisseur",
)
return None
def _predict_by_sous_categorie(
self, sous_categorie: str
) -> Optional[TagPrediction]:
"""Prédit le tag basé sur la sous-catégorie.
Retourne le tag le plus fréquemment utilisé pour cette sous-catégorie.
"""
# Requête pour trouver toutes les dépenses avec cette sous-catégorie et un tag
stmt = (
select(Depense.tag_id, Tag.nom, func.count(Depense.id))
.join(Tag, Depense.tag_id == Tag.id)
.where(
Depense.sous_categorie.ilike(f"%{sous_categorie}%"),
Depense.tag_id.is_not(None),
)
.group_by(Depense.tag_id, Tag.nom)
.order_by(func.count(Depense.id).desc())
)
result = self.session.execute(stmt).first()
if result:
tag_id, tag_name, count = result
total_stmt = select(func.count(Depense.id)).where(
Depense.sous_categorie.ilike(f"%{sous_categorie}%")
)
total = self.session.execute(total_stmt).scalar() or 0
confidence = (count / total * 100) if total > 0 else 0
return TagPrediction(
tag_id=tag_id,
tag_name=tag_name,
confidence=round(confidence, 1),
reason=f"Basé sur {count} occurrence(s) pour cette sous-catégorie",
)
return None
def predict_batch(self, depenses_data: list[dict]) -> list[TagPrediction]:
"""Prédit les tags pour une liste de dépenses.
Args:
depenses_data: Liste de dicts contenant les infos des dépenses
Returns:
Liste de TagPrediction dans le même ordre
"""
predictions = []
for depense in depenses_data:
prediction = self.predict_for_depense(depense)
predictions.append(prediction)
return predictions