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()