Refactoring: Funktion "export_period" aufgeteilt

gcstruct/gchr.py

@@ -66,6 +66,7 @@ class GCHR:
             "BeginFiscalYear": self.first_month_of_financial_year,
         }
 
+    @property
     def bookkeep_filter(self):
         period = [self.current_year + str(i).zfill(2) for i in range(1, 13)]
         if self.first_month_of_financial_year != "01":
@@ -107,120 +108,24 @@ class GCHR:
     def export_period(self, year, month):
         self.set_bookkeep_period(year, month)
         # Übersetzungstabelle laden
-        df_translate = pd.read_csv(
+        df_translate_import = pd.read_csv(
             self.account_translation,
             decimal=",",
             sep=";",
             encoding="latin-1",
             converters={i: str for i in range(0, 200)},
         )
-        logging.info(df_translate.shape)
-        df_translate["duplicated"] = df_translate.duplicated()
-        logging.info(df_translate[df_translate["duplicated"]])
-        df_translate = df_translate[
-            [
-                "Konto_Nr_Händler",
-                "Konto_Nr_SKR51",
-                "Marke",
-                "Marke_HBV",
-                "Standort",
-                "Standort_HBV",
-            ]
-        ]
-
-        row = (
-            df_translate[["Marke", "Marke_HBV", "Standort", "Standort_HBV"]]
-            .drop_duplicates()
-            .sort_values(by=["Marke", "Standort"])
-            .iloc[:1]
-            .to_dict(orient="records")[0]
-        )
-        row["Konto_Nr_Händler"] = "01-01-0861-00-00-00"
-        row["Konto_Nr_SKR51"] = "01-01-0861-00-00-00"
-
-        df_translate = pd.concat([df_translate, pd.DataFrame.from_records([row])])
-        # print(df_translate.tail())
-        # df_translate.drop(columns=['duplicated'], inplace=True)
-        df_translate.drop_duplicates(inplace=True)
-        df_translate.set_index("Konto_Nr_Händler")
+        df_translate = self.prepare_translation(df_translate_import)
 
         # Kontensalden laden
-        df2 = []
-        for csv_file in self.account_bookings:
-            df2.append(
-                pd.read_csv(
-                    csv_file,
-                    decimal=",",
-                    sep=";",
-                    encoding="latin-1",
-                    converters={0: str, 1: str},
-                )
-            )
-        df_bookings = pd.concat(df2)
-        # Kontensalden auf gegebenen Monat filtern
-        filter_from = self.current_year + self.first_month_of_financial_year
-        filter_prev = self.last_year + self.first_month_of_financial_year
-
-        if self.first_month_of_financial_year > self.current_month:
-            filter_from = self.last_year + self.first_month_of_financial_year
-            filter_prev = self.last_year2 + self.first_month_of_financial_year
-        filter_to = self.current_year + self.current_month
-        filter_opening = self.current_year + "00"
-        filter_prev_opening = self.last_year + "00"
-        prev_year_closed = True
-
-        df_opening_balance = df_bookings[(df_bookings["Bookkeep Period"] == filter_opening)]
-        if df_opening_balance.shape[0] == 0:
-            df_opening_balance = df_bookings[
-                (df_bookings["Bookkeep Period"] == filter_prev_opening)
-                | ((df_bookings["Bookkeep Period"] >= filter_prev) & (df_bookings["Bookkeep Period"] < filter_from))
-            ].copy()
-            df_opening_balance["Bookkeep Period"] = filter_opening
-            prev_year_closed = False
-        # df_opening_balance = df_opening_balance.merge(df_translate, how='inner', on='Konto_Nr_Händler')
-        df_opening_balance = df_opening_balance[(df_opening_balance["Konto_Nr_Händler"].str.contains(r"-[013]\d\d+-"))]
-        df_opening_balance["amount"] = (df_opening_balance["Debit Amount"] + df_opening_balance["Credit Amount"]).round(
-            2
-        )
-        # df_opening_balance.drop(columns=['Debit Amount', 'Credit Amount', 'Debit Quantity', 'Credit Quantity'], inplace=True)
-        # df_opening_balance = df_opening_balance.groupby(['Marke', 'Standort']).sum()
-        opening_balance = df_opening_balance["amount"].aggregate("sum").round(2)
-        logging.info("Gewinn/Verlustvortrag")
-        logging.info(opening_balance)
-
-        if not prev_year_closed:
-            row = {
-                "Konto_Nr_Händler": "01-01-0861-00-00-00",
-                "Bookkeep Period": filter_opening,
-                "Debit Amount": opening_balance * -1,
-                "Credit Amount": 0,
-                "Debit Quantity": 0,
-                "Credit Quantity": 0,
-                "amount": opening_balance * -1,
-            }
-            df_opening_balance = pd.concat([df_opening_balance, pd.DataFrame.from_records([row])])
-
-        df_bookings = df_bookings[
-            (df_bookings["Bookkeep Period"] >= filter_from) & (df_bookings["Bookkeep Period"] <= filter_to)
-        ]
-        df_bookings["amount"] = (df_bookings["Debit Amount"] + df_bookings["Credit Amount"]).round(2)
-        df_stats = df_bookings.copy()
-        # df_stats = df_stats[df_stats['Konto_Nr_Händler'].str.match(r'-[24578]\d\d\d-')]
-        df_stats["Konto_Nr_Händler"] = df_stats["Konto_Nr_Händler"].str.replace(r"-(\d\d\d+)-", r"-\1_STK-", regex=True)
-        df_stats["amount"] = (df_bookings["Debit Quantity"] + df_bookings["Credit Quantity"]).round(2)
-
-        df_bookings = pd.concat([df_opening_balance, df_bookings, df_stats])
-        df_bookings = df_bookings[df_bookings["amount"] != 0.00]
+        df_bookings = self.load_bookings_from_file()
 
         if df_bookings.shape[0] == 0:
             logging.error("ABBRUCH!!! Keine Daten vorhanden!")
             return False
 
-        bk_filter = self.bookkeep_filter()
-        period_no = list(bk_filter.keys()).index(filter_to) + 1
-
-        # Spalten konvertieren
-        df_bookings["period"] = df_bookings["Bookkeep Period"].apply(lambda x: bk_filter[x])
+        filter_to = self.current_year + self.current_month
+        period_no = list(self.bookkeep_filter.keys()).index(filter_to) + 1
 
         logging.info("df_bookings: " + str(df_bookings.shape))
         # Join auf Übersetzung
@@ -246,7 +151,7 @@ class GCHR:
         # df_combined.to_csv(account_invalid, decimal=',', sep=';', encoding='latin-1', index=False)
         # Gruppieren
         # df_grouped = df_combined.groupby(['Konto_Nr_SKR51', 'period']).sum()
-        df = df_combined.pivot_table(
+        df_pivot = df_combined.pivot_table(
             index=["Konto_Nr_SKR51"],
             columns=["period"],
             values="amount",
@@ -255,13 +160,86 @@ class GCHR:
             margins_name="CumulatedYear",
         )
 
-        logging.info("df_pivot: " + str(df.shape))
+        logging.info("df_pivot: " + str(df_pivot.shape))
+
+        df = self.special_translation(df_pivot, makes)
+
+        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)
+        makes = makes.rename(columns=col_dict).to_dict(orient="records")
+        sites = sites.rename(columns=col_dict).to_dict(orient="records")
+
+        df_invalid = df[df["IsNumeric"] == False]
+        df_invalid.to_csv(self.account_invalid, decimal=",", sep=";", encoding="latin-1", index=False)
+        export_csv = self.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['OpeningBalance'] = 0.0
+        logging.info(df.shape)
+        self.export_xml(df.to_dict(orient="records"), self.bookkeep_filter, period_no, makes, sites)
+
+        # Join auf Übersetzung - nicht zugeordnet
+        df_ignored = df_bookings.merge(df_translate, how="left", on="Konto_Nr_Händler")
+        df_ignored = df_ignored[
+            df_ignored["Konto_Nr_SKR51"].isna()
+        ]  # [['Konto_Nr_Händler', 'Bookkeep Period', 'amount', 'quantity']]
+        if not df_ignored.empty:
+            df_ignored = df_ignored.pivot_table(
+                index=["Konto_Nr_Händler"],
+                columns=["period"],
+                values="amount",
+                aggfunc=np.sum,
+                margins=True,
+                margins_name="CumulatedYear",
+            )
+            df_ignored.to_csv(self.account_ignored, decimal=",", sep=";", encoding="latin-1")
+        return self.export_filename
+
+    def prepare_translation(self, df_translate: pd.DataFrame):
+        logging.info(df_translate.shape)
+        df_translate["duplicated"] = df_translate.duplicated()
+        logging.info(df_translate[df_translate["duplicated"]])
+        df_translate = df_translate[
+            [
+                "Konto_Nr_Händler",
+                "Konto_Nr_SKR51",
+                "Marke",
+                "Marke_HBV",
+                "Standort",
+                "Standort_HBV",
+            ]
+        ]
+
+        row = (
+            df_translate[["Marke", "Marke_HBV", "Standort", "Standort_HBV"]]
+            .drop_duplicates()
+            .sort_values(by=["Marke", "Standort"])
+            .iloc[:1]
+            .to_dict(orient="records")[0]
+        )
+        row["Konto_Nr_Händler"] = "01-01-0861-00-00-00"
+        row["Konto_Nr_SKR51"] = "01-01-0861-00-00-00"
+
+        df_translate = pd.concat([df_translate, pd.DataFrame.from_records([row])])
+        # print(df_translate.tail())
+        # df_translate.drop(columns=['duplicated'], inplace=True)
+        df_translate.drop_duplicates(inplace=True)
+        df_translate.set_index("Konto_Nr_Händler")
+        return df_translate
+
+    def special_translation(self, df: pd.DataFrame, makes: pd.DataFrame):
         df = self.extract_acct_info(df)
         # df = df_translate.reset_index(drop=True).drop(columns=['Kostenträger_Ebene']).drop_duplicates()
         logging.info(df.shape)
         logging.info(df.columns)
         logging.info(df.head())
         # df = df.merge(df_translate, how='inner', on='Konto_Nr_SKR51')
+
         logging.info("df: " + str(df.shape))
         df["Bilanz"] = df["Konto_Nr"].str.match(r"^[013]")
         df["Kontoart"] = np.where(df["Bilanz"], "1", "2")
@@ -352,52 +330,82 @@ class GCHR:
         )
         df["Kostenträger"] = np.where(df["Service_40_70"] == True, "70", df["Kostenträger"])
 
