planner/webservice/plan_export.py

import pandas as pd
import numpy as np

base_dir = "/home/robert/projekte/planner/export"

id_header = ["Ebene" + str(i) for i in range(1, 11)]
# values2_header = ['VJ', 'AJ', 'FC', 'Plan_ori', 'Plan_Prozent', 'Stk', 'VAK', 'BE_Prozent', 'Plan_VJ', 'Plan_Stk_VJ', 'Plan',
#                   'Jan', 'Feb', 'Mar', 'Apr', 'Mai', 'Jun', 'Jul', 'Aug', 'Sep', 'Okt', 'Nov', 'Dez', 'Periode13']
values2_header = [
    "Plan",
    "Jan",
    "Feb",
    "Mar",
    "Apr",
    "Mai",
    "Jun",
    "Jul",
    "Aug",
    "Sep",
    "Okt",
    "Nov",
    "Dez",
    "Periode13",
    "Plan_ori",
    "Plan_Prozent",
    "Stk",
    "VAK",
    "BE_Prozent",
    "frei",
    "VJ",
    "VJ_Stk",
    "AJ",
    "AJ_Stk",
    "AJ_Okt",
    "AJ_Okt_Stk",
    "FC",
    "FC_Stk",
    "Plan_VJ",
    "Plan_VJ_Stk",
]
season_header = ["Jan", "Feb", "Mar", "Apr", "Mai", "Jun", "Jul", "Aug", "Sep", "Okt", "Nov", "Dez"]
info_header = ["text", "costcenter", "department"]
header = info_header + id_header + values2_header
season_export_header = info_header + season_header

source_header = ["department", "text", "costcenter", "Ebene1", "Plan", "Periode13"]
export_header = [
    "Betrieb Nr",
    "Zeile mit Bez",
    "Bereich",
    "Vstufe 1",
    "Gesamt",
    "Periode13",
]  # 'Version', 'Konto', 'Jahr']


def expand(df, header, values_label):
    for i, key in enumerate(header):
        df[key] = df[values_label].str[i]
    return df


def apply_season(df):
    df["Saison"] = df["Ebene1"].str.contains("Umsatzerlöse|Materialaufwand|Verkaufsabh. Kosten")
    df["saison_sum"] = df[season_header].sum(axis=1)
    for i, key in enumerate(season_header):
        df["temp"] = np.where(
            (df["Saison"]) & (df[key + "_2"] != 8.3333), df["Plan"] * df[key + "_2"] / 100, df["Plan"] / 12
        )
        df[key] = np.where(df["saison_sum"] == 0, df["temp"], df[key] * df["Minus1"])
    df["Dez"] = df["Plan"] - df[season_header].sum(axis=1) + df["Dez"]
    return df


def data_cleansing(filename):
    df = pd.read_json(filename)
    df["values2"] = df["values2"].apply(lambda v: list(v.items()))
    df = df.explode("values2")
    df["department"], df["values2"] = zip(*df["values2"])

    df["id"] = df["id"].str.split(";")
    df = expand(df, id_header, "id")
    df = expand(df, values2_header, "values2")
    return df


def export_plan(filename, version, target_year, amount_value):
    df = data_cleansing(f"{base_dir}/{filename}.json")
    season = df[(df["level"] == 2) & (df["Ebene1"] == "Umsatzerlöse")].copy()
    season["Dez"] = (100 - season[season_header].sum(axis=1) + season["Dez"]).round(4)
    season[season_export_header].to_csv(
        f"{base_dir}/Planner_{target_year}_{version}_Saison.csv", encoding="latin_1", sep=";", decimal=",", index=False
    )

    df["Minus1"] = np.where(df["Ebene1"] != "Umsatzerlöse", -1, 1)
    df["Plan"] = df[amount_value] * df["Minus1"]
    if amount_value == "Plan":
        df["Periode13"] = df["Periode13"] * df["Minus1"]
    else:
        df["Periode13"] = 0
    plan = df[df["planlevel"] == True]

    plan = pd.merge(plan, season, how="left", on=["Ebene2", "department"], suffixes=("", "_2"))
    plan = apply_season(plan)
    plan = plan[source_header + season_header].rename(columns=dict(zip(source_header, export_header)))

    # Reisacher Spezialbedingungen
    plan["Zeile"] = plan["Zeile mit Bez"].str.slice(stop=4)
    plan["Zeile"] = np.where(
        plan["Zeile mit Bez"].isin(["BMW aus Leasingrücklauf BFS", "BMW aus Leasingrücklauf Alphabet"]),
        "3040",
        plan["Zeile"],
    )
    plan["Zeile"] = np.where(
        plan["Zeile mit Bez"].isin(["BMW an Wiederverkäufer BFS", "BMW an Wiederverkäufer Alphabet"]),
        "3120",
        plan["Zeile"],
    )

    desciption = pd.read_csv(
        f"{base_dir}/../data/Planner_Zeilen_Bez.csv", sep=";", encoding="latin-1", dtype={0: str, 1: str}
    )
    plan = pd.merge(plan, desciption, how="left", on=["Zeile"], suffixes=["", "_3"])
    plan["Zeile mit Bez"] = plan["Zeile mit Bez_3"]
    plan.drop(["Zeile mit Bez_3"], axis=1, inplace=True)

    if amount_value == "Stk":
        plan = plan[plan["Vstufe 1"] == "Umsatzerlöse"]
        plan["Vstufe 1"] = "Verk. Stückzahlen"

    plan["Version"] = version
    plan["Konto"] = ""
    plan["Jahr"] = target_year

    plan.to_csv(
        f"{base_dir}/Planner_{target_year}_{version}_{amount_value}.csv",
        encoding="latin_1",
        sep=";",
        decimal=",",
        index=False,
    )


if __name__ == "__main__":
    # export_plan('V3', '2021', 'Plan')
    # export_plan('V3', '2021', 'Stk')
    filename = "../save/2023_V1_20230214225753"
    export_plan(filename, "V1", "2023", "Plan")
    export_plan(filename, "V1", "2023", "Stk")