feat: extraction locataires et opérations par cellules de tableau
Remplace les parseurs texte+regex (fragiles sur l'alignement en colonnes des PDF Oralia mal construits) par une extraction géométrique : reconstruction des lignes visuelles par regroupement vertical tolérant des mots, puis affectation de chaque valeur à sa colonne via les filets du tableau (pdfplumber find_tables). Corrections apportées : - locataires : montant/libellé « divers » dans la bonne colonne (plus le total cumulé du lot), nom de locataire correct (le logo/en-tête hors filets est ignoré), lignes multi-période et pages recollées. - opérations : fournisseur séparé de la description (TOTALENERGIES ≠ DIDIER NETTOYAGE, PPR ≠ BOUVARD), colonne Déductible remplie, Débit/Crédit distingués, fournisseur des honoraires reporté sur le bloc, fragments de description recollés (LATAPY, AUDOUIN). - code lot : gère « S10 - », « S 17 - » (espace) et « S01 SOLDE » (sans tiret). Branchés dans extractor.py avec repli sur les anciens parseurs si un tableau n'a pas de filets détectables. Validés par réconciliation comptable : locataires 124/124 lots, opérations 7/7 PDF et 25/25 catégories, au centime. Ajoute le bouton « Relancer l'extraction » dans l'écran d'édition : re-extrait depuis le PDF stocké et met en évidence les différences avec la version précédente (panneau récapitulatif + anneaux « modifié » sur les cartes). Le diff des opérations s'aligne par contenu (robuste aux changements d'ordre/nombre). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -1,10 +1,16 @@
|
||||
"""Orchestrateur principal pour l'extraction des comptes rendus de gérance."""
|
||||
|
||||
import logging
|
||||
|
||||
from .parsers.locataires import extract_situation_locataires
|
||||
from .parsers.locataires_table import extract_situation_locataires_from_pdf
|
||||
from .parsers.metadata import extract_metadata
|
||||
from .parsers.operations import extract_recapitulatif_operations
|
||||
from .parsers.operations_table import extract_recapitulatif_operations_from_pdf
|
||||
from .parsers.pdf import read_pdf
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def extract_compte_rendu(pdf_path: str) -> dict:
|
||||
"""Extrait toutes les informations d'un PDF de compte rendu de gérance.
|
||||
@@ -26,8 +32,28 @@ def extract_compte_rendu(pdf_path: str) -> dict:
|
||||
"""
|
||||
content = read_pdf(pdf_path)
|
||||
|
||||
# Extraction des locataires par cellules de tableau (géométrique) : robuste aux
|
||||
# colonnes vides et aux lignes mal alignées. Repli sur l'ancien parseur texte si
|
||||
# le tableau n'a pas de filets détectables ou en cas d'erreur inattendue.
|
||||
try:
|
||||
situation = extract_situation_locataires_from_pdf(pdf_path)
|
||||
except Exception:
|
||||
logger.exception("Extraction locataires par cellules échouée, repli sur le parseur texte")
|
||||
situation = []
|
||||
if not situation:
|
||||
situation = extract_situation_locataires(content.text)
|
||||
|
||||
# Opérations par cellules de tableau, même repli sur le parseur texte.
|
||||
try:
|
||||
operations = extract_recapitulatif_operations_from_pdf(pdf_path)
|
||||
except Exception:
|
||||
logger.exception("Extraction opérations par cellules échouée, repli sur le parseur texte")
|
||||
operations = []
|
||||
if not operations:
|
||||
operations = extract_recapitulatif_operations(content.text)
|
||||
|
||||
return {
|
||||
"metadata": extract_metadata(content.text, content.words),
|
||||
"situation_locataires": extract_situation_locataires(content.text),
|
||||
"recapitulatif_operations": extract_recapitulatif_operations(content.text),
|
||||
"situation_locataires": situation,
|
||||
"recapitulatif_operations": operations,
|
||||
}
|
||||
|
||||
322
src/plesna_gerance/parsers/locataires_table.py
Normal file
322
src/plesna_gerance/parsers/locataires_table.py
Normal file
@@ -0,0 +1,322 @@
|
||||
"""Extraction de la situation des locataires par cellules de tableau (géométrique).
|
||||
|
||||
Contrairement à :func:`plesna_gerance.parsers.locataires.extract_situation_locataires`
|
||||
(qui parse le texte mis en page avec des regex de position), cette implémentation
|
||||
reconstruit chaque **ligne visuelle** à partir des coordonnées des mots (regroupement
|
||||
par ``y``) et affecte chaque valeur à **sa colonne** via les filets du tableau
|
||||
(bandes ``x`` déduites de la ligne d'en-tête).
|
||||
|
||||
Avantages sur l'approche texte + regex :
|
||||
|
||||
- robuste aux **cellules vides** (pas de décalage d'indices : un montant est lu
|
||||
dans la bande de sa colonne, pas au n-ième rang de la ligne) ;
|
||||
- robuste aux **lignes mal alignées** (les PDF Oralia n'ont pas de baseline nette :
|
||||
le regroupement par ``y`` tolérant recompose la ligne comme le ferait l'œil) ;
|
||||
- ignore d'office le **texte hors des filets** (logo/adresse/en-tête réimprimé en
|
||||
haut de page), qui polluait la détection du nom de locataire.
|
||||
|
||||
La sortie est **identique** en structure à ``extract_situation_locataires`` pour
|
||||
rester interchangeable (mêmes clés ``lot`` / ``locataire`` / ``lignes`` / ``totaux``).
|
||||
"""
|
||||
|
||||
import re
|
||||
from unicodedata import normalize as _normalize
|
||||
|
||||
import pdfplumber
|
||||
|
||||
from ..utils.amounts import extract_amounts_from_line
|
||||
from ..utils.dates import parse_french_date
|
||||
|
||||
# Tolérance verticale (points PDF) pour regrouper les mots d'une même ligne visuelle.
|
||||
_Y_TOL = 3.0
|
||||
|
||||
_LOT_RE = re.compile(
|
||||
r"^Lot\s+(\d{4})\s+"
|
||||
r"(Loc\.\s*Commercial|Appartement\s+T\d|Studio|Garage|Cave|Parking)"
|
||||
)
|
||||
_PERIODE_RE = re.compile(r"Du\s+\d{2}\.\d{2}\.\d{2}\s+Au\s+(\d{2}\.\d{2}\.\d{2})")
|
||||
|
||||
# En-tête de colonne (normalisé sans accents) -> clé canonique.
|
||||
_HEADER_MAP = {
|
||||
"locataires": "loc",
|
||||
"periode": "periode",
|
||||
"loyers": "loyers",
|
||||
"taxes": "taxes",
|
||||
"provisions": "provisions",
|
||||
"divers": "divlbl", # libellé divers
|
||||
"total": "total",
|
||||
"regles": "regles",
|
||||
"impayes": "impayes",
|
||||
}
|
||||
|
||||
|
||||
def _strip_accents(text: str) -> str:
|
||||
return "".join(c for c in _normalize("NFD", text) if ord(c) < 128).lower()
|
||||
|
||||
|
||||
def _num(cell: str) -> float:
|
||||
"""Dernier montant d'une cellule (les dates DD.MM.YY sont exclues)."""
|
||||
amounts = extract_amounts_from_line(cell or "")
|
||||
return amounts[-1] if amounts else 0.0
|
||||
|
||||
|
||||
def _column_keys(header_cells, page) -> list[str | None]:
|
||||
"""Associe chaque colonne du tableau à une clé canonique via son en-tête.
|
||||
|
||||
La colonne sans en-tête qui suit « Divers » porte le **montant** divers
|
||||
(le libellé étant dans la colonne « Divers »).
|
||||
"""
|
||||
keys: list[str | None] = []
|
||||
prev: str | None = None
|
||||
for cell in header_cells:
|
||||
label = ""
|
||||
if cell is not None:
|
||||
label = _strip_accents((page.crop(cell).extract_text() or "").strip())
|
||||
key = _HEADER_MAP.get(label)
|
||||
if key is None:
|
||||
key = "divamt" if prev == "divlbl" else None
|
||||
keys.append(key)
|
||||
if key is not None:
|
||||
prev = key
|
||||
return keys
|
||||
|
||||
|
||||
def _rows_from_page(page) -> list[dict]:
|
||||
"""Retourne les lignes visuelles du tableau « situation locataires » d'une page.
|
||||
|
||||
Chaque ligne est un ``dict`` {clé_colonne: texte}. Retourne ``[]`` si la page
|
||||
ne contient pas ce tableau.
|
||||
"""
|
||||
# Repérer le tableau « situation locataires » (en-tête commençant par « Locataires »).
|
||||
table = None
|
||||
for candidate in page.find_tables():
|
||||
header = candidate.rows[0].cells
|
||||
if header and header[0] is not None:
|
||||
first = _strip_accents(page.crop(header[0]).extract_text() or "")
|
||||
if "locataires" in first:
|
||||
table = candidate
|
||||
break
|
||||
if table is None:
|
||||
return []
|
||||
|
||||
header = table.rows[0].cells
|
||||
keys = _column_keys(header, page)
|
||||
bands = [(c[0], c[2]) if c is not None else None for c in header]
|
||||
|
||||
def column_of(x_center: float) -> int | None:
|
||||
for i, band in enumerate(bands):
|
||||
if band and band[0] - 1 <= x_center <= band[1] + 1:
|
||||
return i
|
||||
return None
|
||||
|
||||
words = page.crop(table.bbox).extract_words()
|
||||
words.sort(key=lambda w: (round((w["top"] + w["bottom"]) / 2, 1), w["x0"]))
|
||||
|
||||
# Regrouper les mots en lignes visuelles (clustering vertical tolérant).
|
||||
clusters: list[list] = []
|
||||
for word in words:
|
||||
y_center = (word["top"] + word["bottom"]) / 2
|
||||
if clusters and abs(y_center - clusters[-1][0]) <= _Y_TOL:
|
||||
clusters[-1][1].append(word)
|
||||
else:
|
||||
clusters.append([y_center, [word]])
|
||||
|
||||
rows: list[dict] = []
|
||||
for _y, line_words in clusters:
|
||||
cells: dict[str, list[str]] = {}
|
||||
for word in sorted(line_words, key=lambda w: w["x0"]):
|
||||
idx = column_of((word["x0"] + word["x1"]) / 2)
|
||||
key = keys[idx] if idx is not None else None
|
||||
if key is not None:
|
||||
cells.setdefault(key, []).append(word["text"])
|
||||
rows.append({k: " ".join(v) for k, v in cells.items()})
|
||||
return rows
|
||||
|
||||
|
||||
def _new_lot(numero: str, lot_type: str) -> dict:
|
||||
return {
|
||||
"lot": {"numero": numero, "type": lot_type},
|
||||
"locataire": {"nom": ""},
|
||||
"lignes": [],
|
||||
"totaux": {
|
||||
"solde_anterieur": 0.0,
|
||||
"loyers": 0.0,
|
||||
"taxes": 0.0,
|
||||
"provisions": 0.0,
|
||||
"divers": 0.0,
|
||||
"total": 0.0,
|
||||
"regles": 0.0,
|
||||
"impayes": 0.0,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _periode(row: dict) -> dict:
|
||||
if not _PERIODE_RE.search(row.get("periode", "")):
|
||||
return {"debut": None, "fin": None}
|
||||
dates = re.findall(r"\d{2}\.\d{2}\.\d{2}", row.get("periode", ""))
|
||||
return {
|
||||
"debut": parse_french_date(dates[0]) if dates else None,
|
||||
"fin": parse_french_date(dates[1]) if len(dates) > 1 else None,
|
||||
}
|
||||
|
||||
|
||||
def _append_periode_line(lot: dict, row: dict) -> None:
|
||||
"""Ajoute une ligne loyer ou divers à partir d'une ligne de période."""
|
||||
loyers = _num(row.get("loyers", ""))
|
||||
taxes = _num(row.get("taxes", ""))
|
||||
provisions = _num(row.get("provisions", ""))
|
||||
divers_montant = _num(row.get("divamt", ""))
|
||||
divers_libelle = row.get("divlbl") or None
|
||||
total = _num(row.get("total", ""))
|
||||
regles = _num(row.get("regles", ""))
|
||||
impayes = _num(row.get("impayes", ""))
|
||||
|
||||
# Ligne purement « divers » : montant divers présent, colonnes loyer vides.
|
||||
is_divers = (
|
||||
divers_montant != 0.0 and loyers == 0.0 and taxes == 0.0 and provisions == 0.0
|
||||
)
|
||||
|
||||
if is_divers:
|
||||
lot["lignes"].append(
|
||||
{
|
||||
"type": "divers",
|
||||
"periode": _periode(row),
|
||||
"loyers": 0.0,
|
||||
"taxes": 0.0,
|
||||
"provisions": 0.0,
|
||||
"divers": {"montant": divers_montant, "libelle": divers_libelle},
|
||||
"total": total,
|
||||
"regles": regles,
|
||||
"impayes": impayes,
|
||||
}
|
||||
)
|
||||
else:
|
||||
lot["lignes"].append(
|
||||
{
|
||||
"type": "loyer",
|
||||
"periode": _periode(row),
|
||||
"loyers": loyers,
|
||||
"taxes": taxes,
|
||||
"provisions": provisions,
|
||||
"divers": (
|
||||
{"montant": divers_montant, "libelle": divers_libelle}
|
||||
if divers_montant
|
||||
else {"montant": 0.0, "libelle": None}
|
||||
),
|
||||
"total": total,
|
||||
"regles": regles,
|
||||
"impayes": impayes,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _fill_totaux(lot: dict, row: dict) -> None:
|
||||
totaux = lot["totaux"]
|
||||
# Un solde antérieur reporté apparaît dans la colonne « période » de la ligne Totaux.
|
||||
solde_totaux = _num(row.get("periode", ""))
|
||||
if solde_totaux and not totaux["solde_anterieur"]:
|
||||
totaux["solde_anterieur"] = solde_totaux
|
||||
totaux["loyers"] = _num(row.get("loyers", ""))
|
||||
totaux["taxes"] = _num(row.get("taxes", ""))
|
||||
totaux["provisions"] = _num(row.get("provisions", ""))
|
||||
totaux["divers"] = _num(row.get("divamt", ""))
|
||||
totaux["total"] = _num(row.get("total", ""))
|
||||
totaux["regles"] = _num(row.get("regles", ""))
|
||||
totaux["impayes"] = _num(row.get("impayes", ""))
|
||||
|
||||
|
||||
def extract_situation_locataires_from_pdf(pdf_path: str) -> list[dict]:
|
||||
"""Extrait la situation des locataires par cellules de tableau.
|
||||
|
||||
Args:
|
||||
pdf_path: Chemin vers le PDF de compte rendu de gérance.
|
||||
|
||||
Returns:
|
||||
Liste des situations par lot (même structure que
|
||||
:func:`plesna_gerance.parsers.locataires.extract_situation_locataires`).
|
||||
"""
|
||||
situations: list[dict] = []
|
||||
current: dict | None = None
|
||||
|
||||
with pdfplumber.open(pdf_path) as pdf:
|
||||
for page in pdf.pages:
|
||||
if "SITUATION DES LOCATAIRES" not in (page.extract_text() or ""):
|
||||
continue
|
||||
|
||||
for row in _rows_from_page(page):
|
||||
loc = (row.get("loc") or "").strip()
|
||||
periode = (row.get("periode") or "").strip()
|
||||
|
||||
# En-tête de colonnes réimprimé.
|
||||
if _strip_accents(loc) == "locataires":
|
||||
continue
|
||||
|
||||
# Nouveau lot.
|
||||
lot_match = _LOT_RE.match(loc)
|
||||
if lot_match:
|
||||
if current:
|
||||
situations.append(current)
|
||||
current = _new_lot(lot_match.group(1), lot_match.group(2))
|
||||
# La ligne d'en-tête de lot peut porter un premier loyer.
|
||||
if _PERIODE_RE.search(periode):
|
||||
_append_periode_line(current, row)
|
||||
continue
|
||||
|
||||
if current is None:
|
||||
continue
|
||||
|
||||
# Ligne Totaux du lot.
|
||||
if loc.startswith("Totaux"):
|
||||
_fill_totaux(current, row)
|
||||
continue
|
||||
|
||||
# Solde Antérieur (libellé + montant dans la colonne période).
|
||||
if periode.startswith("Solde Antérieur"):
|
||||
montant = _num(periode)
|
||||
current["lignes"].append(
|
||||
{
|
||||
"type": "solde_anterieur",
|
||||
"periode": {"debut": None, "fin": None},
|
||||
"loyers": montant,
|
||||
"taxes": 0.0,
|
||||
"provisions": 0.0,
|
||||
"divers": {"montant": 0.0, "libelle": None},
|
||||
"total": _num(row.get("total", "")) or montant,
|
||||
"regles": _num(row.get("regles", "")),
|
||||
"impayes": _num(row.get("impayes", "")),
|
||||
}
|
||||
)
|
||||
current["totaux"]["solde_anterieur"] = montant
|
||||
continue
|
||||
|
||||
# Rappel de loyer.
|
||||
if loc.startswith("Rappel"):
|
||||
current["lignes"].append(
|
||||
{
|
||||
"type": "rappel_loyer",
|
||||
"periode": _periode(row),
|
||||
"loyers": _num(row.get("loyers", "")),
|
||||
"taxes": 0.0,
|
||||
"provisions": 0.0,
|
||||
"divers": {"montant": 0.0, "libelle": None},
|
||||
"total": _num(row.get("total", "")),
|
||||
"regles": _num(row.get("regles", "")),
|
||||
"impayes": _num(row.get("impayes", "")),
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
# Ligne de période (loyer ou divers).
|
||||
if _PERIODE_RE.search(periode):
|
||||
_append_periode_line(current, row)
|
||||
continue
|
||||
|
||||
# Nom du locataire (cellule « Locataires » seule, nom pas encore trouvé).
|
||||
if loc and not current["locataire"]["nom"]:
|
||||
current["locataire"]["nom"] = loc
|
||||
|
||||
if current:
|
||||
situations.append(current)
|
||||
|
||||
return situations
|
||||
@@ -26,8 +26,10 @@ def _extract_lot_code_from_description(description: str) -> str | None:
|
||||
if not description:
|
||||
return None
|
||||
|
||||
# Pattern: lettre majuscule + 1-2 chiffres, suivi de " - " ou fin de mot
|
||||
match = re.search(r"\b[A-Z](\d{1,2})\s*-", description)
|
||||
# Pattern: lettre majuscule + (espace optionnelle) + 1-2 chiffres, terminé par
|
||||
# un espace, un tiret ou la fin. Gère "S10 - ...", "S 17 - ..." (espace dans le
|
||||
# code) et "S01 SOLDE ..." (code lot non suivi d'un tiret).
|
||||
match = re.search(r"\b[A-Z]\s*(\d{1,2})(?=[\s-]|$)", description)
|
||||
if match:
|
||||
lot_num = match.group(1)
|
||||
# Formater sur 4 chiffres (ex: "6" -> "0006", "12" -> "0012")
|
||||
|
||||
242
src/plesna_gerance/parsers/operations_table.py
Normal file
242
src/plesna_gerance/parsers/operations_table.py
Normal file
@@ -0,0 +1,242 @@
|
||||
"""Extraction du récapitulatif des opérations par cellules de tableau (géométrique).
|
||||
|
||||
Même principe que :mod:`plesna_gerance.parsers.locataires_table` : on reconstruit
|
||||
chaque **ligne visuelle** (regroupement des mots par ``y``) et on affecte chaque
|
||||
valeur à **sa colonne** via les filets du tableau (bandes ``x``).
|
||||
|
||||
Gains sur l'ancien parseur texte (:mod:`plesna_gerance.parsers.operations`) :
|
||||
|
||||
- le **fournisseur** (colonne de gauche, en MAJUSCULES) est proprement séparé de la
|
||||
**description** (colonne du milieu) — plus de report erroné (ex. TOTALENERGIES
|
||||
étiqueté DIDIER NETTOYAGE) ;
|
||||
- chaque **montant** tombe dans sa colonne (Débit / Crédit / TVA / Locatif /
|
||||
Déductible), sans décalage dû aux cellules vides.
|
||||
|
||||
La sortie est identique en structure à ``extract_recapitulatif_operations``.
|
||||
"""
|
||||
|
||||
from unicodedata import normalize as _normalize
|
||||
|
||||
import pdfplumber
|
||||
|
||||
from ..utils.amounts import extract_amounts_from_line
|
||||
from .operations import _extract_lot_code_from_description
|
||||
|
||||
_Y_TOL = 3.0
|
||||
|
||||
# Libellé PDF (début de cellule) -> catégorie normalisée.
|
||||
_CAT_KEYWORDS = {
|
||||
"DEPENSES LOCATIVES": "DEPENSES_LOCATIVES",
|
||||
"DEPENSES DEDUCTIBLES": "DEPENSES_DEDUCTIBLES",
|
||||
"DEPENSES NON RECUPERABLES": "DEPENSES_NON_RECUPERABLES",
|
||||
"DEPENSES RECUPERABLES PAR LOT": "DEPENSES_RECUPERABLES",
|
||||
"HONORAIRES DE GESTION": "HONORAIRES_DE_GESTION",
|
||||
"DIVERS": "DIVERS",
|
||||
}
|
||||
|
||||
_AMOUNT_KEYS = ("debit", "credit", "tva", "locatif", "deductible")
|
||||
|
||||
|
||||
def _strip_accents(text: str) -> str:
|
||||
return "".join(c for c in _normalize("NFD", text) if ord(c) < 128).lower()
|
||||
|
||||
|
||||
def _num(cell: str) -> float | None:
|
||||
"""Montant d'une cellule, ou ``None`` si la cellule ne contient pas de montant."""
|
||||
amounts = extract_amounts_from_line(cell or "")
|
||||
return amounts[-1] if amounts else None
|
||||
|
||||
|
||||
def _match_category(text: str) -> str | None:
|
||||
up = (text or "").upper().strip()
|
||||
for keyword, normalized in _CAT_KEYWORDS.items():
|
||||
if up.startswith(keyword):
|
||||
return normalized
|
||||
return None
|
||||
|
||||
|
||||
def _is_fournisseur(text: str) -> bool:
|
||||
"""Une cellule de gauche est un fournisseur si elle est en MAJUSCULES.
|
||||
|
||||
Distingue « BOUVARD ENTREPRISE » (fournisseur) de « Travaux divers »
|
||||
(sous-catégorie en casse mixte).
|
||||
"""
|
||||
letters = [c for c in text if c.isalpha()]
|
||||
return bool(letters) and all(c.isupper() for c in letters) and not _match_category(text)
|
||||
|
||||
|
||||
def _looks_like_continuation(text: str) -> bool:
|
||||
"""Fragment de description débordé sur la ligne suivante (ex. « Y », « IN »)."""
|
||||
return len(text) <= 4 and " " not in text and text.isalpha()
|
||||
|
||||
|
||||
def _column_keys(header_cells, page) -> list[str | None]:
|
||||
keys: list[str | None] = []
|
||||
for cell in header_cells:
|
||||
label = ""
|
||||
if cell is not None:
|
||||
label = _strip_accents((page.crop(cell).extract_text() or "").strip())
|
||||
if "debit" in label:
|
||||
key = "debit"
|
||||
elif "credit" in label:
|
||||
key = "credit"
|
||||
elif "t.v.a" in label or "tva" in label:
|
||||
key = "tva"
|
||||
elif "locatif" in label:
|
||||
key = "locatif"
|
||||
elif "deductible" in label:
|
||||
key = "deductible"
|
||||
elif "recapitulatif" in label:
|
||||
key = "desc"
|
||||
elif label == "":
|
||||
key = "left"
|
||||
else:
|
||||
key = None
|
||||
keys.append(key)
|
||||
return keys
|
||||
|
||||
|
||||
def _rows_from_page(page) -> list[dict]:
|
||||
"""Lignes visuelles du tableau « récapitulatif des opérations » d'une page."""
|
||||
table = None
|
||||
for candidate in page.find_tables():
|
||||
header = " ".join(
|
||||
(page.crop(c).extract_text() or "") if c is not None else ""
|
||||
for c in candidate.rows[0].cells
|
||||
)
|
||||
if "Locatif" in header and "ductible" in header:
|
||||
table = candidate
|
||||
break
|
||||
if table is None:
|
||||
return []
|
||||
|
||||
header = table.rows[0].cells
|
||||
keys = _column_keys(header, page)
|
||||
bands = [(c[0], c[2]) if c is not None else None for c in header]
|
||||
|
||||
def column_of(x_center: float) -> int | None:
|
||||
for i, band in enumerate(bands):
|
||||
if band and band[0] - 1 <= x_center <= band[1] + 1:
|
||||
return i
|
||||
return None
|
||||
|
||||
words = page.crop(table.bbox).extract_words()
|
||||
words.sort(key=lambda w: (round((w["top"] + w["bottom"]) / 2, 1), w["x0"]))
|
||||
|
||||
clusters: list[list] = []
|
||||
for word in words:
|
||||
y_center = (word["top"] + word["bottom"]) / 2
|
||||
if clusters and abs(y_center - clusters[-1][0]) <= _Y_TOL:
|
||||
clusters[-1][1].append(word)
|
||||
else:
|
||||
clusters.append([y_center, [word]])
|
||||
|
||||
rows: list[dict] = []
|
||||
for _y, line_words in clusters:
|
||||
cells: dict[str, list[str]] = {}
|
||||
for word in sorted(line_words, key=lambda w: w["x0"]):
|
||||
idx = column_of((word["x0"] + word["x1"]) / 2)
|
||||
key = keys[idx] if idx is not None else None
|
||||
if key is not None:
|
||||
cells.setdefault(key, []).append(word["text"])
|
||||
rows.append({k: " ".join(v) for k, v in cells.items()})
|
||||
return rows
|
||||
|
||||
|
||||
def extract_recapitulatif_operations_from_pdf(pdf_path: str) -> list[dict]:
|
||||
"""Extrait le récapitulatif des opérations par cellules de tableau.
|
||||
|
||||
Returns:
|
||||
Liste plate des opérations (même structure que
|
||||
:func:`plesna_gerance.parsers.operations.extract_recapitulatif_operations`).
|
||||
"""
|
||||
operations: list[dict] = []
|
||||
current_cat: str | None = None
|
||||
current_fournisseur: str | None = None
|
||||
current_sous_cat: str | None = None
|
||||
block_id = 0
|
||||
|
||||
with pdfplumber.open(pdf_path) as pdf:
|
||||
for page in pdf.pages:
|
||||
if "RECAPITULATIF DES OPERATIONS" not in (page.extract_text() or ""):
|
||||
continue
|
||||
|
||||
for row in _rows_from_page(page):
|
||||
left = (row.get("left") or "").strip()
|
||||
desc = (row.get("desc") or "").strip()
|
||||
montants = {k: _num(row.get(k, "")) for k in _AMOUNT_KEYS}
|
||||
has_amount = any(v is not None for v in montants.values())
|
||||
|
||||
low = _strip_accents(desc)
|
||||
|
||||
# En-tête, totaux, solde : ignorés.
|
||||
if low.startswith("recapitulatif"):
|
||||
continue
|
||||
if (
|
||||
low.startswith("totaux")
|
||||
or low.startswith("total des reglements")
|
||||
or "solde crediteur" in low
|
||||
):
|
||||
continue
|
||||
|
||||
# En-tête de catégorie (sans montant).
|
||||
cat = _match_category(left) or _match_category(desc)
|
||||
if cat and not has_amount:
|
||||
current_cat = cat
|
||||
current_fournisseur = None
|
||||
current_sous_cat = None
|
||||
block_id += 1
|
||||
continue
|
||||
|
||||
# Colonne de gauche : fournisseur (MAJUSCULES) ou sous-catégorie.
|
||||
if left:
|
||||
if _is_fournisseur(left):
|
||||
current_fournisseur = left
|
||||
else:
|
||||
current_sous_cat = left
|
||||
|
||||
# Ligne sans montant : sous-catégorie (col1) ou continuation de description.
|
||||
if not has_amount:
|
||||
if desc:
|
||||
if (
|
||||
_looks_like_continuation(desc)
|
||||
and operations
|
||||
and operations[-1]["_block"] == block_id
|
||||
):
|
||||
operations[-1]["description"] = (
|
||||
operations[-1]["description"] + desc
|
||||
).strip()
|
||||
else:
|
||||
current_sous_cat = desc
|
||||
continue
|
||||
|
||||
# Ligne avec montant : une opération.
|
||||
operations.append(
|
||||
{
|
||||
"categorie": current_cat,
|
||||
"sous_categorie": current_sous_cat or "",
|
||||
"fournisseur": current_fournisseur,
|
||||
"description": desc,
|
||||
"lot_concerne": None,
|
||||
"lot_numero": _extract_lot_code_from_description(desc),
|
||||
"montants": {k: (montants[k] or 0.0) for k in _AMOUNT_KEYS},
|
||||
"_block": block_id,
|
||||
}
|
||||
)
|
||||
|
||||
# Report du fournisseur unique d'un bloc sur les opérations qui en manquent
|
||||
# (cas des honoraires : le nom du gestionnaire n'apparaît qu'une fois, au milieu).
|
||||
by_block: dict[int, list[dict]] = {}
|
||||
for op in operations:
|
||||
by_block.setdefault(op["_block"], []).append(op)
|
||||
for block_ops in by_block.values():
|
||||
first = next((o["fournisseur"] for o in block_ops if o["fournisseur"]), None)
|
||||
if first:
|
||||
for op in block_ops:
|
||||
if not op["fournisseur"]:
|
||||
op["fournisseur"] = first
|
||||
|
||||
for op in operations:
|
||||
op.pop("_block", None)
|
||||
|
||||
return operations
|
||||
Reference in New Issue
Block a user