Python-Projekte/gcstruct/skr51.py

import pandas as pd
# import numpy as np
from pathlib import Path


class skr51_validation():
    current_dir = Path('C:/Projekte/Python/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)]

        result = []
        for x in whitelist.split(','):
            if '-' in x:
                result += [str(i).zfill(2) for i in range(int(x.split('-')[0]), int(x.split('-')[1]) + 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

    @staticmethod
    def format_konto(k):
        return f"{k['Kontonummer']}-{k['Marke']}-{k['Standort']}-{k['Kostenstelle']}-{k['Absatzkanal']}-{k['Kostentraeger']}"

    @staticmethod
    def format_konto_neu(k):
        return f"{k['Kontonummer_neu']}-{k['Marke_neu']}-{k['Standort_neu']}-{k['Kostenstelle_neu']}-{k['Absatzkanal_neu']}-{k['Kostentraeger_neu']}"


if __name__ == '__main__':
    skr = skr51_validation()
    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)