GCHR Bookings ausgelagert

gchr2.py

@@ -1,6 +1,7 @@
 import os
 
-from gcstruct import gchr
+from gcstruct.gchr import GCHR
 
 if __name__ == "__main__":
-    gchr.gchr_export(os.getcwd())
+    gchr = GCHR(os.getcwd())
+    gchr.export_all_periods()

gcstruct/gchr.py

@@ -7,11 +7,12 @@ from typing import Callable
 
 import pandas as pd
 
+from gcstruct.gchr_bookings import GchrBookings
 from gcstruct.gchr_export import (
     ACCOUNT_INFO,
     GchrExportConfig,
-    export_skr51_xml,
-    header,
+    GchrExportFormat,
+    get_export_fn,
 )
 from gcstruct.gchr_translate import load_translation
 
@@ -27,7 +28,7 @@ class GchrConfig:
 
 class GCHR:
     booking_date: datetime
-    _df_bookings: pd.DataFrame = None
+    bookings: GchrBookings
     _df_translate: pd.DataFrame = None
     df_translate2: pd.DataFrame = None
     makes: dict[str, str] = None
@@ -43,27 +44,15 @@ class GCHR:
         os.makedirs(self.base_dir + "/logs", exist_ok=True)
 
         self.account_translation = f"{self.base_dir}/data/Kontenrahmen_uebersetzt.csv"
-        self.account_bookings = list(Path(self.base_dir).joinpath("data").glob("GuV_Bilanz_Salden*.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)
 
-    def set_bookkeep_period(self, year: str, month: str) -> None:
-        self.current_year = year
-        self.current_month = month
-        self.period = f"{year}-{month}"
-        prot_file = f"{self.export_info_dir}/protokoll_{self.period}.log"
-        logging.basicConfig(
-            filename=prot_file,
-            filemode="w",
-            encoding="utf-8",
-            level=logging.DEBUG,
-            force=True,
-        )
-
     @property
     def debug_file(self) -> str:
         return f"{self.logs_dir}/debug_{self.timestamp}.csv"
@@ -74,32 +63,6 @@ class GCHR:
 
     # self.account_invalid = f"{self.export_info_dir}/ungueltig_{self.period}.csv"
 
-    @property
-    def last_year(self) -> str:
-        return str(int(self.current_year) - 1)
-
-    @property
-    def last_year2(self) -> str:
-        return str(int(self.current_year) - 2)
-
-    @property
-    def next_year(self) -> str:
-        return str(int(self.current_year) + 1)
-
-    @property
-    def bookkeep_filter(self) -> dict[str, str]:
-        period = [self.current_year + str(i).zfill(2) for i in range(1, 13)]
-        if self.first_month_of_financial_year != "01":
-            if self.first_month_of_financial_year > self.current_month:
-                period = [self.last_year + str(i).zfill(2) for i in range(1, 13)] + period
-            else:
-                period = period + [self.next_year + str(i).zfill(2) for i in range(1, 13)]
-            fm = int(self.first_month_of_financial_year)
-            period = period[fm - 1 : fm + 12]
-        period = [self.current_year + "00"] + period
-        rename_to = ["OpeningBalance"] + ["Period" + str(i).zfill(2) for i in range(1, 13)]
-        return dict(zip(period, rename_to))
-
     def export_all_periods(self, overwrite=False, today=None) -> None:
         periods = GCHR.get_all_periods(today)
 
@@ -121,10 +84,9 @@ class GCHR:
         return periods
 
     def export_period(self, year: str, month: str) -> str:
-        self.set_bookkeep_period(year, month)
-
+        export_fn = get_export_fn(GchrExportFormat.SKR51)
         # Kontensalden laden
-        df_bookings = self.filter_bookings()
+        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)
 
@@ -132,8 +94,8 @@ class GCHR:
             logging.error("ABBRUCH!!! Keine Daten vorhanden!")
             return False
 
-        filter_to = self.current_year + self.current_month
-        period_no = list(self.bookkeep_filter.keys()).index(filter_to) + 1
+        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
@@ -170,7 +132,8 @@ class GCHR:
         col_dict = dict(zip(from_label, to_label))
         df = df.rename(columns=col_dict)
 
-        export_csv = self.export_filename[:-4] + ".csv"
+        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")
 
@@ -181,7 +144,7 @@ class GCHR:
         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 = self.export_filename
+            filename = export_filename
             if i > 0:
                 filename = f"{filename[:-4]}_{main_site}.xml"
             export_cfg = GchrExportConfig(
@@ -192,12 +155,12 @@ class GCHR:
                 sites_used,
                 self.first_month_of_financial_year,
                 period_no,
-                self.bookkeep_filter,
+                self.bookings.bookkeep_filter,
                 filename,
                 df.to_dict(orient="records"),
             )
-            export_cfg.header = header(export_cfg)
-            export_skr51_xml(export_cfg)
+
+            export_fn(export_cfg)
 
         # Join auf Übersetzung - nicht zugeordnet
         df_ignored = df_bookings.merge(self.df_translate, how="left", on="Konto_Nr_Händler")
@@ -212,7 +175,7 @@ class GCHR:
                 margins_name="CumulatedYear",
             )
             df_ignored.to_csv(self.account_ignored, decimal=",", sep=";", encoding="latin-1")