-        from_label = [
-            "Marke",
-            "Standort",
-            "Konto_Nr",
-            "Kostenstelle",
-            "Absatzkanal",
-            "Kostenträger",
-        ]
-        to_label = ["Make", "Site", "Account", "Origin", "SalesChannel", "CostCarrier"]
-        df = df.rename(columns=dict(zip(from_label, to_label)))
-        makes = makes.rename(columns=dict(zip(from_label, to_label))).to_dict(orient="records")
-        sites = sites.rename(columns=dict(zip(from_label, to_label))).to_dict(orient="records")
-
-        df["CostAccountingString"] = df["Make"] + df["Site"] + df["Origin"] + df["SalesChannel"] + df["CostCarrier"]
-        df["IsNumeric"] = (
-            (df["CostAccountingString"].str.isdigit()) & (df["Account"].str.isdigit()) & (df["Account"].str.len() == 4)
+        df["KRM"] = df["Marke"] + df["Standort"] + df["Kostenstelle"] + df["Absatzkanal"] + df["Kostenträger"]
+        df["IsNumeric"] = (df["KRM"].str.isdigit()) & (df["Konto_Nr"].str.isdigit()) & (df["Konto_Nr"].str.len() == 4)
+        return df
+
+    def load_bookings_from_file(self):
+        df2 = []
+        for csv_file in self.account_bookings:
+            df2.append(
+                pd.read_csv(
+                    csv_file,
+                    decimal=",",
+                    sep=";",
+                    encoding="latin-1",
+                    converters={0: str, 1: str},
+                )
+            )
+        df_bookings = pd.concat(df2)
+        # Kontensalden auf gegebenen Monat filtern
+        filter_from = self.current_year + self.first_month_of_financial_year
+        filter_prev = self.last_year + self.first_month_of_financial_year
+
+        if self.first_month_of_financial_year > self.current_month:
+            filter_from = self.last_year + self.first_month_of_financial_year
+            filter_prev = self.last_year2 + self.first_month_of_financial_year
+        filter_to = self.current_year + self.current_month
+        filter_opening = self.current_year + "00"
+        filter_prev_opening = self.last_year + "00"
+        prev_year_closed = True
+
+        df_opening_balance = df_bookings[(df_bookings["Bookkeep Period"] == filter_opening)]
+        if df_opening_balance.shape[0] == 0:
+            df_opening_balance = df_bookings[
+                (df_bookings["Bookkeep Period"] == filter_prev_opening)
+                | ((df_bookings["Bookkeep Period"] >= filter_prev) & (df_bookings["Bookkeep Period"] < filter_from))
+            ].copy()
+            df_opening_balance["Bookkeep Period"] = filter_opening
+            prev_year_closed = False
+        # df_opening_balance = df_opening_balance.merge(df_translate, how='inner', on='Konto_Nr_Händler')
+        df_opening_balance = df_opening_balance[(df_opening_balance["Konto_Nr_Händler"].str.contains(r"-[013]\d\d+-"))]
+        df_opening_balance["amount"] = (df_opening_balance["Debit Amount"] + df_opening_balance["Credit Amount"]).round(
+            2
         )
+        # df_opening_balance.drop(columns=['Debit Amount', 'Credit Amount', 'Debit Quantity', 'Credit Quantity'], inplace=True)
+        # df_opening_balance = df_opening_balance.groupby(['Marke', 'Standort']).sum()
+        opening_balance = df_opening_balance["amount"].aggregate("sum").round(2)
+        logging.info("Gewinn/Verlustvortrag")
+        logging.info(opening_balance)
 
-        df_invalid = df[df["IsNumeric"] == False]
-        df_invalid.to_csv(self.account_invalid, decimal=",", sep=";", encoding="latin-1", index=False)
-        export_csv = self.export_filename[:-4] + ".csv"
+        if not prev_year_closed:
+            row = {
+                "Konto_Nr_Händler": "01-01-0861-00-00-00",
+                "Bookkeep Period": filter_opening,
+                "Debit Amount": opening_balance * -1,
+                "Credit Amount": 0,
+                "Debit Quantity": 0,
+                "Credit Quantity": 0,
+                "amount": opening_balance * -1,
+            }
+            df_opening_balance = pd.concat([df_opening_balance, pd.DataFrame.from_records([row])])
 
-        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['OpeningBalance'] = 0.0
-        logging.info(df.shape)
-        self.export_xml(df.to_dict(orient="records"), bk_filter, period_no, makes, sites)
+        df_bookings = df_bookings[
+            (df_bookings["Bookkeep Period"] >= filter_from) & (df_bookings["Bookkeep Period"] <= filter_to)
+        ]
+        df_bookings["amount"] = (df_bookings["Debit Amount"] + df_bookings["Credit Amount"]).round(2)
 
-        # Join auf Übersetzung - nicht zugeordnet
-        df_ignored = df_bookings.merge(df_translate, how="left", on="Konto_Nr_Händler")
-        df_ignored = df_ignored[
-            df_ignored["Konto_Nr_SKR51"].isna()
-        ]  # [['Konto_Nr_Händler', 'Bookkeep Period', 'amount', 'quantity']]
-        if not df_ignored.empty:
-            df_ignored = df_ignored.pivot_table(
-                index=["Konto_Nr_Händler"],
-                columns=["period"],
-                values="amount",
-                aggfunc=np.sum,
-                margins=True,
-                margins_name="CumulatedYear",
-            )
-            df_ignored.to_csv(self.account_ignored, decimal=",", sep=";", encoding="latin-1")
-        return self.export_filename
+        # Buchungen kopieren und als Statistikkonten anhängen
+        df_stats = df_bookings.copy()
+        # df_stats = df_stats[df_stats['Konto_Nr_Händler'].str.match(r'-[24578]\d\d\d-')]
+        df_stats["Konto_Nr_Händler"] = df_stats["Konto_Nr_Händler"].str.replace(r"-(\d\d\d+)-", r"-\1_STK-", regex=True)
+        df_stats["amount"] = (df_bookings["Debit Quantity"] + df_bookings["Credit Quantity"]).round(2)
+
+        df_bookings = pd.concat([df_opening_balance, df_bookings, df_stats])
+
+        # Spalten konvertieren
+        df_bookings["period"] = df_bookings["Bookkeep Period"].apply(lambda x: self.bookkeep_filter[x])
+        return df_bookings[df_bookings["amount"] != 0.00]
 
     @property
     def export_filename(self):