pdf_oralia_vibe/src/plesna_gerance/parsers/operations_table.py

"""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 ..utils.lots import extract_lot_numero_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_numero_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