planner/webservice/hbv_export.py

import pandas as pd
import numpy as np
from datetime import datetime
from gnupg_encrypt import encrypt
import os
from pathlib import Path


current_year = '2023'
current_version = 'V1'

base_dir = Path().cwd()
config_dir = base_dir / 'config/hbv'
export_dir = base_dir / 'export'
hb_format = config_dir / 'hb_format.csv'
hb_department = config_dir / 'hb_department.csv'
hb_translation = config_dir / 'hb_translation.csv'
plan_amount = export_dir / f'Planner_{current_year}_{current_version}_Stk.csv'
plan_values = export_dir / f'Planner_{current_year}_{current_version}_Plan.csv'

hb_ignored = export_dir / 'hbv/ignoriert.csv'

current_date = datetime.now().strftime('%d%m%Y%H%M%S')
# current_date = '24032021112656'


def main():
    # Übersetzungstabelle importieren
    df_translation = pd.read_csv(hb_translation, decimal=',', sep=';',
                                 encoding='latin-1', converters={i: str for i in range(0, 200)})
    # df_translation['column_no_join'] = np.where(df_translation['column_no']
    # .isin(['1', '3', '4']), df_translation['column_no'], '0')
    # Department-Zuordnung importieren
    df_department = pd.read_csv(hb_department, decimal=',', sep=';',
                                encoding='latin-1', converters={i: str for i in range(0, 200)})

    # Planwerte importieren
    values_converter = {i: str for i in range(0, 200)}
    values_converter[4] = lambda x: np.float64(x.replace(',', '.') if x != '' else 0.0)
    values_converter[5] = values_converter[4]
    df_values = pd.read_csv(plan_values, decimal=',', sep=';',
                            encoding='latin-1', converters=values_converter)
    df_values['Gesamt'] = df_values['Gesamt'] + df_values['Periode13']
    df_values['type'] = '2'
    df_values['type'] = np.where(df_values['Vstufe 1'].isin(['Materialaufwand']), '3', df_values['type'])
    df_amount = pd.read_csv(plan_amount, decimal=',', sep=';', 
                            encoding='latin-1', converters=values_converter)
    df_amount['type'] = '1'
    df: pd.DataFrame = df_values.append(df_amount)

    # Planwerte alle positiv
    df['Minus1'] = np.where(df['Vstufe 1'].isin(['Umsatzerlöse', 'Verk. Stückzahlen']) | df['Zeile'].isin(['7410', '7440']), 1, -1)
    df['Gesamt'] = df['Gesamt'] * df['Minus1']

    # Planwerte übersetzen
    df = df.merge(df_department, how='inner', left_on='Betrieb Nr', right_on='department_id')
    df = df.merge(df_translation, how='left', left_on=['Zeile', 'type'], right_on=['from', 'type'])
    # fehlende Übersetzung
    df_ignored = df[(df['to'].isna()) & (df['Gesamt'] != 0)]
    df_ignored.to_csv(hb_ignored, decimal=',', sep=';', encoding='latin-1', index=False)

    # Planwerte formatieren und exportieren
    rename_from = ['bm_code', 'BV_NUMMER', 'FILIAL_NR', 'to', 'column_no', 'Jahr', 'Gesamt']
    rename_to = ['BM_CODE', 'BV_NUMMER', 'FILIAL_NR', 'ZEILE', 'SPALTE', 'JAHR', 'WERT']
    df_valid = df[df['to'].notna()].rename(columns=dict(zip(rename_from, rename_to)))
    df_valid['SPALTE'] = df_valid['SPALTE'].str.zfill(3)
    group_by = ['BM_CODE', 'BV_NUMMER', 'FILIAL_NR']
    df_valid = df_valid[rename_to].groupby(group_by)

    for group in df_valid.groups:
        g = dict(zip(group_by, group))
        filename = export_dir / f"hbv/{current_year}/{g['BV_NUMMER']}_{g['FILIAL_NR']}/HB{g['BM_CODE']}{current_year}00{g['BV_NUMMER']}{g['FILIAL_NR']}0{current_date}.dat"
        os.makedirs(filename.parent, exist_ok=True)
        df_group = df_valid.get_group(group).groupby(rename_to[:-1]).sum().reset_index()
        with open(filename, 'w') as fwh:
            for row in df_group.to_dict(orient='records'):
                fwh.write("I0155{BV_NUMMER}{FILIAL_NR}0{ZEILE}{SPALTE}00{JAHR}{WERT:16.2f}03\n".format(**row))
        encrypt(filename)


if __name__ == '__main__':
    main()