|
@@ -5,6 +5,7 @@ import xml.etree.ElementTree as ET
|
|
|
from dataclasses import dataclass
|
|
|
from datetime import datetime
|
|
|
from pathlib import Path
|
|
|
+from typing import Callable
|
|
|
from xml.dom import minidom
|
|
|
|
|
|
import numpy as np
|
|
@@ -36,18 +37,29 @@ TRANSLATE = [
|
|
|
]
|
|
|
|
|
|
|
|
|
+@dataclass
|
|
|
+class GchrExportConfig:
|
|
|
+ main_site: str
|
|
|
+ current_year: str
|
|
|
+ current_month: str
|
|
|
+ makes_used: dict[str, str]
|
|
|
+ sites_used: dict[str, str]
|
|
|
+ first_month: str
|
|
|
+ period_no: str
|
|
|
+ bookkeep_filter: dict[str, str]
|
|
|
+ extraction_date: datetime
|
|
|
+ export_file: str
|
|
|
+ bookkeep_records = dict[str, list[str]]
|
|
|
+ header: dict[str, str] | None = None
|
|
|
+
|
|
|
+
|
|
|
@dataclass
|
|
|
class GchrConfig:
|
|
|
first_month_of_financial_year: str
|
|
|
data_path: str
|
|
|
gcstruct_path: str
|
|
|
export_path: str
|
|
|
-
|
|
|
-
|
|
|
-@dataclass
|
|
|
-class GchrExportConfig:
|
|
|
- current_year: str
|
|
|
- current_month: str
|
|
|
+ export_fn = Callable[[GchrExportConfig], None]
|
|
|
|
|
|
|
|
|
class GCHR:
|
|
@@ -90,18 +102,19 @@ class GCHR:
|
|
|
self.last_year2 = str(int(self.current_year) - 2)
|
|
|
self.next_year = str(int(self.current_year) + 1)
|
|
|
|
|
|
- def header(self, makes_used: list[str], sites_used: list[str], main_site: str) -> dict[str, str]:
|
|
|
+ @staticmethod
|
|
|
+ def header(export_cfg: GchrExportConfig) -> dict[str, str]:
|
|
|
return {
|
|
|
"Country": "DE",
|
|
|
- "MainBmCode": main_site,
|
|
|
- "Month": self.current_month,
|
|
|
- "Year": self.current_year,
|
|
|
+ "MainBmCode": export_cfg.main_site,
|
|
|
+ "Month": export_cfg.current_month,
|
|
|
+ "Year": export_cfg.current_year,
|
|
|
"Currency": "EUR",
|
|
|
- "NumberOfMakes": len(makes_used),
|
|
|
- "NumberOfSites": len(sites_used),
|
|
|
- "ExtractionDate": self.booking_date.strftime("%d.%m.%Y"),
|
|
|
- "ExtractionTime": self.booking_date.strftime("%H:%M:%S"),
|
|
|
- "BeginFiscalYear": self.first_month_of_financial_year,
|
|
|
+ "NumberOfMakes": len(export_cfg.makes_used),
|
|
|
+ "NumberOfSites": len(export_cfg.sites_used),
|
|
|
+ "ExtractionDate": export_cfg.extraction_date.strftime("%d.%m.%Y"),
|
|
|
+ "ExtractionTime": export_cfg.extraction_date.strftime("%H:%M:%S"),
|
|
|
+ "BeginFiscalYear": export_cfg.first_month,
|
|
|
}
|
|
|
|
|
|
@property
|
|
@@ -187,8 +200,16 @@ class GCHR:
|
|
|
|
|
|
df = df_pivot.merge(self.df_translate2, how="inner", on="Konto_Nr_SKR51")
|
|
|
|
|
|
- makes_used = sorted(list(set(df["Marke"].to_list())))
|
|
|
- sites_used = sorted(list(set((df["Marke"] + "-" + df["Standort"]).to_list())))
|
|
|
+ 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"]
|
|
@@ -209,15 +230,20 @@ class GCHR:
|
|
|
filename = self.export_filename
|
|
|
if i > 0:
|
|
|
filename = f"{filename[:-4]}_{main_site}.xml"
|
|
|
- self.export_skr51_xml(
|
|
|
- df.to_dict(orient="records"),
|
|
|
- self.bookkeep_filter,
|
|
|
- period_no,
|
|
|
+ export_cfg = GchrExportConfig(
|
|
|
+ main_site,
|
|
|
+ year,
|
|
|
+ month,
|
|
|
makes_used,
|
|
|
sites_used,
|
|
|
- main_site,
|
|
|
+ self.first_month_of_financial_year,
|
|
|
+ period_no,
|
|
|
+ self.bookkeep_filter,
|
|
|
filename,
|
|
|
+ df.to_dict(orient="records"),
|
|
|
)
|
|
|
+ export_cfg.header = self.header(export_cfg)
|
|
|
+ self.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")
|
|
@@ -493,42 +519,35 @@ class GCHR:
|
|
|
def export_filename_for_period(self, year: str, month: str) -> str:
|
|
|
return f"{self.base_dir}/Export/{year}/export_{year}-{month}.xml"
|
|
|
|
|
|
- def export_skr51_xml(
|
|
|
- self,
|
|
|
- records: dict[str, list[str]],
|
|
|
- bk_filter: dict[str, str],
|
|
|
- period_no: str,
|
|
|
- makes_used: list[str],
|
|
|
- sites_used: list[str],
|
|
|
- main_site: str,
|
|
|
- filename: str,
|
|
|
- ):
|
|
|
- record_elements = ACCOUNT_INFO + ["Decimals"] + list(bk_filter.values())[:period_no] + ["CumulatedYear"]
|
|
|
+ @staticmethod
|
|
|
+ def export_skr51_xml(export_cfg: GchrExportConfig):
|
|
|
+ record_elements = (
|
|
|
+ ACCOUNT_INFO
|
|
|
+ + ["Decimals"]
|
|
|
+ + list(export_cfg.bookkeep_filter.values())[: export_cfg.period_no]
|
|
|
+ + ["CumulatedYear"]
|
|
|
+ )
|
|
|
root = ET.Element("HbvData")
|
|
|
h = ET.SubElement(root, "Header")
|
|
|
- for k, v in self.header(makes_used, sites_used, main_site).items():
|
|
|
+ for k, v in export_cfg.header.items():
|
|
|
ET.SubElement(h, k).text = str(v)
|
|
|
|
|
|
make_list = ET.SubElement(root, "MakeList")
|
|
|
- for make in makes_used:
|
|
|
- if make not in self.makes:
|
|
|
- continue
|
|
|
+ for make, make_code in export_cfg.makes_used.items():
|
|
|
e = ET.SubElement(make_list, "MakeListEntry")
|
|
|
ET.SubElement(e, "Make").text = make
|
|
|
- ET.SubElement(e, "MakeCode").text = self.makes[make]
|
|
|
+ ET.SubElement(e, "MakeCode").text = make_code
|
|
|
|
|
|
bm_code_list = ET.SubElement(root, "BmCodeList")
|
|
|
- for s in sites_used:
|
|
|
+ for s, bmcode in export_cfg.sites_used.items():
|
|
|
make, site = s.split("-")
|
|
|
- if s not in self.sites:
|
|
|
- continue
|
|
|
e = ET.SubElement(bm_code_list, "BmCodeEntry")
|
|
|
ET.SubElement(e, "Make").text = make
|
|
|
ET.SubElement(e, "Site").text = site
|
|
|
- ET.SubElement(e, "BmCode").text = self.sites[s]
|
|
|
+ ET.SubElement(e, "BmCode").text = bmcode
|
|
|
|
|
|
record_list = ET.SubElement(root, "RecordList")
|
|
|
- for row in records:
|
|
|
+ for row in export_cfg.bookkeep_records:
|
|
|
record = ET.SubElement(record_list, "Record")
|
|
|
for e in record_elements:
|
|
|
child = ET.SubElement(record, e)
|
|
@@ -539,7 +558,7 @@ class GCHR:
|
|
|
field = "{:.0f}".format(field * 100)
|
|
|
child.text = str(field)
|
|
|
|
|
|
- with open(filename, "w", encoding="utf-8") as fwh:
|
|
|
+ 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]:
|