Python-Projekte/gchr/gchr.py

import json
import logging
import os
from datetime import datetime
from pathlib import Path

# import numpy as np
import pandas as pd

from gchr.gchr_bookings import GchrBookings
from gchr.gchr_export import get_export_fn
from gchr.gchr_model import ACCOUNT_INFO, GchrConfig, GchrExportConfig
from gchr.gchr_translate import load_translation


class GCHR:
    cfg: GchrConfig
    bookings: GchrBookings
    _df_translate: pd.DataFrame = None
    df_translate2: pd.DataFrame = None
    makes: dict[str, str] = None
    sites: dict[str, str] = None
    timestamp: str

    def __init__(self, base_dir: str) -> None:
        self.base_dir = base_dir
        cfg_file = f"{self.base_dir}\\config\\gchr2.json"
        if Path(cfg_file).exists():
            with open(cfg_file, "r") as frh:
                self.cfg = GchrConfig(**json.load(frh))
        else:
            os.makedirs(f"{self.base_dir}\\config", exist_ok=True)
            self.cfg = GchrConfig(
                first_month_of_financial_year="01",
                data_dir=f"{self.base_dir}\\data",
                gcstruct_dir=f"{self.base_dir}\\..\\GCStruct_Aufbereitung",
                export_dir=f"{self.base_dir}\\Export",
                export_format="SKR51",
            )
            with open(cfg_file, "w") as fwh:
                json.dump(self.cfg.__dict__, fwh, indent=2)

        os.makedirs(self.cfg.data_dir, exist_ok=True)
        os.makedirs(f"{self.cfg.export_dir}\\temp", exist_ok=True)
        os.makedirs(f"{self.base_dir}\\logs", exist_ok=True)

        self.account_translation = f"{self.cfg.data_dir}\\Kontenrahmen_uebersetzt.csv"

        self.bookings = GchrBookings(self.base_dir, self.cfg.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.bookings.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(self.cfg.export_format)
        # Kontensalden laden
        df_bookings = self.bookings.filter_bookings(year, month)
        all_periods = list(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 and all_periods[0][-2:] == "00")
            or self.bookings.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")

        if True:
            # Nur Konten mit Inhalt exportieren
            entries_with_content = pd.DataFrame(df["CumulatedYear"] != 0)
            period_columns = [p for p in df.columns if p.startswith("Period")]
            for p in period_columns:
                entries_with_content[p] = df[p] != 0
            df = df[entries_with_content.any(axis="columns")]

        # 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" and "_" not in self.sites[s]
        ]

        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=main_site,
                current_year=year,
                current_month=month,
                makes_used=makes_used,
                sites_used=sites_used,
                first_month=self.cfg.first_month_of_financial_year,
                period_no=period_no,
                bookkeep_filter=self.bookings.bookkeep_filter,
                extraction_date=self.bookings.booking_date,
                export_file=filename,
                bookkeep_records=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 = df_ignored.merge(self.df_translate, how="inner", on="Konto_Nr_Händler")
            # df_ignored["Kontoart"] =
            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.cfg.export_dir}\\{self.bookings.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.cfg.export_dir}\\ungueltig.csv"

    def export_filename_for_period(self, year: str, month: str) -> str:
        return f"{self.cfg.export_dir}\\{year}\\export_{year}-{month}.xml"