Python-Projekte/gcstruct/skr51.py

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


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)