GCHR verschoben

gcstruct/gchr.py → gchr/gchr.py

@@ -1,211 +1,211 @@
-import logging
-import os
-from dataclasses import dataclass
-from datetime import datetime
-from pathlib import Path
-from typing import Callable
-
-import pandas as pd
-
-from gcstruct.gchr_bookings import GchrBookings
-from gcstruct.gchr_export import (
-    ACCOUNT_INFO,
-    GchrExportConfig,
-    GchrExportFormat,
-    get_export_fn,
-)
-from gcstruct.gchr_translate import load_translation
-
-
-@dataclass
-class GchrConfig:
-    first_month_of_financial_year: str
-    data_dir: str
-    gcstruct_dir: str
-    export_dir: str
-    export_fn = Callable[[GchrExportConfig], None]
-
-
-class GCHR:
-    booking_date: datetime
-    bookings: GchrBookings
-    _df_translate: pd.DataFrame = None
-    df_translate2: pd.DataFrame = None
-    makes: dict[str, str] = None
-    sites: dict[str, str] = None
-    current_year: str
-    current_month: str
-    timestamp: str
-
-    def __init__(self, base_dir: str) -> None:
-        self.base_dir = base_dir
-        os.makedirs(self.base_dir + "/data", exist_ok=True)
-        os.makedirs(self.base_dir + "/export/temp", exist_ok=True)
-        os.makedirs(self.base_dir + "/logs", exist_ok=True)
-
-        self.account_translation = f"{self.base_dir}/data/Kontenrahmen_uebersetzt.csv"
-
-        self.first_month_of_financial_year = "10"
-        self.bookings = GchrBookings(self.base_dir, self.first_month_of_financial_year)
-
-        self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-
-        pd.set_option("display.max_rows", 500)
-        pd.set_option("display.float_format", lambda x: "%.2f" % x)
-
-    @property
-    def debug_file(self) -> str:
-        return f"{self.logs_dir}/debug_{self.timestamp}.csv"
-
-    @property
-    def account_ignored(self) -> str:
-        return f"{self.export_info_dir}/ignoriert_{self.period}.csv"
-
-    # self.account_invalid = f"{self.export_info_dir}/ungueltig_{self.period}.csv"
-
-    def export_all_periods(self, overwrite=False, today=None) -> None:
-        periods = GCHR.get_all_periods(today)
-
-        for year, month in periods:
-            filename = self.export_filename_for_period(year, month)
-            if overwrite or not Path(filename).exists():
-                os.makedirs(Path(filename).parent.joinpath("info"), exist_ok=True)
-                self.export_period(year, month)
-
-    @staticmethod
-    def get_all_periods(today=None) -> list[tuple[str, str]]:
-        dt = datetime.now()
-        if today is not None:
-            dt = datetime.fromisoformat(today)
-        prev = str(dt.year - 1)
-        periods = [(prev, str(x).zfill(2)) for x in range(dt.month, 13)] + [
-            (str(dt.year), str(x).zfill(2)) for x in range(1, dt.month)
-        ]
-        return periods
-
-    def export_period(self, year: str, month: str) -> str:
-        export_fn = get_export_fn(GchrExportFormat.SKR51)
-        # Kontensalden laden
-        df_bookings = self.bookings.filter_bookings(year, month)
-        all_periods = set(df_bookings["Bookkeep Period"].to_list())
-        bookkeep_period_date = datetime(int(year), int(month), 28)
-
-        if df_bookings.shape[0] == 0 or len(all_periods) <= 1 or self.booking_date < bookkeep_period_date:
-            logging.error("ABBRUCH!!! Keine Daten vorhanden!")
-            return False
-
-        filter_to = year + month
-        period_no = list(self.bookings.bookkeep_filter.keys()).index(filter_to) + 1
-
-        logging.info("df_bookings: " + str(df_bookings.shape))
-        # Join auf Übersetzung
-        df_combined = df_bookings.merge(self._df_translate, how="inner", on="Konto_Nr_Händler")
-        logging.info(f"df_combined: {df_combined.shape}")
-
-        df_pivot = df_combined.pivot_table(
-            index=["Konto_Nr_SKR51"],
-            columns=["period"],
-            values="amount",
-            aggfunc="sum",
-            margins=True,
-            margins_name="CumulatedYear",
-        )
-        df_pivot.drop(index="CumulatedYear", inplace=True)
-
-        logging.info("df_pivot: " + str(df_pivot.shape))
-
-        df = df_pivot.merge(self.df_translate2, how="inner", on="Konto_Nr_SKR51")
-
-        makes_used = {}
-        for m in sorted(list(set(df["Marke"].to_list()))):
-            if m not in self.makes:
-                continue
-            makes_used[m] = self.makes[m]
-        sites_used = {}
-        for s in sorted(list(set((df["Marke"] + "-" + df["Standort"]).to_list()))):
-            if s not in self.sites:
-                continue
-            sites_used[s] = self.sites[s]
-
-        from_label = ["Marke", "Standort", "Konto_Nr", "Kostenstelle", "Absatzkanal", "Kostenträger", "KRM"]
-        to_label = ["Make", "Site", "Account", "Origin", "SalesChannel", "CostCarrier", "CostAccountingString"]
-        col_dict = dict(zip(from_label, to_label))
-        df = df.rename(columns=col_dict)
-
-        export_filename = self.export_filename_for_period(year, month)
-        export_csv = export_filename[:-4] + ".csv"
-        df.to_csv(export_csv, decimal=",", sep=";", encoding="latin-1", index=False)
-        df = df[df["IsNumeric"] != False].groupby(ACCOUNT_INFO, as_index=False).aggregate("sum")
-
-        # Infos ergänzen
-        df["Decimals"] = 2
-        # df.sort_values(by=["Konto_Nr_SKR51"], inplace=True)
-        logging.info(df.shape)
-        main_sites = [self.sites[s] for s in sites_used if s in self.sites and self.sites[s] != "0000"]
-
-        for i, main_site in enumerate(main_sites):
-            filename = export_filename
-            if i > 0:
-                filename = f"{filename[:-4]}_{main_site}.xml"
-            export_cfg = GchrExportConfig(
-                main_site,
-                year,
-                month,
-                makes_used,
-                sites_used,
-                self.first_month_of_financial_year,
-                period_no,
-                self.bookings.bookkeep_filter,
-                filename,
-                df.to_dict(orient="records"),
-            )
-
-            export_fn(export_cfg)
-
-        # Join auf Übersetzung - nicht zugeordnet
-        df_ignored = df_bookings.merge(self.df_translate, how="left", on="Konto_Nr_Händler")
-        df_ignored = df_ignored[df_ignored["Konto_Nr_SKR51"].isna()]
-        if not df_ignored.empty:
-            df_ignored = df_ignored.pivot_table(
-                index=["Konto_Nr_Händler"],
-                columns=["period"],
-                values="amount",
-                aggfunc="sum",
-                margins=True,
-                margins_name="CumulatedYear",
-            )
-            df_ignored.to_csv(self.account_ignored, decimal=",", sep=";", encoding="latin-1")
-        return export_filename
-
-    @property
-    def df_translate(self) -> pd.DataFrame:
-        if self._df_translate is None:
-            self.makes, self.sites, self._df_translate, self.df_translate2 = load_translation(
-                self.account_translation, self.debug_file, self.export_invalid_filename
-            )
-        return self._df_translate
-
-    @property
-    def export_info_dir(self) -> str:
-        return f"{self.base_dir}/Export/{self.current_year}/info/"
-
-    @property
-    def logs_dir(self) -> str:
-        return f"{self.base_dir}/Logs/"
-
-    @property
-    def export_invalid_filename(self) -> str:
-        return f"{self.base_dir}/Export/ungueltig.csv"
-
-    def export_filename_for_period(self, year: str, month: str) -> str:
-        return f"{self.base_dir}/Export/{year}/export_{year}-{month}.xml"
-
-
-def gchr_local() -> None:
-    base_dir = os.getcwd() + "/../GCHR2_Testdaten/Kunden"
-    for path in Path(base_dir).glob("*"):
-        if not path.is_dir():
-            continue
-        print(path.name)
-        gchr = GCHR(str(path))
-        gchr.export_all_periods()
+import logging
+import os
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Callable
+
+import pandas as pd
+
+from gcstruct.gchr_bookings import GchrBookings
+from gcstruct.gchr_export import (
+    ACCOUNT_INFO,
+    GchrExportConfig,
+    GchrExportFormat,
+    get_export_fn,
+)
+from gcstruct.gchr_translate import load_translation
+
+
+@dataclass
+class GchrConfig:
+    first_month_of_financial_year: str
+    data_dir: str
+    gcstruct_dir: str
+    export_dir: str
+    export_fn = Callable[[GchrExportConfig], None]
+
+
+class GCHR:
+    booking_date: datetime
+    bookings: GchrBookings
+    _df_translate: pd.DataFrame = None
+    df_translate2: pd.DataFrame = None
+    makes: dict[str, str] = None
+    sites: dict[str, str] = None
+    current_year: str
+    current_month: str
+    timestamp: str
+
+    def __init__(self, base_dir: str) -> None:
+        self.base_dir = base_dir
+        os.makedirs(self.base_dir + "/data", exist_ok=True)
+        os.makedirs(self.base_dir + "/export/temp", exist_ok=True)
+        os.makedirs(self.base_dir + "/logs", exist_ok=True)
+
+        self.account_translation = f"{self.base_dir}/data/Kontenrahmen_uebersetzt.csv"
+
+        self.first_month_of_financial_year = "10"
+        self.bookings = GchrBookings(self.base_dir, self.first_month_of_financial_year)
+
+        self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+
+        pd.set_option("display.max_rows", 500)
+        pd.set_option("display.float_format", lambda x: "%.2f" % x)
+
+    @property
+    def debug_file(self) -> str:
+        return f"{self.logs_dir}/debug_{self.timestamp}.csv"
+
+    @property
+    def account_ignored(self) -> str:
+        return f"{self.export_info_dir}/ignoriert_{self.period}.csv"
+
+    # self.account_invalid = f"{self.export_info_dir}/ungueltig_{self.period}.csv"
+
+    def export_all_periods(self, overwrite=False, today=None) -> None:
+        periods = GCHR.get_all_periods(today)
+
+        for year, month in periods:
+            filename = self.export_filename_for_period(year, month)
+            if overwrite or not Path(filename).exists():
+                os.makedirs(Path(filename).parent.joinpath("info"), exist_ok=True)
+                self.export_period(year, month)
+
+    @staticmethod
+    def get_all_periods(today=None) -> list[tuple[str, str]]:
+        dt = datetime.now()
+        if today is not None:
+            dt = datetime.fromisoformat(today)
+        prev = str(dt.year - 1)
+        periods = [(prev, str(x).zfill(2)) for x in range(dt.month, 13)] + [
+            (str(dt.year), str(x).zfill(2)) for x in range(1, dt.month)
+        ]
+        return periods
+
+    def export_period(self, year: str, month: str) -> str:
+        export_fn = get_export_fn(GchrExportFormat.SKR51)
+        # Kontensalden laden
+        df_bookings = self.bookings.filter_bookings(year, month)
+        all_periods = set(df_bookings["Bookkeep Period"].to_list())
+        bookkeep_period_date = datetime(int(year), int(month), 28)
+
+        if df_bookings.shape[0] == 0 or len(all_periods) <= 1 or self.booking_date < bookkeep_period_date:
+            logging.error("ABBRUCH!!! Keine Daten vorhanden!")
+            return False
+
+        filter_to = year + month
+        period_no = list(self.bookings.bookkeep_filter.keys()).index(filter_to) + 1
+
+        logging.info("df_bookings: " + str(df_bookings.shape))
+        # Join auf Übersetzung
+        df_combined = df_bookings.merge(self._df_translate, how="inner", on="Konto_Nr_Händler")
+        logging.info(f"df_combined: {df_combined.shape}")
+
+        df_pivot = df_combined.pivot_table(
+            index=["Konto_Nr_SKR51"],
+            columns=["period"],
+            values="amount",
+            aggfunc="sum",
+            margins=True,
+            margins_name="CumulatedYear",
+        )
+        df_pivot.drop(index="CumulatedYear", inplace=True)
+
+        logging.info("df_pivot: " + str(df_pivot.shape))
+
+        df = df_pivot.merge(self.df_translate2, how="inner", on="Konto_Nr_SKR51")
+
+        makes_used = {}
+        for m in sorted(list(set(df["Marke"].to_list()))):
+            if m not in self.makes:
+                continue
+            makes_used[m] = self.makes[m]
+        sites_used = {}
+        for s in sorted(list(set((df["Marke"] + "-" + df["Standort"]).to_list()))):
+            if s not in self.sites:
+                continue
+            sites_used[s] = self.sites[s]
+
+        from_label = ["Marke", "Standort", "Konto_Nr", "Kostenstelle", "Absatzkanal", "Kostenträger", "KRM"]
+        to_label = ["Make", "Site", "Account", "Origin", "SalesChannel", "CostCarrier", "CostAccountingString"]
+        col_dict = dict(zip(from_label, to_label))
+        df = df.rename(columns=col_dict)
+
+        export_filename = self.export_filename_for_period(year, month)
+        export_csv = export_filename[:-4] + ".csv"
+        df.to_csv(export_csv, decimal=",", sep=";", encoding="latin-1", index=False)
+        df = df[df["IsNumeric"] != False].groupby(ACCOUNT_INFO, as_index=False).aggregate("sum")
+
+        # Infos ergänzen
+        df["Decimals"] = 2
+        # df.sort_values(by=["Konto_Nr_SKR51"], inplace=True)
+        logging.info(df.shape)
+        main_sites = [self.sites[s] for s in sites_used if s in self.sites and self.sites[s] != "0000"]
+
+        for i, main_site in enumerate(main_sites):
+            filename = export_filename
+            if i > 0:
+                filename = f"{filename[:-4]}_{main_site}.xml"
+            export_cfg = GchrExportConfig(
+                main_site,
+                year,
+                month,
+                makes_used,
+                sites_used,
+                self.first_month_of_financial_year,
+                period_no,
+                self.bookings.bookkeep_filter,
+                filename,
+                df.to_dict(orient="records"),
+            )
+
+            export_fn(export_cfg)
+
+        # Join auf Übersetzung - nicht zugeordnet
+        df_ignored = df_bookings.merge(self.df_translate, how="left", on="Konto_Nr_Händler")
+        df_ignored = df_ignored[df_ignored["Konto_Nr_SKR51"].isna()]
+        if not df_ignored.empty:
+            df_ignored = df_ignored.pivot_table(
+                index=["Konto_Nr_Händler"],
+                columns=["period"],
+                values="amount",
+                aggfunc="sum",
+                margins=True,
+                margins_name="CumulatedYear",
+            )
+            df_ignored.to_csv(self.account_ignored, decimal=",", sep=";", encoding="latin-1")
+        return export_filename
+
+    @property
+    def df_translate(self) -> pd.DataFrame:
+        if self._df_translate is None:
+            self.makes, self.sites, self._df_translate, self.df_translate2 = load_translation(
+                self.account_translation, self.debug_file, self.export_invalid_filename
+            )
+        return self._df_translate
+
+    @property
+    def export_info_dir(self) -> str:
+        return f"{self.base_dir}/Export/{self.current_year}/info/"
+
+    @property
+    def logs_dir(self) -> str:
+        return f"{self.base_dir}/Logs/"
+
+    @property
+    def export_invalid_filename(self) -> str:
+        return f"{self.base_dir}/Export/ungueltig.csv"
+
+    def export_filename_for_period(self, year: str, month: str) -> str:
+        return f"{self.base_dir}/Export/{year}/export_{year}-{month}.xml"
+
+
+def gchr_local() -> None:
+    base_dir = os.getcwd() + "/../GCHR2_Testdaten/Kunden"
+    for path in Path(base_dir).glob("*"):
+        if not path.is_dir():
+            continue
+        print(path.name)
+        gchr = GCHR(str(path))
+        gchr.export_all_periods()