-        return self.export_filename
+        return export_filename
 
     @property
     def df_translate(self) -> pd.DataFrame:
@@ -222,94 +185,6 @@ class GCHR:
             )
         return self._df_translate
 
-    def load_bookings_from_file(self) -> None:
-        df_list: list[pd.DataFrame] = []
-        timestamps: list[float] = []
-
-        for csv_file in self.account_bookings:
-            df_list.append(
-                pd.read_csv(
-                    csv_file,
-                    decimal=",",
-                    sep=";",
-                    encoding="latin-1",
-                    converters={0: str, 1: str},
-                )
-            )
-            timestamps.append(Path(csv_file).stat().st_mtime)
-        self.booking_date = datetime.fromtimestamp(max(timestamps))
-        df = pd.concat(df_list)
-        df["amount"] = (df["Debit Amount"] + df["Credit Amount"]).round(2)
-        return df
-
-    @property
-    def df_bookings(self) -> pd.DataFrame:
-        if self._df_bookings is None:
-            self._df_bookings = self.load_bookings_from_file()
-        return self._df_bookings
-
-    def filter_bookings(self) -> pd.DataFrame:
-        # 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 = self.df_bookings[(self.df_bookings["Bookkeep Period"] == filter_opening)]
-        if df_opening_balance.shape[0] == 0:
-            df_opening_balance = self.df_bookings[
-                (self.df_bookings["Bookkeep Period"] == filter_prev_opening)
-                | (
-                    (self.df_bookings["Bookkeep Period"] >= filter_prev)
-                    & (self.df_bookings["Bookkeep Period"] < filter_from)
-                )
-            ].copy()
-            df_opening_balance["Bookkeep Period"] = filter_opening
-            prev_year_closed = False
-
-        df_opening_balance = df_opening_balance[(df_opening_balance["Konto_Nr_Händler"].str.contains(r"-[013]\d\d+-"))]
-        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_filtered = self.df_bookings[
-            (self.df_bookings["Bookkeep Period"] >= filter_from) & (self.df_bookings["Bookkeep Period"] <= filter_to)
-        ]
-
-        # Buchungen kopieren und als Statistikkonten anhängen
-        df_stats = df_filtered.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_filtered["Debit Quantity"] + df_filtered["Credit Quantity"]).round(2)
-
-        df_combined = pd.concat([df_opening_balance, df_filtered, df_stats])
-
-        # Spalten konvertieren
-        df_combined["period"] = df_combined["Bookkeep Period"].apply(lambda x: self.bookkeep_filter[x])
-        return df_combined[df_combined["amount"] != 0.00]
-
-    @property
-    def export_filename(self) -> str:
-        return self.export_filename_for_period(self.current_year, self.current_month)
-
     @property
     def export_info_dir(self) -> str:
         return f"{self.base_dir}/Export/{self.current_year}/info/"
@@ -329,12 +204,8 @@ class GCHR:
 def gchr_local() -> None:
     base_dir = os.getcwd() + "/../GCHR2_Testdaten/Kunden"
     for path in Path(base_dir).glob("*"):
-        if path.is_dir():
-            print(path.name)
-            gchr_export(str(path))
-
-
-def gchr_export(base_dir: str) -> None:
-    gchr = GCHR(base_dir)
-    # gchr.export_all_periods(overwrite=True, today="2022-08-01")
-    gchr.export_all_periods()
+        if not path.is_dir():
+            continue
+        print(path.name)
+        gchr = GCHR(str(path))
+        gchr.export_all_periods()

