plesna/scripts/history_stagging.py

import logging
from pathlib import Path

import pandas as pd

from scripts.flux import consume_fluxes

from .flux import Destination, ExcelSource, Flux, SplitDestination, Transformation

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


def extract_cat(cat: pd.DataFrame):
    cat_drop = list(cat[cat["Nouvelles"] == "NE PAS IMPORTER"]["Anciennes"])
    cat_trans = cat[cat["Nouvelles"] != "NE PAS IMPORTER"]

    trans = {}
    for _, (old, new) in cat_trans.iterrows():
        trans[old] = new

    return trans, cat_drop


def lot_naming(value):
    if str(value).isnumeric():
        return str(value).zfill(2)
    return "PC"


def trans_2017_2021(
    dfs: list[pd.DataFrame], stagging_columns: list[str], **kwrds
) -> pd.DataFrame:
    df, cat = dfs
    cat_trans, cat_drop = extract_cat(cat)

    df = df[~df["Categorie"].isin(cat_drop)]

    df = df.assign(
        Immeuble=df["immeuble"],
        Porte=df["porte"],
        Débit=df["Débit"].fillna(0),
        Crédit=df["Crédit"].fillna(0),
        Lot=df["porte"].apply(lot_naming),
        Année=df["Date"].astype(str).str.slice(0, 4),
        Mois=df["Date"].astype(str).str.slice(5, 7),
        Catégorie=df["Categorie"].replace(cat_trans),
        Fournisseur="",
    )

    return df[stagging_columns]


def trans_2022_charge(
    dfs: list[pd.DataFrame], stagging_columns: list[str], **kwrds
) -> pd.DataFrame:
    df = dfs[0]
    df = df.assign(
        Immeuble=df["immeuble"],
        Porte=df["lot"],
        Débit=df["Débits"].fillna(0),
        Crédit=df["Crédits"].fillna(0),
        Lot=df["lot"].apply(lot_naming),
        Année=df["annee"],
        Mois=df["mois"],
        Catégorie=df["Catégorie Charge"],
        # Catégorie=df["Catégorie Charge"].replace(trans),
        Fournisseur="",
        Régie="Oralia - Gelas",
        Libellé="",
    )
    return df[stagging_columns]


def trans_2022_loc(
    dfs: list[pd.DataFrame], stagging_columns: list[str], **kwrds
) -> pd.DataFrame:
    df = dfs[0]
    df = df.assign(
        Immeuble=df["immeuble"],
        Porte=df["lot"],
        Débit=0,
        Crédit=df["Réglés"].fillna(0),
        Lot=df["lot"].apply(lot_naming),
        Année=df["annee"],
        Mois=df["mois"],
        Catégorie="Loyer Charge",
        Fournisseur="",
        Régie="Oralia - Gelas",
        Libellé="",
    )
    return df[stagging_columns]


def trans_2023(
    dfs: list[pd.DataFrame], year: str, stagging_columns: list[str], **kwrds
) -> pd.DataFrame:
    df = dfs[0]
    df = df.assign(
        Débit=df["Débit"].fillna(0),
        Crédit=df["Crédit"].fillna(0),
        Lot=lot_naming(df["Porte"]),
        Année=year,
    )
    return df[stagging_columns]


STAGGING_COLUMNS = [
    "Régie",
    "Immeuble",
    "Porte",
    "Lot",
    "Année",
    "Mois",
    "Catégorie",
    "Fournisseur",
    "Libellé",
    "Débit",
    "Crédit",
]

FLUXES = {
    "2017 2021 - charge et locataire.xlsx": Flux(
        sources=[
            ExcelSource(
                filename="2017 2021 - charge et locataire.xlsx", sheet_name="DB CRG"
            ),
            ExcelSource(
                filename="2017 2021 - charge et locataire.xlsx",
                sheet_name="Catégories",
            ),
        ],
        transformation=Transformation(
            function=trans_2017_2021,
            extra_kwrds={"stagging_columns": STAGGING_COLUMNS},
        ),
        destination=SplitDestination(name="crg", split_column="Année"),
    ),
    "2022 - charge.xlsx": Flux(
        sources=[
            ExcelSource(filename="2022 - charge.xlsx", sheet_name="Sheet1"),
        ],
        transformation=Transformation(
            function=trans_2022_charge,
            extra_kwrds={"stagging_columns": STAGGING_COLUMNS},
        ),
        destination=SplitDestination(name="crg", split_column="Année"),
    ),
    "2022 - locataire.xlsx": Flux(
        sources=[
            ExcelSource(filename="2022 - locataire.xlsx", sheet_name="Sheet1"),
        ],
        transformation=Transformation(
            function=trans_2022_loc,
            extra_kwrds={"stagging_columns": STAGGING_COLUMNS},
        ),
        destination=SplitDestination(name="crg", split_column="Année"),
    ),
    "2023 - charge et locataire.xlsx": Flux(
        sources=[
            ExcelSource(
                filename="2023 - charge et locataire.xlsx",
                sheet_name="DB CRG 2023 ...",
            ),
        ],
        transformation=Transformation(
            function=trans_2023,
            extra_kwrds={"year": 2023, "stagging_columns": STAGGING_COLUMNS},
        ),
        destination=SplitDestination(name="crg", split_column="Année"),
    ),
}

if __name__ == "__main__":
    data_path = Path("datas/")
    assert data_path.exists()
    history_path = data_path / "Histoire"
    assert history_path.exists()
    history_crg_path = history_path / "CRG"
    assert history_crg_path.exists()

    staging_path = data_path / "staging"
    assert staging_path.exists()
    staging_crg_path = staging_path / "CRG"
    assert staging_crg_path.exists()

    crg_files = consume_fluxes(
        fluxes=FLUXES,
        origin_path=history_crg_path,
        dest_path=staging_crg_path,
    )
    print(crg_files)