"""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 ..utils.amounts import parse_amount from ..utils.lots import ( LOT_NUMERO_INCONNU, normalize_extraction_lots, normalize_lot_numero, ) from . import storage from .models import Depense, Document, Immeuble, Locataire, Lot, Revenu, Tag 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. Le numero est normalise sur 2 chiffres pour qu'un meme lot saisi "6", "06" ou "0006" ne soit pas duplique. """ numero = normalize_lot_numero(numero) or LOT_NUMERO_INCONNU 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 @staticmethod def _normalize_amount(value: Any) -> float | None: """Normalise un montant en float (ou None si non interprétable). Garantit qu'un type non numérique issu de l'extraction (string, dict…) n'entre jamais en base dans une colonne Float. """ if isinstance(value, bool): # bool est un int en Python, on l'exclut return None if isinstance(value, (int, float)): return float(value) if isinstance(value, str): return parse_amount(value) return None def save_document( self, data: dict[str, Any], source_file: str = None, pdf_content: bytes = 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 pdf_content: Binary content of the PDF file (optional, for storage) 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 and overwrite=False """ # Uniformiser les numéros de lot avant toute persistance (JSON + tables) normalize_extraction_lots(data) 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 and preserve existing file paths if overwriting existing_pdf_path = None existing_json_path = None existing = self.check_duplicate(reference, doc_date) if existing: if overwrite: # Preserve existing file paths for reuse existing_pdf_path = existing.pdf_path existing_json_path = existing.json_path # Delete existing document (cascade will delete related data) # But DON'T delete files - we'll reuse or update them self.session.delete(existing) self.session.flush() else: 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"), ) # Compute storage paths for PDF and JSON pdf_path = None json_path = None if pdf_content is not None: # New PDF provided - compute new paths pdf_path, json_path = storage.compute_document_paths( reference=reference, doc_date=doc_date, immeuble_adresse=immeuble.adresse, ) # Delete old files if paths are different if existing_pdf_path and existing_pdf_path != pdf_path: storage.delete_document_files(existing_pdf_path, None) if existing_json_path and existing_json_path != json_path: storage.delete_document_files(None, existing_json_path) elif existing_json_path: # No new PDF but we have existing paths - preserve them pdf_path = existing_pdf_path json_path = existing_json_path # 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=self._normalize_amount(solde_info.get("montant")), solde_type=solde_info.get("type"), solde_date_arrete=self._parse_date(solde_info.get("date_arrete")), pdf_path=pdf_path, json_path=json_path, ) self.session.add(document) self.session.flush() # Save files to storage if pdf_content is not None and pdf_path and json_path: # New PDF provided - save both files storage.save_pdf(pdf_content, pdf_path) storage.save_json(data, json_path) elif json_path: # No new PDF but we have a json_path - update the JSON file storage.save_json(data, json_path) # Process situation_locataires (revenus) for situation in data.get("situation_locataires", []): self._save_situation_locataire(document, immeuble, situation) # Process recapitulatif_operations (depenses) # 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 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") or LOT_NUMERO_INCONNU, 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, tag_id: int = None, ) -> 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 tag_id=tag_id, # Tag assigné manuellement 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), } 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)