gcstruct/gchr_bookings.py

@@ -0,0 +1,142 @@
+import logging
+from datetime import datetime
+from pathlib import Path
+
+import pandas as pd
+
+
+class GchrBookings:
+    base_dir: str
+    account_bookings: list[str]
+    _df_bookings: pd.DataFrame
+    booking_date: datetime
+
+    def __init__(self, base_dir: list[str], first_month: str = None):
+        self.base_dir = base_dir
+        self.account_bookings = list(Path(self.base_dir).joinpath("data").glob("GuV_Bilanz_Salden*.csv"))
+        self.first_month_of_financial_year = first_month or "01"
+
+        self._df_bookings = self.load_bookings_from_file()
+
+    def load_bookings_from_file(self) -> None:
+        df_list: list[pd.DataFrame] = []
+        timestamps: list[float] = []
+
+        for csv_file in self.account_bookings:
+            df_list.append(
+                pd.read_csv(
+                    csv_file,
+                    decimal=",",
+                    sep=";",
+                    encoding="latin-1",
+                    converters={0: str, 1: str},
+                )
+            )
+            timestamps.append(Path(csv_file).stat().st_mtime)
+        self.booking_date = datetime.fromtimestamp(max(timestamps))
+        df = pd.concat(df_list)
+        df["amount"] = (df["Debit Amount"] + df["Credit Amount"]).round(2)
+        return df
+
+    def set_bookkeep_period(self, year: str, month: str) -> None:
+        self.current_year = year
+        self.current_month = month
+        self.period = f"{year}-{month}"
+        prot_file = f"{self.export_info_dir}/protokoll_{self.period}.log"
+        logging.basicConfig(
+            filename=prot_file,
+            filemode="w",
+            encoding="utf-8",
+            level=logging.DEBUG,
+            force=True,
+        )
+
+    @property
+    def last_year(self) -> str:
+        return str(int(self.current_year) - 1)
+
+    @property
+    def last_year2(self) -> str:
+        return str(int(self.current_year) - 2)
+
+    @property
+    def next_year(self) -> str:
+        return str(int(self.current_year) + 1)
+
+    def filter_bookings(self, year: str, month: str) -> pd.DataFrame:
+        self.set_bookkeep_period(year, month)
+
+        # 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 = self._df_bookings[(self._df_bookings["Bookkeep Period"] == filter_opening)]
+        if df_opening_balance.shape[0] == 0:
+            df_opening_balance = self._df_bookings[
+                (self._df_bookings["Bookkeep Period"] == filter_prev_opening)
+                | (
+                    (self._df_bookings["Bookkeep Period"] >= filter_prev)
+                    & (self._df_bookings["Bookkeep Period"] < filter_from)
+                )
+            ].copy()
+            df_opening_balance["Bookkeep Period"] = filter_opening
+            prev_year_closed = False
+
+        df_opening_balance = df_opening_balance[(df_opening_balance["Konto_Nr_Händler"].str.contains(r"-[013]\d\d+-"))]
+        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_filtered = self._df_bookings[
+            (self._df_bookings["Bookkeep Period"] >= filter_from) & (self._df_bookings["Bookkeep Period"] <= filter_to)
+        ]
+
+        # Buchungen kopieren und als Statistikkonten anhängen
+        df_stats = df_filtered.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_filtered["Debit Quantity"] + df_filtered["Credit Quantity"]).round(2)
+
+        df_combined = pd.concat([df_opening_balance, df_filtered, df_stats])
+
+        # Spalten konvertieren
+        df_combined["period"] = df_combined["Bookkeep Period"].apply(lambda x: self.bookkeep_filter[x])
+        return df_combined[df_combined["amount"] != 0.00]
+
+    @property
+    def bookkeep_filter(self) -> dict[str, str]:
+        period = [self.current_year + str(i).zfill(2) for i in range(1, 13)]
+        if self.first_month_of_financial_year != "01":
+            if self.first_month_of_financial_year > self.current_month:
+                period = [self.last_year + str(i).zfill(2) for i in range(1, 13)] + period
+            else:
+                period = period + [self.next_year + str(i).zfill(2) for i in range(1, 13)]
+            fm = int(self.first_month_of_financial_year)
+            period = period[fm - 1 : fm + 12]
+        period = [self.current_year + "00"] + period
+        rename_to = ["OpeningBalance"] + ["Period" + str(i).zfill(2) for i in range(1, 13)]
+        return dict(zip(period, rename_to))
+
+    @property
+    def export_info_dir(self) -> str:
+        return f"{self.base_dir}/Export/{self.current_year}/info/"

