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(__file__).parent.parent.resolve()
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']}"
            / f"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()