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_export import ( ACCOUNT_INFO, GchrExportConfig, export_skr51_xml, header, ) 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 _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 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.account_bookings = list(Path(self.base_dir).joinpath("data").glob("GuV_Bilanz_Salden*.csv")) self.first_month_of_financial_year = "10" 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" @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) @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) 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: self.set_bookkeep_period(year, month) # Kontensalden laden df_bookings = self.filter_bookings() 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 = 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 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_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.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 = self.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.bookkeep_filter, filename, df.to_dict(orient="records"), ) export_cfg.header = header(export_cfg) 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") 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 self.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 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/" @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 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()