|
@@ -64,8 +64,8 @@ class GchrConfig:
|
|
|
|
|
|
class GCHR:
|
|
|
booking_date: datetime
|
|
|
- df_bookings: pd.DataFrame = None
|
|
|
- df_translate: pd.DataFrame = None
|
|
|
+ _df_bookings: pd.DataFrame = None
|
|
|
+ _df_translate: pd.DataFrame = None
|
|
|
df_translate2: pd.DataFrame = None
|
|
|
makes: dict[str, str] = None
|
|
|
sites: dict[str, str] = None
|
|
@@ -85,8 +85,8 @@ class GCHR:
|
|
|
def set_bookkeep_period(self, year: str, month: str) -> None:
|
|
|
self.current_year = year
|
|
|
self.current_month = month
|
|
|
- period = f"{year}-{month}"
|
|
|
- prot_file = f"{self.export_info_dir}/protokoll_{period}.log"
|
|
|
+ self.period = f"{year}-{month}"
|
|
|
+ prot_file = f"{self.export_info_dir}/protokoll_{self.period}.log"
|
|
|
logging.basicConfig(
|
|
|
filename=prot_file,
|
|
|
filemode="w",
|
|
@@ -94,13 +94,28 @@ class GCHR:
|
|
|
level=logging.DEBUG,
|
|
|
force=True,
|
|
|
)
|
|
|
- self.debug_file = f"{self.export_info_dir}/debug_{period}.csv"
|
|
|
- self.account_ignored = f"{self.export_info_dir}/ignoriert_{period}.csv"
|
|
|
- # self.account_invalid = f"{self.export_info_dir}/ungueltig_{period}.csv"
|
|
|
|
|
|
- self.last_year = str(int(self.current_year) - 1)
|
|
|
- self.last_year2 = str(int(self.current_year) - 2)
|
|
|
- self.next_year = str(int(self.current_year) + 1)
|
|
|
+ @property
|
|
|
+ def debug_file(self) -> str:
|
|
|
+ return f"{self.export_info_dir}/debug_{self.period}.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"
|
|
|
+
|
|
|
+ @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)
|
|
|
|
|
|
@staticmethod
|
|
|
def header(export_cfg: GchrExportConfig) -> dict[str, str]:
|
|
@@ -131,7 +146,8 @@ class GCHR:
|
|
|
rename_to = ["OpeningBalance"] + ["Period" + str(i).zfill(2) for i in range(1, 13)]
|
|
|
return dict(zip(period, rename_to))
|
|
|
|
|
|
- def extract_acct_info(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
|
+ @staticmethod
|
|
|
+ def extract_acct_info(df: pd.DataFrame) -> pd.DataFrame:
|
|
|
acct_info = [
|
|
|
"Marke",
|
|
|
"Standort",
|
|
@@ -166,8 +182,6 @@ class GCHR:
|
|
|
|
|
|
def export_period(self, year: str, month: str) -> str:
|
|
|
self.set_bookkeep_period(year, month)
|
|
|
- # Übersetzungstabelle laden
|
|
|
- self.get_translation()
|
|
|
|
|
|
# Kontensalden laden
|
|
|
df_bookings = self.filter_bookings()
|
|
@@ -183,7 +197,7 @@ class GCHR:
|
|
|
|
|
|
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")
|
|
|
+ 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(
|
|
@@ -243,7 +257,7 @@ class GCHR:
|
|
|
df.to_dict(orient="records"),
|
|
|
)
|
|
|
export_cfg.header = self.header(export_cfg)
|
|
|
- self.export_skr51_xml(export_cfg)
|
|
|
+ GCHR.export_skr51_xml(export_cfg)
|
|
|
|
|
|
# Join auf Übersetzung - nicht zugeordnet
|
|
|
df_ignored = df_bookings.merge(self.df_translate, how="left", on="Konto_Nr_Händler")
|
|
@@ -260,8 +274,9 @@ class GCHR:
|
|
|
df_ignored.to_csv(self.account_ignored, decimal=",", sep=";", encoding="latin-1")
|
|
|
return self.export_filename
|
|
|
|
|
|
- def get_translation(self) -> pd.DataFrame:
|
|
|
- if self.df_translate is None:
|
|
|
+ @property
|
|
|
+ def df_translate(self) -> pd.DataFrame:
|
|
|
+ if self._df_translate is None:
|
|
|
df_translate_import = pd.read_csv(
|
|
|
self.account_translation,
|
|
|
decimal=",",
|
|
@@ -270,49 +285,58 @@ class GCHR:
|
|
|
converters={i: str for i in range(0, 200)},
|
|
|
).reset_index()
|
|
|
|
|
|
- df_makes = df_translate_import[["Marke", "Marke_HBV"]].copy().drop_duplicates()
|
|
|
- df_makes = df_makes[df_makes["Marke_HBV"] != "0000"]
|
|
|
- self.makes = dict([(e["Marke"], e["Marke_HBV"]) for e in df_makes.to_dict(orient="records")])
|
|
|
- self.makes["99"] = "0000"
|
|
|
- df_sites = df_translate_import[["Marke", "Standort", "Standort_HBV"]].copy().drop_duplicates()
|
|
|
- df_sites["Standort_HBV"] = np.where(
|
|
|
- df_sites["Standort_HBV"].str.len() != 6, "0000", df_sites["Standort_HBV"]
|
|
|
- )
|
|
|
- self.sites = dict(
|
|
|
- [(e["Marke"] + "-" + e["Standort"], e["Standort_HBV"]) for e in df_sites.to_dict(orient="records")]
|
|
|
- )
|
|
|
+ self.makes = GCHR.get_makes_from_translation(df_translate_import)
|
|
|
+ self.sites = GCHR.get_sites_from_translation(df_translate_import)
|
|
|
|
|
|
- df_prepared = self.prepare_translation(df_translate_import)
|
|
|
- self.df_translate = self.special_translation(df_prepared)
|
|
|
+ df_prepared = GCHR.prepare_translation(df_translate_import)
|
|
|
+ self._df_translate = self.special_translation(df_prepared)
|
|
|
self.df_translate2 = (
|
|
|
- self.df_translate.drop(columns=["Konto_Nr_Händler"])
|
|
|
- .copy()
|
|
|
+ self._df_translate.copy()
|
|
|
+ .drop(columns=["Konto_Nr_Händler"])
|
|
|
.drop_duplicates()
|
|
|
.set_index("Konto_Nr_SKR51")
|
|
|
)
|
|
|
- return self.df_translate
|
|
|
+ return self._df_translate
|
|
|
|
|
|
- def prepare_translation(self, df_translate_import: pd.DataFrame):
|
|
|
- df_translate = df_translate_import[
|
|
|
+ @staticmethod
|
|
|
+ def get_makes_from_translation(df_translate_import: pd.DataFrame) -> dict[str, str]:
|
|
|
+ df_makes = df_translate_import[["Marke", "Marke_HBV"]].copy().drop_duplicates()
|
|
|
+ df_makes = df_makes[df_makes["Marke_HBV"] != "0000"]
|
|
|
+ makes = dict([(e["Marke"], e["Marke_HBV"]) for e in df_makes.to_dict(orient="records")])
|
|
|
+ makes["99"] = "0000"
|
|
|
+ return makes
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_sites_from_translation(df_translate_import: pd.DataFrame) -> dict[str, str]:
|
|
|
+ df_sites = df_translate_import[["Marke", "Standort", "Standort_HBV"]].copy().drop_duplicates()
|
|
|
+ df_sites["Standort_HBV"] = np.where(df_sites["Standort_HBV"].str.len() != 6, "0000", df_sites["Standort_HBV"])
|
|
|
+ sites = dict(
|
|
|
+ [(e["Marke"] + "-" + e["Standort"], e["Standort_HBV"]) for e in df_sites.to_dict(orient="records")]
|
|
|
+ )
|
|
|
+ return sites
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def prepare_translation(df_translate_import: pd.DataFrame) -> pd.DataFrame:
|
|
|
+ df = df_translate_import[
|
|
|
[
|
|
|
"Konto_Nr_Händler",
|
|
|
"Konto_Nr_SKR51",
|
|
|
]
|
|
|
].drop_duplicates()
|
|
|
- logging.info(df_translate.shape)
|
|
|
+ logging.info(df.shape)
|
|
|
|
|
|
row = {
|
|
|
"Konto_Nr_Händler": "01-01-0861-00-00-00",
|
|
|
"Konto_Nr_SKR51": "01-01-0861-00-00-00",
|
|
|
}
|
|
|
- df_translate = pd.concat([df_translate, pd.DataFrame.from_records([row])])
|
|
|
- df_translate.set_index("Konto_Nr_Händler")
|
|
|
- return df_translate
|
|
|
+ df = pd.concat([df, pd.DataFrame.from_records([row])])
|
|
|
+ df.set_index("Konto_Nr_Händler")
|
|
|
+ return df
|
|
|
|
|
|
def special_translation(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
|
df["Konto_Nr_Händler"] = df["Konto_Nr_Händler"].str.upper()
|
|
|
df["Konto_Nr_SKR51"] = df["Konto_Nr_SKR51"].str.upper()
|
|
|
- df = self.extract_acct_info(df)
|
|
|
+ df = GCHR.extract_acct_info(df)
|
|
|
df["Konto_Nr"] = df["Konto_Nr"].str.upper()
|
|
|
logging.info(df.shape)
|
|
|
logging.info(df.columns)
|
|
@@ -426,11 +450,11 @@ class GCHR:
|
|
|
return df[df["IsNumeric"] == True][TRANSLATE]
|
|
|
|
|
|
def load_bookings_from_file(self) -> None:
|
|
|
- df2 = []
|
|
|
- timestamps = []
|
|
|
+ df_list: list[pd.DataFrame] = []
|
|
|
+ timestamps: list[float] = []
|
|
|
|
|
|
for csv_file in self.account_bookings:
|
|
|
- df2.append(
|
|
|
+ df_list.append(
|
|
|
pd.read_csv(
|
|
|
csv_file,
|
|
|
decimal=",",
|
|
@@ -441,12 +465,17 @@ class GCHR:
|
|
|
)
|
|
|
timestamps.append(Path(csv_file).stat().st_mtime)
|
|
|
self.booking_date = datetime.fromtimestamp(max(timestamps))
|
|
|
- self.df_bookings = pd.concat(df2)
|
|
|
- self.df_bookings["amount"] = (self.df_bookings["Debit Amount"] + self.df_bookings["Credit Amount"]).round(2)
|
|
|
+ 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:
|
|
|
- if self.df_bookings is None:
|
|
|
- self.load_bookings_from_file()
|
|
|
# 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
|
|
@@ -561,14 +590,16 @@ class GCHR:
|
|
|
with open(export_cfg.export_file, "w", encoding="utf-8") as fwh:
|
|
|
fwh.write(minidom.parseString(ET.tostring(root)).toprettyxml(indent=" "))
|
|
|
|
|
|
- def convert_to_row(self, node: list[ET.Element]) -> list[str]:
|
|
|
+ @staticmethod
|
|
|
+ def convert_to_row(node: list[ET.Element]) -> list[str]:
|
|
|
return [child.text for child in node]
|
|
|
|
|
|
- def convert_xml_to_csv(self, xmlfile: str, csvfile: str) -> bool:
|
|
|
+ @staticmethod
|
|
|
+ def convert_xml_to_csv(xmlfile: str, csvfile: str) -> bool:
|
|
|
with open(xmlfile) as frh:
|
|
|
record_list = ET.parse(frh).getroot().find("RecordList")
|
|
|
header = [child.tag for child in record_list.find("Record")]
|
|
|
- bookings = [self.convert_to_row(node) for node in record_list.findall("Record")]
|
|
|
+ bookings = [GCHR.convert_to_row(node) for node in record_list.findall("Record")]
|
|
|
with open(csvfile, "w") as fwh:
|
|
|
cwh = csv.writer(fwh, delimiter=";")
|
|
|
cwh.writerow(header)
|
|
@@ -580,7 +611,7 @@ class GCHR:
|
|
|
self.sites = {"01-01": "1844"}
|
|
|
with open(csvfile, "r", encoding="latin-1") as frh:
|
|
|
csv_reader = csv.DictReader(frh, delimiter=";")
|
|
|
- self.export_skr51_xml(csv_reader, self.bookkeep_filter(), 1, list(self.sites.values())[0], xmlfile)
|
|
|
+ GCHR.export_skr51_xml(csv_reader, self.bookkeep_filter(), 1, list(self.sites.values())[0], xmlfile)
|
|
|
|
|
|
|
|
|
def gchr_local() -> None:
|
|
@@ -595,12 +626,3 @@ 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 __name__ == "__main__":
|
|
|
- gchr_local()
|
|
|
- # import cProfile
|
|
|
- # cProfile.run(
|
|
|
- # "gchr_local()",
|
|
|
- # "gchr_local.prof",
|
|
|
- # )
|