gcstruct/gchr_export.py

@@ -1,6 +1,8 @@
 import xml.etree.ElementTree as ET
+from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime
+from enum import StrEnum, auto
 from xml.dom import minidom
 
 ACCOUNT_INFO = [
@@ -14,6 +16,12 @@ ACCOUNT_INFO = [
 ]
 
 
+class GchrExportFormat(StrEnum):
+    SKR51 = auto()
+    Volkswagen = auto()
+    Opel = auto()
+
+
 @dataclass
 class GchrExportConfig:
     main_site: str
@@ -27,7 +35,6 @@ class GchrExportConfig:
     extraction_date: datetime
     export_file: str
     bookkeep_records = dict[str, list[str]]
-    header: dict[str, str] | None = None
 
 
 def export_skr51_xml(export_cfg: GchrExportConfig):
@@ -39,7 +46,7 @@ def export_skr51_xml(export_cfg: GchrExportConfig):
     )
     root = ET.Element("HbvData")
     h = ET.SubElement(root, "Header")
-    for k, v in export_cfg.header.items():
+    for k, v in export_skr51_header(export_cfg).items():
         ET.SubElement(h, k).text = str(v)
 
     make_list = ET.SubElement(root, "MakeList")
@@ -72,7 +79,7 @@ def export_skr51_xml(export_cfg: GchrExportConfig):
         fwh.write(minidom.parseString(ET.tostring(root)).toprettyxml(indent="  "))
 
 
-def header(export_cfg: GchrExportConfig) -> dict[str, str]:
+def export_skr51_header(export_cfg: GchrExportConfig) -> dict[str, str]:
     return {
         "Country": "DE",
         "MainBmCode": export_cfg.main_site,
@@ -85,3 +92,19 @@ def header(export_cfg: GchrExportConfig) -> dict[str, str]:
         "ExtractionTime": export_cfg.extraction_date.strftime("%H:%M:%S"),
         "BeginFiscalYear": export_cfg.first_month,
     }
+
+
+GchrExportFn = Callable[[GchrExportConfig], None]
+
+
+EXPORT_FN: dict[GchrExportFormat, GchrExportFn] = {
+    GchrExportFormat.SKR51: export_skr51_xml,
+}
+
+
+def export_dummy(export_cfg: GchrExportConfig) -> None:
+    pass
+
+
+def get_export_fn(export_format: GchrExportFormat) -> GchrExportFn:
+    return EXPORT_FN.get(export_format, export_dummy)

gcstruct/tests/test_gchr.py

@@ -3,6 +3,7 @@ import unittest
 import pandas as pd
 
 from gcstruct.gchr import GCHR
+from gcstruct.gchr_bookings import GchrBookings
 from gcstruct.gchr_translate import TRANSLATE
 
 
@@ -12,14 +13,16 @@ class TestGchr(unittest.TestCase):
 
     def test_single_booking_files(self):
         gchr = GCHR(self.base_dir_1)
-        self.assertEqual(len(gchr.account_bookings), 1)
-        self.assertEqual(gchr.account_bookings[0].name, "GuV_Bilanz_Salden.csv")
+        self.assertIsInstance(gchr.bookings, GchrBookings)
+        self.assertEqual(len(gchr.bookings.account_bookings), 1)
+        self.assertEqual(gchr.bookings.account_bookings[0].name, "GuV_Bilanz_Salden.csv")
 
     def test_multiple_booking_files(self):
         gchr = GCHR(self.base_dir_2)
-        self.assertEqual(len(gchr.account_bookings), 2)
-        self.assertEqual(gchr.account_bookings[0].name, "GuV_Bilanz_Salden.csv")
-        self.assertEqual(gchr.account_bookings[1].name, "GuV_Bilanz_Salden_deop03.csv")
+        self.assertIsInstance(gchr.bookings, GchrBookings)
+        self.assertEqual(len(gchr.bookings.account_bookings), 2)
+        self.assertEqual(gchr.bookings.account_bookings[0].name, "GuV_Bilanz_Salden.csv")
+        self.assertEqual(gchr.bookings.account_bookings[1].name, "GuV_Bilanz_Salden_deop03.csv")
 
     def test_translation_existing(self):
         gchr = GCHR(self.base_dir_1)