Python-Projekte/gcstruct/skr51.py

from itertools import product

# import numpy as np
from pathlib import Path

import pandas as pd


class skr51_validation:
    current_dir = Path("/home/robert/projekte/python/gcstruct/SKR51")

    def __init__(self):
        filter_csv = pd.read_csv(
            self.current_dir.joinpath("SKR51_Kostenrechnungsmerkmale.csv"), sep=";", quotechar='"'
        )  # , encoding='latin-1')
        filter_csv["Filter"] = filter_csv["Filter"].apply(self.create_list)
        self.filter = filter_csv.set_index("Block").to_dict(orient="dict")["Filter"]
        # print(self.filter)

        explode_cols = ["Kostenstelle", "Absatzkanal", "Kostenträger"]
        matrix_csv = pd.read_csv(
            self.current_dir.joinpath("Plausibilitaetsmatrix.csv"), sep=";", converters={i: str for i in range(0, 200)}
        )  # , encoding='latin-1'  .set_index('Konto_Nr')
        matrix_csv["valid"] = True
        for col in explode_cols:
            matrix_csv[col] = matrix_csv[col].apply(self.create_list)
        self.matrix = matrix_csv
        # self.matrix['Kostenstelle'] = matrix_csv[['Konto_Nr', 'Kostenstelle', 'valid']].explode('Kostenstelle')
        # print(self.matrix[self.matrix['Konto_Nr'] == '3300'])

    def create_list(self, whitelist):
        if whitelist == "":
            # return [str(i).zfill(2) for i in range(0, 100)]
            return ["00"]

        result = []
        for x in whitelist.split(","):
            if "-" in x:
                start, end = map(int, x.split("-"))
                result += [str(i).zfill(2) for i in range(start, end + 1)]
            else:
                result.append(x)
        return result

    def check_bookings(self, bookings_file):
        export_csv = pd.read_csv(
            self.current_dir.joinpath(bookings_file),
            sep=";",
            decimal=",",
            converters={i: str for i in range(0, 200)},
            encoding="latin-1",
        )
        export_csv = export_csv.merge(
            self.matrix[["Konto_Nr", "valid"]], how="left", on="Konto_Nr", suffixes=(None, "_matrix")
        )
        export_csv["Konto_Nr_valid"] = export_csv["valid"] == True
        temp = export_csv["Konto_Nr_Händler"]
        export_csv.drop(["Konto_Nr_Händler"], axis=1, inplace=True)
        export_csv.insert(0, "Konto_Nr_Händler", temp)

        for col in ["Marke", "Standort"]:
            export_csv[col + "_valid"] = export_csv[col].isin(self.filter[col])
            export_csv["valid"] = export_csv["valid"] & export_csv[col + "_valid"]

        for col in ["Kostenstelle", "Absatzkanal", "Kostenträger"]:
            export_csv = export_csv.merge(
                self.matrix[["Konto_Nr", col, "valid"]].explode(col),
                how="left",
                on=["Konto_Nr", col],
                suffixes=(None, "_" + col),
            )
            export_csv[col + "_valid"] = export_csv[col].isin(self.filter[col]) & export_csv["valid_" + col] == True
            export_csv.drop(["valid_" + col], axis=1, inplace=True)
            export_csv["valid"] = export_csv["valid"] & export_csv[col + "_valid"]
        return export_csv

    def maximum_accounts(self):
        accounts = self.matrix.to_dict(orient="records")
        header = [
            "Account",
            "Make",
            "Site",
            "Origin",
            "SalesChannel",
            "CostCarrier",
            "CostAccountingString",
            "Decimals",
            "OpeningBalance",
            "Period01",
            "CumulatedYear",
            "Marke_HBV",
        ]
        with open(self.current_dir.joinpath("maximum.csv"), "w", encoding="latin-1") as fwh:
            fwh.write(";".join(header) + "\n")

            for a in accounts:
                for entry in product(a["Kostenstelle"], a["Absatzkanal"], a["Kostenträger"]):
                    row = [a["Konto_Nr"], "01", "01", *entry, "0101" + "".join(entry), "2", "0", "100", "100", "0000"]
                    fwh.write(";".join(row) + "\n")
        return None

    @staticmethod
    def format_konto(k):
        return "{0}-{1}-{2}-{3}-{4}-{5}".format(
            k["Kontonummer"], k["Marke"], k["Standort"], k["Kostenstelle"], k["Absatzkanal"], k["Kostentraeger"]
        )

    @staticmethod
    def format_konto_neu(k):
        return "{0}-{1}-{2}-{3}-{4}-{5}".format(
            k["Kontonummer_neu"],
            k["Marke_neu"],
            k["Standort_neu"],
            k["Kostenstelle_neu"],
            k["Absatzkanal_neu"],
            k["Kostentraeger_neu"],
        )


if __name__ == "__main__":
    skr = skr51_validation()
    skr.maximum_accounts()

    # df = skr.check_bookings('Kontenrahmen_kombiniert.csv')
    # df[['Konto_Nr_Händler', 'Marke', 'Marke_valid', 'Standort', 'Standort_valid', 'Konto_Nr', 'Konto_Nr_valid',
    #     'Kostenstelle', 'Kostenstelle_valid', 'Absatzkanal', 'Absatzkanal_valid', 'Kostenträger', 'Kostenträger_valid',
    #     'Kostenträger_Ebene', 'valid']].to_csv(skr.current_dir.joinpath('SKR51_Validierung.csv'),
    # sep=';', encoding='latin-1', index=False)