Python-Projekte/gcstruct/gchr.py

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