feat: add homepage
This commit is contained in:
314
src/plesna_gerance/database/service.py
Normal file
314
src/plesna_gerance/database/service.py
Normal file
@@ -0,0 +1,314 @@
|
||||
"""Database service for saving and querying extracted data."""
|
||||
|
||||
import json
|
||||
from datetime import date, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from .models import Document, Immeuble, Lot, Locataire, Revenu, Depense
|
||||
|
||||
|
||||
class DuplicateDocumentError(Exception):
|
||||
"""Raised when attempting to save a document that already exists."""
|
||||
|
||||
def __init__(self, reference: str, doc_date: date):
|
||||
self.reference = reference
|
||||
self.date = doc_date
|
||||
super().__init__(
|
||||
f"Document avec reference={reference} et date={doc_date} existe deja"
|
||||
)
|
||||
|
||||
|
||||
class DatabaseService:
|
||||
"""Service for database operations."""
|
||||
|
||||
def __init__(self, session: Session):
|
||||
self.session = session
|
||||
|
||||
def check_duplicate(self, reference: str, doc_date: date) -> Document | None:
|
||||
"""Check if a document with this reference and date already exists."""
|
||||
stmt = select(Document).where(
|
||||
Document.reference == reference, Document.date == doc_date
|
||||
)
|
||||
return self.session.execute(stmt).scalar_one_or_none()
|
||||
|
||||
def get_or_create_immeuble(
|
||||
self, code: str, adresse: str = None, ville: str = None, code_postal: str = None
|
||||
) -> Immeuble:
|
||||
"""Get existing immeuble or create new one."""
|
||||
stmt = select(Immeuble).where(Immeuble.code == code)
|
||||
immeuble = self.session.execute(stmt).scalar_one_or_none()
|
||||
|
||||
if immeuble is None:
|
||||
immeuble = Immeuble(
|
||||
code=code, adresse=adresse, ville=ville, code_postal=code_postal
|
||||
)
|
||||
self.session.add(immeuble)
|
||||
self.session.flush() # Get the ID
|
||||
|
||||
return immeuble
|
||||
|
||||
def get_or_create_lot(
|
||||
self, immeuble_id: int, numero: str, lot_type: str = None
|
||||
) -> Lot:
|
||||
"""Get existing lot or create new one."""
|
||||
stmt = select(Lot).where(Lot.immeuble_id == immeuble_id, Lot.numero == numero)
|
||||
lot = self.session.execute(stmt).scalar_one_or_none()
|
||||
|
||||
if lot is None:
|
||||
lot = Lot(immeuble_id=immeuble_id, numero=numero, type=lot_type)
|
||||
self.session.add(lot)
|
||||
self.session.flush()
|
||||
|
||||
return lot
|
||||
|
||||
def get_or_create_locataire(
|
||||
self, lot_id: int, nom: str, date_debut: date = None
|
||||
) -> Locataire:
|
||||
"""Get existing locataire or create new one."""
|
||||
stmt = select(Locataire).where(Locataire.lot_id == lot_id, Locataire.nom == nom)
|
||||
# If date_debut is provided, include it in the search
|
||||
if date_debut:
|
||||
stmt = stmt.where(Locataire.date_debut == date_debut)
|
||||
else:
|
||||
stmt = stmt.where(Locataire.date_debut.is_(None))
|
||||
|
||||
locataire = self.session.execute(stmt).scalar_one_or_none()
|
||||
|
||||
if locataire is None:
|
||||
locataire = Locataire(lot_id=lot_id, nom=nom, date_debut=date_debut)
|
||||
self.session.add(locataire)
|
||||
self.session.flush()
|
||||
|
||||
return locataire
|
||||
|
||||
def _parse_date(self, date_str: str | None) -> date | None:
|
||||
"""Parse ISO date string to date object."""
|
||||
if not date_str:
|
||||
return None
|
||||
try:
|
||||
return datetime.strptime(date_str, "%Y-%m-%d").date()
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
def save_document(self, data: dict[str, Any], source_file: str = None) -> 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
|
||||
|
||||
Returns:
|
||||
The created Document instance
|
||||
|
||||
Raises:
|
||||
DuplicateDocumentError: If document already exists
|
||||
"""
|
||||
metadata = data.get("metadata", {})
|
||||
doc_info = metadata.get("document", {})
|
||||
immeuble_info = metadata.get("immeuble", {})
|
||||
editeur_info = metadata.get("editeur", {})
|
||||
solde_info = metadata.get("solde", {})
|
||||
|
||||
# Extract key fields
|
||||
reference = doc_info.get("reference", "")
|
||||
doc_date = self._parse_date(doc_info.get("date"))
|
||||
|
||||
if not reference or not doc_date:
|
||||
raise ValueError("Document must have reference and date")
|
||||
|
||||
# Check for duplicates
|
||||
existing = self.check_duplicate(reference, doc_date)
|
||||
if existing:
|
||||
raise DuplicateDocumentError(reference, doc_date)
|
||||
|
||||
# Get or create immeuble
|
||||
immeuble = self.get_or_create_immeuble(
|
||||
code=immeuble_info.get("code", "UNKNOWN"),
|
||||
adresse=immeuble_info.get("adresse"),
|
||||
ville=immeuble_info.get("ville"),
|
||||
code_postal=immeuble_info.get("code_postal"),
|
||||
)
|
||||
|
||||
# Create document
|
||||
document = Document(
|
||||
reference=reference,
|
||||
date=doc_date,
|
||||
type=doc_info.get("type"),
|
||||
source_file=source_file,
|
||||
immeuble_id=immeuble.id,
|
||||
json_data=json.dumps(data, ensure_ascii=False, default=str),
|
||||
editeur_nom=editeur_info.get("nom"),
|
||||
editeur_siret=editeur_info.get("siret"),
|
||||
solde_montant=solde_info.get("montant"),
|
||||
solde_type=solde_info.get("type"),
|
||||
solde_date_arrete=self._parse_date(solde_info.get("date_arrete")),
|
||||
)
|
||||
self.session.add(document)
|
||||
self.session.flush()
|
||||
|
||||
# Process situation_locataires (revenus)
|
||||
for situation in data.get("situation_locataires", []):
|
||||
self._save_situation_locataire(document, immeuble, situation)
|
||||
|
||||
# Process recapitulatif_operations (depenses)
|
||||
for operation in data.get("recapitulatif_operations", []):
|
||||
self._save_operation(document, immeuble, operation)
|
||||
|
||||
self.session.commit()
|
||||
return document
|
||||
|
||||
def _save_situation_locataire(
|
||||
self, document: Document, immeuble: Immeuble, situation: dict
|
||||
) -> None:
|
||||
"""Save locataire situation (revenus lines)."""
|
||||
lot_info = situation.get("lot", {})
|
||||
locataire_info = situation.get("locataire", {})
|
||||
|
||||
# Get or create lot
|
||||
lot = self.get_or_create_lot(
|
||||
immeuble_id=immeuble.id,
|
||||
numero=lot_info.get("numero", "0000"),
|
||||
lot_type=lot_info.get("type"),
|
||||
)
|
||||
|
||||
# Get or create locataire
|
||||
locataire = self.get_or_create_locataire(
|
||||
lot_id=lot.id, nom=locataire_info.get("nom", "INCONNU")
|
||||
)
|
||||
|
||||
# Save each revenue line
|
||||
for ligne in situation.get("lignes", []):
|
||||
periode = ligne.get("periode", {})
|
||||
divers = ligne.get("divers", {})
|
||||
|
||||
revenu = Revenu(
|
||||
document_id=document.id,
|
||||
lot_id=lot.id,
|
||||
locataire_id=locataire.id,
|
||||
type_ligne=ligne.get("type", "loyer"),
|
||||
periode_debut=self._parse_date(periode.get("debut")),
|
||||
periode_fin=self._parse_date(periode.get("fin")),
|
||||
loyers=ligne.get("loyers", 0.0) or 0.0,
|
||||
taxes=ligne.get("taxes", 0.0) or 0.0,
|
||||
provisions=ligne.get("provisions", 0.0) or 0.0,
|
||||
divers_montant=divers.get("montant", 0.0) or 0.0 if divers else 0.0,
|
||||
divers_libelle=divers.get("libelle") if divers else None,
|
||||
total=ligne.get("total", 0.0) or 0.0,
|
||||
regles=ligne.get("regles", 0.0) or 0.0,
|
||||
impayes=ligne.get("impayes", 0.0) or 0.0,
|
||||
)
|
||||
self.session.add(revenu)
|
||||
|
||||
def _save_operation(
|
||||
self, document: Document, immeuble: Immeuble, operation: dict
|
||||
) -> None:
|
||||
"""Save operation (depense)."""
|
||||
montants = operation.get("montants", {})
|
||||
|
||||
# Check if operation is linked to a specific lot
|
||||
lot_id = None
|
||||
lot_numero = operation.get("lot_numero")
|
||||
if lot_numero:
|
||||
lot = self.get_or_create_lot(immeuble_id=immeuble.id, numero=lot_numero)
|
||||
lot_id = lot.id
|
||||
|
||||
depense = Depense(
|
||||
document_id=document.id,
|
||||
immeuble_id=immeuble.id,
|
||||
lot_id=lot_id, # Can be NULL for immeuble-level expenses
|
||||
categorie=operation.get("categorie"),
|
||||
sous_categorie=operation.get("sous_categorie"),
|
||||
fournisseur=operation.get("fournisseur"),
|
||||
description=operation.get("description"),
|
||||
debit=montants.get("debit", 0.0) or 0.0,
|
||||
credit=montants.get("credit", 0.0) or 0.0,
|
||||
tva=montants.get("tva", 0.0) or 0.0,
|
||||
locatif=montants.get("locatif", 0.0) or 0.0,
|
||||
deductible=montants.get("deductible", 0.0) or 0.0,
|
||||
)
|
||||
self.session.add(depense)
|
||||
|
||||
def list_documents(self, limit: int = 100, offset: int = 0) -> list[Document]:
|
||||
"""List all imported documents."""
|
||||
stmt = (
|
||||
select(Document)
|
||||
.order_by(Document.date.desc(), Document.created_at.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
result = self.session.execute(stmt)
|
||||
return list(result.scalars().all())
|
||||
|
||||
def get_document_by_id(self, doc_id: int) -> Document | None:
|
||||
"""Get a document by ID."""
|
||||
return self.session.get(Document, doc_id)
|
||||
|
||||
def get_revenus_summary(
|
||||
self, immeuble_id: int = None, year: int = None
|
||||
) -> list[dict]:
|
||||
"""Get revenue summary grouped by period."""
|
||||
stmt = select(Revenu)
|
||||
|
||||
if immeuble_id:
|
||||
stmt = stmt.join(Lot).where(Lot.immeuble_id == immeuble_id)
|
||||
|
||||
if year:
|
||||
stmt = stmt.where(
|
||||
Revenu.periode_debut >= date(year, 1, 1),
|
||||
Revenu.periode_debut <= date(year, 12, 31),
|
||||
)
|
||||
|
||||
result = self.session.execute(stmt)
|
||||
revenus = result.scalars().all()
|
||||
|
||||
# Aggregate
|
||||
total_loyers = sum(r.loyers for r in revenus)
|
||||
total_regles = sum(r.regles for r in revenus)
|
||||
total_impayes = sum(r.impayes for r in revenus)
|
||||
|
||||
return {
|
||||
"total_loyers": total_loyers,
|
||||
"total_regles": total_regles,
|
||||
"total_impayes": total_impayes,
|
||||
"count": len(revenus),
|
||||
}
|
||||
|
||||
def get_depenses_summary(self, immeuble_id: int = None, year: int = None) -> dict:
|
||||
"""Get expenses summary grouped by category."""
|
||||
stmt = select(Depense)
|
||||
|
||||
if immeuble_id:
|
||||
stmt = stmt.where(Depense.immeuble_id == immeuble_id)
|
||||
|
||||
if year:
|
||||
stmt = stmt.join(Document).where(
|
||||
Document.date >= date(year, 1, 1), Document.date <= date(year, 12, 31)
|
||||
)
|
||||
|
||||
result = self.session.execute(stmt)
|
||||
depenses = result.scalars().all()
|
||||
|
||||
# Aggregate by category
|
||||
by_category = {}
|
||||
for d in depenses:
|
||||
cat = d.categorie or "AUTRE"
|
||||
if cat not in by_category:
|
||||
by_category[cat] = {"debit": 0.0, "credit": 0.0, "count": 0}
|
||||
by_category[cat]["debit"] += d.debit
|
||||
by_category[cat]["credit"] += d.credit
|
||||
by_category[cat]["count"] += 1
|
||||
|
||||
total_debit = sum(d.debit for d in depenses)
|
||||
total_credit = sum(d.credit for d in depenses)
|
||||
|
||||
return {
|
||||
"by_category": by_category,
|
||||
"total_debit": total_debit,
|
||||
"total_credit": total_credit,
|
||||
"count": len(depenses),
|
||||
}
|
||||
Reference in New Issue
Block a user