From 6f2530014f799449628ed0b18c7a4d5b55e713ce Mon Sep 17 00:00:00 2001 From: Bertrand Benjamin Date: Sun, 26 Jul 2026 05:17:39 +0200 Subject: [PATCH] fix: conserve l'identite d'un document re-extrait Le remplacement par ID mettait a jour le document en le supprimant puis en le recreant : la ligne changeait d'ID, cassant les liens qui la referencent (URL d'edition, telechargement du PDF) juste apres un balayage de re-extraction. Le document est desormais mis a jour sur place, seules les donnees derivees (revenus, depenses) etant regenerees. Co-Authored-By: Claude Opus 5 --- src/plesna_gerance/database/service.py | 73 ++++++++++++++++---------- tests/test_reextraction.py | 20 ++++++- 2 files changed, 63 insertions(+), 30 deletions(-) diff --git a/src/plesna_gerance/database/service.py b/src/plesna_gerance/database/service.py index 501052a..373d1bc 100644 --- a/src/plesna_gerance/database/service.py +++ b/src/plesna_gerance/database/service.py @@ -171,6 +171,7 @@ class DatabaseService: # Check for duplicates and preserve existing file paths if overwriting existing_pdf_path = None existing_json_path = None + reused: Document | None = None # document mis a jour sur place if replace_document_id is not None: existing = self.session.get(Document, replace_document_id) if existing is None: @@ -180,20 +181,28 @@ class DatabaseService: collision = self.check_duplicate(reference, doc_date) if collision is not None and collision.id != existing.id: raise DuplicateDocumentError(reference, doc_date) - overwrite = True + existing_pdf_path = existing.pdf_path + existing_json_path = existing.json_path + # Les donnees derivees sont regenerees, mais la ligne document est + # conservee : son ID survit a la re-extraction, donc les liens qui + # la referencent (URL d'edition, PDF) restent valides. + existing.revenus.clear() + existing.depenses.clear() + self.session.flush() + reused = existing else: 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) + 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( @@ -223,23 +232,29 @@ class DatabaseService: 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) + # Create document (ou mise a jour sur place lors d'un remplacement) + fields = { + "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, + } + if reused is not None: + document = reused + for key, value in fields.items(): + setattr(document, key, value) + else: + document = Document(**fields) + self.session.add(document) self.session.flush() # Save files to storage diff --git a/tests/test_reextraction.py b/tests/test_reextraction.py index df7a64a..45df0b1 100644 --- a/tests/test_reextraction.py +++ b/tests/test_reextraction.py @@ -6,7 +6,7 @@ from datetime import date import pytest from plesna_gerance.database import storage -from plesna_gerance.database.models import Depense, Document, Tag +from plesna_gerance.database.models import Depense, Document, Revenu, Tag from plesna_gerance.database.service import DatabaseService, DuplicateDocumentError @@ -36,6 +36,24 @@ def test_replace_document_met_a_jour_meme_si_la_reference_change( assert db_session.query(Document).count() == 1 assert updated.reference == "REF001-CORRIGEE" assert updated.date == date(2024, 1, 15) + # L'ID survit au remplacement : les liens vers le document restent valides. + assert updated.id == original_id + + +def test_replace_document_regenere_les_donnees_derivees(db_session, sample_data): + """Revenus et depenses sont remplaces, pas cumules.""" + service = DatabaseService(db_session) + original = service.save_document(data=sample_data) + + new_data = copy.deepcopy(sample_data) + new_data["recapitulatif_operations"][0]["montants"]["debit"] = 90.0 + new_data["situation_locataires"][0]["lignes"][0]["loyers"] = 600.0 + service.save_document(data=new_data, replace_document_id=original.id) + + depenses = db_session.query(Depense).filter_by(document_id=original.id).all() + revenus = db_session.query(Revenu).filter_by(document_id=original.id).all() + assert len(depenses) == 1 and depenses[0].debit == 90.0 + assert len(revenus) == 1 and revenus[0].loyers == 600.0 def test_replace_document_conserve_le_pdf_et_reecrit_le_json(